[fix](ci) Add review findings to one draft before publishing (#69447)

### What problem does this PR solve?

Related PR: #68499

The review runner's bulk REST create-review request rejects five valid
inline findings on #68499 with HTTP 422 (`Position could not be
resolved`; line/side also fails). In a controlled experiment at the same
head/base, all five original findings were accepted when added
sequentially to **one PENDING review** through GraphQL. The experimental
draft was deleted afterward. The exact GitHub internal diff limit is not
established.

For reviews with inline findings, this change creates one draft, adds
one thread per request, verifies the complete comment set, then
publishes once. It converts the existing path/position/body input to
line/side using the authoritative local PR diff. Summary-only reviews
keep their current submission path.

The invocation marker, exclusive submission intent, frozen head/base
checks, priority/convergence gate, and exact final readback remain
required. The helper rechecks competing bot reviews before publication.
Add failures trigger best-effort cleanup only of the exactly owned
pending draft; uncertain writes are never retried. `review-draft.json`
is included in failure artifacts, and GraphQL errors in HTTP 200
responses retain HTTP diagnostics. Ambiguous publication, unavailable
cleanup, and hard termination boundaries are documented.

Live formal publication and the GitHub Actions identity still need
rollout validation. The existing experiment validates draft
construction, not end-to-end CI delivery.

.github/scripts/review-github-diagnostics.md

@@ -15,6 +15,7 @@ under REVIEW_GITHUB_DIAGNOSTICS_DIR. Records include the command exit status or
 timeout, the last HTTP method/path/status, GitHub request ID, rate-limit and
 retry headers, gh version, and bounded response/stdout/stderr excerpts. A
 successful gh exit followed by a Python JSON decode error is also recorded.
+GraphQL errors returned with HTTP 200 are recorded as failures as well.
 A transport failure without a response has status: null; it is not inferred
 to be an HTTP error.
 
@@ -46,3 +47,38 @@ RUN_GH_HTTP_TESTS=1 python3 -m unittest discover -s .github/scripts -p 'test_*.p
 The opt-in HTTP test uses real gh only against a loopback server, with fake
 tokens and a temporary config; it never sends a request to GitHub. Linux-only
 process-supervision tests require a Linux host.
+
+## Staged final review submission
+
+When a review has inline findings, submit_review.py creates one PENDING review,
+adds each finding with a separate addPullRequestReviewThread mutation, verifies
+the full draft comment set, then submits the review once. This avoids asking
+the bulk REST create-review endpoint to resolve all targeted file diffs in one
+request. The input still uses path/position/body; line and side are derived
+from the authoritative local pr.diff before any write. A summary-only review
+keeps the direct REST submission path.
+
+The invocation marker, frozen head/base guards, exclusive review-submission.json
+intent and complete final readback remain required. A pending review, partial
+comment set or local journal never proves delivery. No write is automatically
+retried. A lost final-submit response may be recovered by exact readback of the
+already submitted review, without another submission event.
+
+failed-code-review-* now also includes review-draft.json: the current stage,
+owned review ID, confirmed thread IDs and current finding index/location. This
+is diagnostic progress, not a resume instruction. Before publication, an error
+triggers best-effort deletion only of the uniquely identified pending draft
+with this invocation's exact body, bot author and head. Foreign or modified
+drafts and submitted reviews are never deleted.
+
+An ambiguous publish attempt is not deleted or repeated. If it remains PENDING,
+or cleanup cannot finish because of an API outage, exhausted budget or process
+termination, inspect the journal and GitHub state before manually clearing the
+owned draft and starting a new run. Never remove the submission intent to
+force a retry. The pipeline must fail rather than report partial delivery.
+
+The motivating #68499 experiment accepted all five original findings in one
+draft via sequential GraphQL mutations, while bulk REST creation rejected the
+same set with 422. That experiment used a personal account and deleted the
+draft; formal publication and the Actions identity still require live rollout
+validation. The exact GitHub internal diff limit is not established.

.github/scripts/submit_review.py

@@ -19,6 +19,7 @@
 """Submit and verify a final review bound to this runner invocation."""
 
 import argparse
+import codecs
 import hashlib
 import json
 import os
@@ -30,12 +31,13 @@ from collections import Counter
 from datetime import datetime, timezone
 from pathlib import Path
 
-from review_github_api import decode_json, redact, run_gh
+from review_github_api import decode_json, record_failure, redact, run_gh
 
 
 SUBMISSION_FILE = "review-submission.json"
 REJECTION_FILE = "review-submission-rejected.json"
 API_ERRORS_FILE = "review-api-errors.jsonl"
+DRAFT_FILE = "review-draft.json"
 RUN_FILE = "review-run.json"
 RESULT_FILE = "review-result.json"
 PRIORITY = re.compile(r"^\[P([0-3])\]\s+\S")
@@ -120,13 +122,15 @@ class GitHubAPIError(ValueError):
         super().__init__(message)
 
 
-def api(path, remaining, *, payload=None, paginated=False):
+def api(path, remaining, *, payload=None, paginated=False, method=None):
     command = ["gh", "api", path]
     if paginated:
         command += ["--paginate", "--slurp"]
+    method = method or ("POST" if payload is not None else "GET")
+    if method != "GET":
+        command += ["--method", method]
     if payload is not None:
-        command += ["--method", "POST", "--input", "-"]
-    method = "POST" if payload is not None else "GET"
+        command += ["--input", "-"]
     try:
         result = run_gh(
             command, input=json.dumps(payload) if payload is not None else None,
@@ -138,6 +142,12 @@ def api(path, remaining, *, payload=None, paginated=False):
         value = decode_json(result, command)
     except json.JSONDecodeError as error:
         raise GitHubAPIError(path, method, error, result.stdout, result.stderr) from None
+    # GraphQL can return HTTP 200 with an errors array. Never count that as a
+    # successful mutation, even if gh changes its exit-code handling.
+    if path == "graphql" and value.get("errors"):
+        error = ValueError("GraphQL errors")
+        record_failure(command, error, result.stdout, result.stderr, diagnostic=result.github_response)
+        raise GitHubAPIError(path, method, error, result.stdout, result.stderr)
     return [item for page in value for item in page] if paginated else value
 
 
@@ -207,6 +217,164 @@ def review_payload(submission, run, root, remaining):
     }, counts, marker
 
 
+THREAD_MUTATION = """mutation($input: AddPullRequestReviewThreadInput!) {
+  addPullRequestReviewThread(input: $input) {
+    thread { id path line diffSide }
+  }
+}"""
+
+
+def diff_path(header):
+    value = header[4:]
+    if value.startswith('"'):
+        # Git quotes UTF-8 bytes with C escapes (including octal), not JSON.
+        value = codecs.escape_decode(value[1:-1].encode())[0].decode()
+    else:
+        value = value.split("\t", 1)[0]
+    return None if value == "/dev/null" else value[2:]
+
+
+def inline_locations(context, comments):
+    """Map the existing position schema using the authoritative local PR diff."""
+    wanted = {(c["path"], c["position"]) for c in comments}
+    locations = {}
+    old_path = new_path = position = None
+    old_line = new_line = 0
+    with (context / "pr.diff").open() as handle:
+        for raw in handle:
+            line = raw.rstrip("\n")
+            if line.startswith("diff --git "):
+                old_path = new_path = position = None
+            elif position is None and line.startswith("--- "):
+                old_path = diff_path(line)
+            elif position is None and line.startswith("+++ "):
+                new_path = diff_path(line)
+            elif line.startswith("@@ "):
+                match = re.match(r"@@ -(\d+)(?:,\d+)? \+(\d+)(?:,\d+)? @@", line)
+                if match is None:
+                    raise ValueError("Invalid hunk in authoritative PR diff")
+                old_line, new_line = map(int, match.groups())
+                # The first hunk header is position zero; subsequent headers
+                # consume a position, as do no-newline markers (not anchors).
+                position = 0 if position is None else position + 1
+            elif position is not None:
+                position += 1
+                key = (new_path or old_path, position)
+                if line.startswith((" ", "+", "-")):
+                    if key in wanted:
+                        locations[key] = {"line": old_line if line[0] == "-" else new_line,
+                                          "side": "LEFT" if line[0] == "-" else "RIGHT"}
+                    old_line += line[0] != "+"
+                    new_line += line[0] != "-"
+    if set(locations) != wanted:
+        missing = sorted(wanted - set(locations))
+        raise ValueError(f"Finding positions are not commentable PR diff lines: {missing}")
+    return [locations[(c["path"], c["position"])] for c in comments]
+
+
+def another_bot_review(reviews, run):
+    return any(r.get("user", {}).get("login") == "github-actions[bot]"
+               and r.get("commit_id") == run["head_sha"]
+               and (r.get("submitted_at") or "") >= run["started_at"] for r in reviews)
+
+
+def owned_draft(review, payload):
+    return (review.get("state") == "PENDING"
+            and review.get("user", {}).get("login") == "github-actions[bot]"
+            and review.get("commit_id") == payload["commit_id"]
+            and review.get("body") == payload["body"])
+
+
+def verify_inline_comments(root, review, payload, run, remaining, *, pending=False):
+    # Replies may arrive during wrap-up. They are not declared findings.
+    comments = [c for c in api(f"{root}/reviews/{review['id']}/comments", remaining, paginated=True)
+                if c.get("in_reply_to_id") is None]
+    expected = Counter((c["path"], c["position"], c["body"]) for c in payload["comments"])
+    actual = Counter((c["path"], c.get("original_position") if pending else c.get("position"), c["body"])
+                     for c in comments)
+    if actual != expected or any(
+        c.get("user", {}).get("login") != "github-actions[bot]"
+        or c.get("commit_id") != run["head_sha"] for c in comments
+    ):
+        raise ValueError("Final review inline comments were not completely verified")
+
+
+def submit_draft(context, run, root, payload, locations, remaining):
+    """Create one draft, add one thread per request, verify, then publish once.
+
+    The exclusive SUBMISSION_FILE remains the no-repost guard. A partial draft
+    is not completion and is never resumed by another helper invocation.
+    """
+    progress = {"stage": "creating", "review_id": None, "thread_ids": []}
+
+    def record(stage, **fields):
+        progress.update(stage=stage, **fields)
+        write_json(context / DRAFT_FILE, progress)
+
+    record("creating")
+    publish_attempted = False
+    try:
+        draft = api(f"{root}/reviews", remaining, payload={
+            "commit_id": payload["commit_id"], "body": payload["body"], "comments": [],
+        })
+        if not owned_draft(draft, payload):
+            raise ValueError("Created review is not this invocation's pending draft")
+        record("created", review_id=draft["id"])
+        for index, (comment, location) in enumerate(zip(payload["comments"], locations)):
+            check_target(run, remaining)
+            record("adding", comment_index=index, path=comment["path"], **location)
+            result = api("graphql", remaining, payload={
+                "query": THREAD_MUTATION, "variables": {"input": {
+                    "pullRequestReviewId": draft["node_id"], "path": comment["path"],
+                    "body": comment["body"], **location,
+                }},
+            })
+            thread = result["data"]["addPullRequestReviewThread"]["thread"]
+            if (thread["path"] != comment["path"] or thread["line"] != location["line"]
+                    or thread["diffSide"] != location["side"]):
+                raise ValueError("Added review thread location did not match")
+            progress["thread_ids"].append(thread["id"])
+            record("added")
+        draft = api(f"{root}/reviews/{draft['id']}", remaining)
+        if not owned_draft(draft, payload):
+            raise ValueError("Pending review author, head, state or summary did not match")
+        verify_inline_comments(root, draft, payload, run, remaining, pending=True)
+        if another_bot_review(api(f"{root}/reviews", remaining, paginated=True), run):
+            raise ValueError("Another bot review was submitted during this run; refusing duplicate submission")
+        check_target(run, remaining)
+        record("submitting")
+        publish_attempted = True
+        api(f"{root}/reviews/{draft['id']}/events", remaining,
+            payload={"event": payload["event"], "body": payload["body"]})
+        record("submitted")
+    except Exception:
+        # Before publishing, deleting only our exactly identified pending draft
+        # prevents an add failure from leaving a draft that blocks future runs.
+        # After a publish attempt, its outcome can be ambiguous: leave it alone
+        # and let the existing exact final-review readback recover delivery.
+        if not publish_attempted:
+            try:
+                drafts = [r for r in api(f"{root}/reviews", remaining, paginated=True)
+                          if owned_draft(r, payload)]
+                if len(drafts) > 1:
+                    raise ValueError("Cannot uniquely identify this invocation's pending draft")
+                if drafts:
+                    current = api(f"{root}/reviews/{drafts[0]['id']}", remaining)
+                    if not owned_draft(current, payload):
+                        raise ValueError("Pending draft changed before cleanup")
+                    api(f"{root}/reviews/{current['id']}", remaining, method="DELETE")
+                    record("deleted", review_id=current["id"])
+            except Exception as cleanup_error:
+                # Keep the original create/add failure as the primary error.
+                progress.update(stage="cleanup_failed", cleanup_error=diagnostic_text(str(cleanup_error)))
+                try:
+                    write_json(context / DRAFT_FILE, progress)
+                except OSError:
+                    pass  # Report below; do not replace the original API failure.
+                print(f"Pending review cleanup needs follow-up: {progress}", file=sys.stderr)
+        raise
+
+
 def verify_completion(context, run, remaining, *, submit=False, submission=None):
     try:
         return _verify_completion(context, run, remaining, submit=submit, submission=submission)
@@ -240,9 +408,7 @@ def _verify_completion(context, run, remaining, *, submit=False, submission=None
     reviews = api(f"{root}/reviews", remaining, paginated=True)
     matching = [r for r in reviews if marker in (r.get("body") or "")]
     if not matching and submit:
-        if any(r.get("user", {}).get("login") == "github-actions[bot]"
-               and r.get("commit_id") == run["head_sha"]
-               and (r.get("submitted_at") or "") >= run["started_at"] for r in reviews):
+        if another_bot_review(reviews, run):
             reason = "Another bot review was submitted during this run; refusing duplicate submission"
             # This is a rejected candidate, not a POST intent or proof of delivery.
             # Keep it separate so completion still requires exact GitHub readback.
@@ -252,13 +418,20 @@ def _verify_completion(context, run, remaining, *, submit=False, submission=None
             raise ValueError(with_api_failure(
                 context, "Final submission was attempted but cannot be verified; refusing to repost"
             ))
+        # Resolve every anchor before creating any GitHub state. The existing
+        # submission schema stays compatible with review agents and old runs.
+        locations = inline_locations(context, payload["comments"]) if payload["comments"] else []
         # Persist the complete intent before POST: if the response is lost, the
         # runner can read back this exact review without repeating the write.
         # Exclusive creation also prevents two accidental simultaneous helper
         # invocations from both posting after their initial read found no review.
         with path.open("x") as handle:
             json.dump(submission, handle)
-        api(f"{root}/reviews", remaining, payload=payload)
+        if payload["comments"]:
+            submit_draft(context, run, root, payload, locations, remaining)
+        else:
+            # A summary-only review does not ask GitHub to parse any diff.
+            api(f"{root}/reviews", remaining, payload=payload)
         matching = [r for r in api(f"{root}/reviews", remaining, paginated=True)
                     if marker in (r.get("body") or "")]
     if len(matching) != 1:
@@ -273,17 +446,7 @@ def _verify_completion(context, run, remaining, *, submit=False, submission=None
             or review.get("state") != expected_state
             or review.get("body") != payload["body"]):
         raise ValueError("Final review author, head, timestamp, state or summary did not match")
-    # Replies may arrive while the model wraps up. They are not part of the
-    # declared inline set and must not invalidate an otherwise complete review.
-    comments = [c for c in api(f"{root}/reviews/{review['id']}/comments", remaining, paginated=True)
-                if c.get("in_reply_to_id") is None]
-    expected = Counter((c["path"], c["position"], c["body"]) for c in payload["comments"])
-    actual = Counter((c["path"], c.get("position"), c["body"]) for c in comments)
-    if actual != expected or any(
-        c.get("user", {}).get("login") != "github-actions[bot]"
-        or c.get("commit_id") != run["head_sha"] for c in comments
-    ):
-        raise ValueError("Final review inline comments were not completely verified")
+    verify_inline_comments(root, review, payload, run, remaining)
     # Recheck after the API reads so a moving PR cannot pass an obsolete review.
     check_target(run, remaining)
     return {

.github/scripts/test_submit_review.py

@@ -52,6 +52,12 @@ class FinalReviewTest(unittest.TestCase):
         self.comments = []
         self.existing = []
         self.posts = []
+        self.thread_inputs = []
+        self.final_events = []
+        self.deletes = []
+        self.diff = ("diff --git a/file.py b/file.py\nnew file mode 100644\n"
+                     "--- /dev/null\n+++ b/file.py\n@@ -0,0 +1,20 @@\n" + "+line\n" * 20)
+        (self.context / "pr.diff").write_text(self.diff)
         self.real_api = submitter.api
         self.api = self.enterContext(mock.patch.object(submitter, "api", side_effect=self.fake_api))
         self.remaining = lambda: 60
@@ -62,21 +68,48 @@ class FinalReviewTest(unittest.TestCase):
         ], "existing_blocking_comment_ids": [], "converged": converged, "rounds": rounds,
             "incomplete_reason": "" if converged else "Round 3 found another candidate; coverage is not settled."}
 
-    def fake_api(self, path, remaining, *, payload=None, paginated=False):
+    def fake_api(self, path, remaining, *, payload=None, paginated=False, method=None):
         if path == self.root:
             return copy.deepcopy(self.pr)
         if path == self.root + "/reviews":
             if payload is None:
                 return copy.deepcopy(self.reviews)
             self.posts.append(copy.deepcopy(payload))
+            event = payload.get("event")
             self.reviews.append({
-                "id": 99, "user": {"login": "github-actions[bot]"},
-                "commit_id": payload["commit_id"], "submitted_at": "9999-01-01T00:00:00Z",
+                "id": 99, "node_id": "review-node-99", "user": {"login": "github-actions[bot]"},
+                "commit_id": payload["commit_id"], "submitted_at": "9999-01-01T00:00:00Z" if event else None,
                 "body": payload["body"], "html_url": "https://github.com/apache/doris/pull/123#pullrequestreview-99",
-                "state": "CHANGES_REQUESTED" if payload["event"] == "REQUEST_CHANGES" else "COMMENTED",
+                "state": ("CHANGES_REQUESTED" if event == "REQUEST_CHANGES" else "COMMENTED") if event else "PENDING",
             })
             self.comments = [dict(c, user={"login": "github-actions[bot]"},
                                   commit_id=payload["commit_id"]) for c in payload["comments"]]
+            if event:
+                self.final_events.append(event)
+            return copy.deepcopy(self.reviews[-1])
+        if path == "graphql":
+            self.assertEqual(submitter.THREAD_MUTATION, payload["query"])
+            inp = copy.deepcopy(payload["variables"]["input"])
+            self.assertEqual("review-node-99", inp["pullRequestReviewId"])
+            self.thread_inputs.append(inp)
+            self.comments.append({"path": inp["path"], "body": inp["body"],
+                                  "position": inp["line"], "original_position": inp["line"],
+                                  "user": {"login": "github-actions[bot]"}, "commit_id": self.run["head_sha"]})
+            return {"data": {"addPullRequestReviewThread": {"thread": {
+                "id": f"thread-{len(self.thread_inputs)}", "path": inp["path"],
+                "line": inp["line"], "diffSide": inp["side"],
+            }}}}
+        if path == self.root + "/reviews/99":
+            if method == "DELETE":
+                self.deletes.append(99)
+                self.reviews = [r for r in self.reviews if r["id"] != 99]
+                self.comments = []
+                return {}
+            return copy.deepcopy(next(r for r in self.reviews if r["id"] == 99))
+        if path == self.root + "/reviews/99/events":
+            self.final_events.append(payload["event"])
+            self.reviews[-1].update(state="CHANGES_REQUESTED" if payload["event"] == "REQUEST_CHANGES" else "COMMENTED",
+                                    submitted_at="9999-01-01T00:00:00Z", body=payload["body"])
             return copy.deepcopy(self.reviews[-1])
         if path == self.root + "/reviews/99/comments":
             return copy.deepcopy(self.comments)
@@ -94,7 +127,8 @@ class FinalReviewTest(unittest.TestCase):
     def gh_response(self, command, **kwargs):
         payload = json.loads(kwargs["input"]) if kwargs.get("input") is not None else None
         paginated = "--paginate" in command
-        value = self.fake_api(command[2], self.remaining, payload=payload, paginated=paginated)
+        method = command[command.index("--method") + 1] if "--method" in command else None
+        value = self.fake_api(command[2], self.remaining, payload=payload, paginated=paginated, method=method)
         return SimpleNamespace(stdout=json.dumps([value] if paginated else value), stderr="")
 
     def test_failed_post_retains_response_candidate_and_no_repost_guard(self):
@@ -154,7 +188,7 @@ class FinalReviewTest(unittest.TestCase):
 
         def lose_post_response(command, **kwargs):
             result = self.gh_response(command, **kwargs)
-            if kwargs.get("input") is not None:
+            if command[2].endswith("/events"):
                 raise subprocess.TimeoutExpired(command, 30, output=b'partial response', stderr=b'timed out')
             return result
 
@@ -220,7 +254,7 @@ class FinalReviewTest(unittest.TestCase):
                 (self.context / submitter.SUBMISSION_FILE).unlink(missing_ok=True)
                 result = self.submit(self.submission(level))
                 self.assertEqual("success", result["state"])
-                self.assertEqual("COMMENT", self.posts[-1]["event"])
+                self.assertEqual("COMMENT", self.final_events[-1])
 
     def test_round_limit_delivers_verified_findings_without_passing(self):
         submission = self.submission(2, converged=False, rounds=3)
@@ -229,8 +263,9 @@ class FinalReviewTest(unittest.TestCase):
         self.assertFalse(result["converged"])
         self.assertEqual(3, result["rounds"])
         self.assertEqual(submission["incomplete_reason"], result["incomplete_reason"])
-        self.assertEqual(submission["comments"], self.posts[0]["comments"])
-        self.assertEqual("COMMENT", self.posts[0]["event"])
+        self.assertEqual(submission["comments"], [{k: c[k] for k in ("path", "position", "body")}
+                                                  for c in self.comments])
+        self.assertEqual("COMMENT", self.final_events[0])
         self.assertIn("Review did not converge after 3 rounds", self.posts[0]["body"])
         self.assertIn(submission["incomplete_reason"], self.posts[0]["body"])
         self.assertEqual(result, self.verify())
@@ -522,7 +557,7 @@ class FinalReviewTest(unittest.TestCase):
 
         def lost_response(path, *args, **kwargs):
             result = original_api(path, *args, **kwargs)
-            if kwargs.get("payload") is not None:
+            if path.endswith("/events"):
                 raise subprocess.TimeoutExpired("gh", 30)
             return result
 
@@ -541,6 +576,7 @@ class FinalReviewTest(unittest.TestCase):
             with self.subTest(priority=level, converged=converged, capacity=capacity), tempfile.TemporaryDirectory() as tmp:
                 context = Path(tmp)
                 (context / "codex_goal_prompt.txt").write_text("Review the PR")
+                (context / "pr.diff").write_text(self.diff)
                 args = SimpleNamespace(context_dir=context, cwd=context, **{
                     k: self.run[k] for k in ("repository", "pr_number", "head_sha", "base_sha")
                 }, model="gpt-6-sol", effort="xhigh", budget_seconds=60)
@@ -592,6 +628,186 @@ class FinalReviewTest(unittest.TestCase):
         with self.assertRaises(subprocess.CalledProcessError):
             self.verify()
 
+    def test_five_comments_use_one_draft_and_separate_mutations(self):
+        submission = self.submission()
+        submission["comments"] = [dict(submission["comments"][0], position=i,
+                                       body=f"[P1] Finding {i}") for i in range(1, 6)]
+        self.submit(submission)
+        self.assertEqual(1, len(self.posts))
+        self.assertNotIn("event", self.posts[0])
+        self.assertEqual([], self.posts[0]["comments"])
+        self.assertEqual(5, len(self.thread_inputs))
+        self.assertEqual({"review-node-99"}, {c["pullRequestReviewId"] for c in self.thread_inputs})
+        self.assertEqual([1, 2, 3, 4, 5], [c["line"] for c in self.thread_inputs])
+        self.assertEqual(["REQUEST_CHANGES"], self.final_events)
+        self.assertEqual([], self.deletes)
+        self.assertEqual("submitted", json.loads((self.context / submitter.DRAFT_FILE).read_text())["stage"])
+
+    def test_invalid_anchor_is_rejected_before_intent_or_write(self):
+        submission = self.submission()
+        submission["comments"][0]["position"] = 999
+        with self.assertRaisesRegex(ValueError, "not commentable"):
+            self.submit(submission)
+        self.assertEqual([], self.posts)
+        self.assertFalse((self.context / submitter.SUBMISSION_FILE).exists())
+
+    def test_add_failure_deletes_only_owned_draft_and_never_reposts(self):
+        self.reviews = [{"id": 88, "state": "PENDING", "body": "Another draft", "commit_id": self.run["head_sha"],
+                         "user": {"login": "human"}}]
+        original = self.fake_api
+
+        def fail_add(path, *args, **kwargs):
+            if path == "graphql":
+                raise submitter.GitHubAPIError(path, "POST", ValueError("rejected"), stderr="Line could not be resolved")
+            return original(path, *args, **kwargs)
+
+        self.api.side_effect = fail_add
+        with self.assertRaisesRegex(ValueError, "Line could not be resolved"), redirect_stderr(io.StringIO()):
+            self.submit()
+        self.assertEqual([99], self.deletes)
+        self.assertEqual([88], [r["id"] for r in self.reviews])
+        self.assertEqual([], self.final_events)
+        with self.assertRaisesRegex(ValueError, "refusing to repost"):
+            self.submit()
+        self.assertEqual(1, len(self.posts))
+
+    def test_lost_create_response_cleans_up_without_recreating(self):
+        original = self.fake_api
+
+        def lose_create(path, *args, **kwargs):
+            result = original(path, *args, **kwargs)
+            if path == self.root + "/reviews" and kwargs.get("payload") is not None:
+                raise subprocess.TimeoutExpired("gh", 30)
+            return result
+
+        self.api.side_effect = lose_create
+        with self.assertRaises(subprocess.TimeoutExpired):
+            self.submit()
+        self.assertEqual([99], self.deletes)
+        self.assertEqual([], self.final_events)
+        with self.assertRaisesRegex(ValueError, "refusing to repost"):
+            self.submit()
+        self.assertEqual(1, len(self.posts))
+
+    def test_wrong_thread_location_is_not_published(self):
+        original = self.fake_api
+
+        def wrong_line(path, *args, **kwargs):
+            result = original(path, *args, **kwargs)
+            if path == "graphql":
+                result["data"]["addPullRequestReviewThread"]["thread"]["line"] += 1
+            return result
+
+        self.api.side_effect = wrong_line
+        with self.assertRaisesRegex(ValueError, "location did not match"):
+            self.submit()
+        self.assertEqual([99], self.deletes)
+        self.assertEqual([], self.final_events)
+
+    def test_incomplete_draft_is_not_published(self):
+        original = self.fake_api
+
+        def missing_comment(path, *args, **kwargs):
+            result = original(path, *args, **kwargs)
+            return [] if path.endswith("/comments") else result
+
+        self.api.side_effect = missing_comment
+        with self.assertRaisesRegex(ValueError, "inline comments"):
+            self.submit()
+        self.assertEqual([99], self.deletes)
+        self.assertEqual([], self.final_events)
+
+    def test_head_moves_between_adds_deletes_draft_without_publishing(self):
+        original = self.fake_api
+
+        def move_head(path, *args, **kwargs):
+            result = original(path, *args, **kwargs)
+            if path == "graphql":
+                self.pr["head"]["sha"] = "c" * 40
+            return result
+
+        submission = self.submission()
+        submission["comments"].append(dict(submission["comments"][0], position=13))
+        self.api.side_effect = move_head
+        with self.assertRaisesRegex(ValueError, "stale"):
+            self.submit(submission)
+        self.assertEqual(1, len(self.thread_inputs))
+        self.assertEqual([99], self.deletes)
+        self.assertEqual([], self.final_events)
+
+    def test_ambiguous_publish_is_never_retried_or_deleted(self):
+        original = self.fake_api
+
+        def fail_publish(path, *args, **kwargs):
+            if path.endswith("/events"):
+                raise subprocess.TimeoutExpired("gh", 30)
+            return original(path, *args, **kwargs)
+
+        self.api.side_effect = fail_publish
+        with self.assertRaises(subprocess.TimeoutExpired):
+            self.submit()
+        self.assertEqual([], self.deletes)
+        self.assertEqual("PENDING", self.reviews[0]["state"])
+        with self.assertRaisesRegex(ValueError, "did not match"):
+            self.verify()
+        with self.assertRaisesRegex(ValueError, "did not match"):
+            self.submit()
+        self.assertEqual(1, len(self.posts))
+        self.assertEqual(1, len(self.thread_inputs))
+        self.assertEqual("submitting", json.loads((self.context / submitter.DRAFT_FILE).read_text())["stage"])
+
+    def test_cleanup_failure_preserves_original_error_and_draft_id(self):
+        original = self.fake_api
+
+        def fail_add_and_cleanup(path, *args, **kwargs):
+            if path == "graphql":
+                raise ValueError("original add failure")
+            if kwargs.get("method") == "DELETE":
+                raise ValueError("cleanup unavailable")
+            return original(path, *args, **kwargs)
+
+        self.api.side_effect = fail_add_and_cleanup
+        with self.assertRaisesRegex(ValueError, "original add failure"), redirect_stderr(io.StringIO()):
+            self.submit()
+        progress = json.loads((self.context / submitter.DRAFT_FILE).read_text())
+        self.assertEqual("cleanup_failed", progress["stage"])
+        self.assertEqual(99, progress["review_id"])
+        self.assertIn("cleanup unavailable", progress["cleanup_error"])
+        self.assertEqual([], self.final_events)
+
+    def test_parallel_bot_review_during_adds_prevents_publication(self):
+        original = self.fake_api
+
+        def other_review(path, *args, **kwargs):
+            result = original(path, *args, **kwargs)
+            if path == "graphql":
+                self.reviews.append({"id": 100, "state": "COMMENTED", "body": "Other review",
+                                     "user": {"login": "github-actions[bot]"},
+                                     "commit_id": self.run["head_sha"], "submitted_at": "9999-01-01T00:00:00Z"})
+            return result
+
+        self.api.side_effect = other_review
+        with self.assertRaisesRegex(ValueError, "Another bot review"):
+            self.submit()
+        self.assertEqual([], self.final_events)
+        self.assertEqual([99], self.deletes)
+        self.assertEqual([100], [r["id"] for r in self.reviews])
+
+    def test_changed_draft_is_neither_published_nor_deleted(self):
+        original = self.fake_api
+
+        def change_draft(path, *args, **kwargs):
+            result = original(path, *args, **kwargs)
+            if path == "graphql":
+                self.reviews[0]["body"] = "Edited outside this invocation"
+            return result
+
+        self.api.side_effect = change_draft
+        with self.assertRaisesRegex(ValueError, "Pending review author"):
+            self.submit()
+        self.assertEqual([], self.final_events)
+        self.assertEqual([], self.deletes)
+
     def test_schema_requires_explicit_severity_and_all_fields(self):
         for value in ({"body": "summary"}, dict(self.submission(), comments="no"),
                       dict(self.submission(), existing_blocking_comment_ids=[1, 1]),
@@ -602,6 +818,47 @@ class FinalReviewTest(unittest.TestCase):
         self.assertEqual([], self.posts)
 
 
+class DiffLocationsTest(unittest.TestCase):
+    def locations(self, diff, pairs):
+        with tempfile.TemporaryDirectory() as tmp:
+            context = Path(tmp)
+            (context / "pr.diff").write_text(diff)
+            return submitter.inline_locations(context, [{"path": p, "position": n} for p, n in pairs])
+
+    def test_context_deletion_addition_and_later_hunk(self):
+        diff = ("diff --git a/a.py b/a.py\n--- a/a.py\n+++ b/a.py\n"
+                "@@ -10,3 +10,3 @@ function\n context\n-old\n+new\n last\n"
+                "@@ -30 +30 @@\n-before\n+after\n")
+        self.assertEqual([{"line": 10, "side": "RIGHT"}, {"line": 11, "side": "LEFT"},
+                          {"line": 11, "side": "RIGHT"}, {"line": 30, "side": "LEFT"},
+                          {"line": 30, "side": "RIGHT"}],
+                         self.locations(diff, [("a.py", i) for i in (1, 2, 3, 6, 7)]))
+        with self.assertRaisesRegex(ValueError, "not commentable"):
+            self.locations(diff, [("a.py", 5)])
+
+    def test_deleted_file_and_renamed_file_use_correct_paths(self):
+        diff = ("diff --git a/old.py b/old.py\n--- a/old.py\n+++ /dev/null\n@@ -1 +0,0 @@\n-gone\n"
+                "diff --git a/before.py b/after.py\n--- a/before.py\n+++ b/after.py\n"
+                "@@ -1 +1 @@\n-old\n+new\n")
+        self.assertEqual([{"line": 1, "side": "LEFT"}, {"line": 1, "side": "RIGHT"}],
+                         self.locations(diff, [("old.py", 1), ("after.py", 2)]))
+
+    def test_git_quoted_utf8_and_space_paths(self):
+        diff = ('diff --git "a/\\344\\270\\255.py" "b/\\344\\270\\255.py"\n'
+                '--- /dev/null\n+++ "b/\\344\\270\\255.py"\n@@ -0,0 +1 @@\n+text\n'
+                'diff --git a/has space.py b/has space.py\n--- /dev/null\n+++ b/has space.py\t\n'
+                '@@ -0,0 +1 @@\n+text\n')
+        self.assertEqual([{"line": 1, "side": "RIGHT"}] * 2,
+                         self.locations(diff, [("中.py", 1), ("has space.py", 1)]))
+
+    def test_no_newline_markers_are_not_anchors(self):
+        diff = ("diff --git a/a.py b/a.py\n--- a/a.py\n+++ b/a.py\n@@ -1 +1 @@\n"
+                "-old\n\\ No newline at end of file\n+new\n\\ No newline at end of file\n")
+        self.assertEqual([{"line": 1, "side": "RIGHT"}], self.locations(diff, [("a.py", 3)]))
+        with self.assertRaisesRegex(ValueError, "not commentable"):
+            self.locations(diff, [("a.py", 2)])
+
+
 class ReviewWorkflowTest(unittest.TestCase):
     def test_queue_is_per_pr_and_failure_artifact_is_explicit(self):
         workflow = (Path(__file__).resolve().parents[1] / "workflows/code-review-runner.yml").read_text()
@@ -619,13 +876,31 @@ class ReviewWorkflowTest(unittest.TestCase):
         self.assertIn("steps.review.outcome == 'failure'", artifact)
         paths = artifact.split("          path: |\n", 1)[1].split("          #", 1)[0]
         self.assertEqual({"${{ env.REVIEW_CONTEXT_DIR }}/" + name for name in (
-            submitter.SUBMISSION_FILE, submitter.REJECTION_FILE, submitter.API_ERRORS_FILE
+            submitter.SUBMISSION_FILE, submitter.REJECTION_FILE, submitter.API_ERRORS_FILE, submitter.DRAFT_FILE
         )}, {line.strip() for line in paths.splitlines()})
         self.assertIn("include-hidden-files: true", artifact)
         self.assertIn("if-no-files-found: ignore", artifact)
 
 
 class GitHubAPITest(unittest.TestCase):
+    def test_graphql_errors_in_http_success_are_recorded_as_failure(self):
+        response = {"data": {"addPullRequestReviewThread": None},
+                    "errors": [{"message": "Line could not be resolved"}]}
+        debug = ("> POST /graphql HTTP/2.0\n\n< HTTP/2.0 200 OK\n"
+                 "< X-Github-Request-Id: GRAPHQL-REQUEST\n\n" + json.dumps(response))
+        with tempfile.TemporaryDirectory() as tmp, \
+             mock.patch.dict(submitter.os.environ, {"REVIEW_GITHUB_DIAGNOSTICS_DIR": tmp}), \
+             mock.patch.object(submitter.subprocess, "run", return_value=SimpleNamespace(
+                 stdout=json.dumps(response), stderr=debug
+             )), redirect_stderr(io.StringIO()):
+            with self.assertRaises(submitter.GitHubAPIError) as failure:
+                submitter.api("graphql", lambda: 60, payload={"query": "mutation"})
+            self.assertIn("Line could not be resolved", failure.exception.details["stdout"])
+            records = [json.loads(p.read_text()) for p in Path(tmp).glob("*.json")]
+            self.assertEqual(1, len(records))
+            self.assertEqual(200, records[0]["http"]["status"])
+            self.assertEqual("GRAPHQL-REQUEST", records[0]["http"]["headers"]["x-github-request-id"])
+
     def test_read_all_pages_and_bound_timeout(self):
         with mock.patch.object(submitter.subprocess, "run", return_value=SimpleNamespace(
             stderr="", stdout='[[{"id":1}], [{"id":2}]]'

.github/workflows/code-review-runner.yml

@@ -1022,6 +1022,7 @@ jobs:
             ${{ env.REVIEW_CONTEXT_DIR }}/review-submission.json
             ${{ env.REVIEW_CONTEXT_DIR }}/review-submission-rejected.json
             ${{ env.REVIEW_CONTEXT_DIR }}/review-api-errors.jsonl
+            ${{ env.REVIEW_CONTEXT_DIR }}/review-draft.json
           # The context directory is hidden; include only these explicit files.
           include-hidden-files: true
           if-no-files-found: ignore