[fix](ci) Retry transient review GET failures within bounded budgets (#69510)

Related PR: #69469

Problem Summary: A single transient GitHub response currently aborts the
review pipeline. [Input resolution for
#69322](https://github.com/apache/doris/actions/runs/39794013905/job/119876748768)
stopped on a GET 503 despite `Retry-After: 2`; [aggregate resolution for
#69438](https://github.com/apache/doris/actions/runs/39760934956/job/119764632659)
stopped on a paginated GET 504 after emitting partial output.

The shared helper now retries recognized body-free REST GETs on
502/503/504, for at most three attempts. Each attempt has a 30-second
limit; attempts and waits share a 90-second budget or the caller's
shorter timeout. It respects Retry-After, records every failed attempt,
discards failed pagination output, and returns only a successful
attempt's complete output. Exhaustion still fails. Writes, GraphQL,
unknown argument forms, other HTTP errors, transport/timeouts and JSON
parsing errors are not retried.

Input resolution can also fail before publishing its PR-number output.
Failure notices now fall back to the authorized trigger's numeric PR
identity, explicitly select the repository, and name the
input-resolution failure in both the notice and terminal error. This
fallback does not permit status publication for an unresolved head/base.

This change can merge independently. It includes the same explicit
`--repo` argument as #69469; a local combination with #69469 and #69504
merged cleanly and passed the macOS suite.

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

@@ -8,7 +8,21 @@ submission helper and head watchdog.
 Use python3 "$REVIEW_GH_HELPER" api ... in these workflow steps. Python helpers
 use run_gh with captured text output, check=True and their existing timeout;
 use decode_json when decoding its result. Request arguments, normal stdout,
-write deduplication, permission decisions and retry policy remain unchanged.
+write deduplication and permission decisions remain unchanged.
+
+Body-free REST GETs using the pipeline's known gh api arguments retry HTTP
+502/503/504 at most twice. Each attempt is limited to 30 seconds; all attempts
+and waits share 90 seconds, or the caller's shorter timeout. Retry-After
+(seconds or an HTTP date) is respected; a malformed header or a wait exceeding
+the remaining budget stops retries. Each failed attempt keeps its diagnostic.
+Only a successful attempt's complete stdout is returned; failed pagination
+output is discarded before restarting the full read. Exhaustion still fails.
+
+Writes, GraphQL, fields/input, unknown argument forms, transport/timeouts, JSON
+parse failures and other HTTP statuses retain single-attempt behavior. The
+failure notification can use the authorized event/input PR number when input
+resolution failed before producing outputs; this does not authorize status
+writes for an unresolved head/base.
 
 On a failure, the helper prints a short diagnostic and saves one JSON record
 under REVIEW_GITHUB_DIAGNOSTICS_DIR. Records include the command exit status or

.github/scripts/review_github_api.py

@@ -16,7 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-"""Capture gh HTTP failures without changing stdout, exit status, or retry policy."""
+"""Capture gh HTTP failures and retry transient failures of read-only REST GETs."""
 
 import json
 import os
@@ -26,6 +26,7 @@ import sys
 import time
 import uuid
 from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
 from pathlib import Path
 from urllib.parse import urlsplit
 
@@ -164,24 +165,92 @@ def record_failure(command, error, stdout="", stderr="", *, diagnostic=None):
     return ordinary, record
 
 
+def is_read_only_get(command):
+    """Recognize only the body-free gh api forms used by the review pipeline.
+
+    Unknown options remain single-attempt. In particular, fields/input imply
+    POST in gh, and GraphQL uses POST even without an explicit method. Never
+    infer safety from the last HTTP response: a write can redirect to GET.
+    """
+    if len(command) < 3 or Path(command[0]).name != "gh" or command[1] != "api":
+        return False
+    endpoint = None
+    args = iter(command[2:])
+    for arg in args:
+        if arg in ("--paginate", "--slurp", "-XGET", "--method=GET"):
+            continue
+        if arg in ("-X", "--method"):
+            if next(args, None) != "GET":
+                return False
+        elif arg in ("-H", "--header"):
+            if next(args, None) is None:
+                return False
+        elif arg.startswith("-") or endpoint is not None:
+            return False
+        else:
+            endpoint = arg
+    return endpoint is not None and urlsplit(endpoint).path.rstrip("/").split("/")[-1] != "graphql"
+
+
+def retry_delay(response, attempt):
+    """Do not shorten a server-requested wait, or guess if it is malformed."""
+    delay = 2 ** attempt
+    value = response["headers"].get("retry-after")
+    if value is None:
+        return delay
+    try:
+        if value.isdigit():
+            return max(delay, int(value))
+        when = parsedate_to_datetime(value)
+        return max(delay, (when - datetime.now(timezone.utc)).total_seconds())
+    except (TypeError, ValueError, OverflowError):
+        return None
+
+
 def run_gh(command, **kwargs):
-    """subprocess.run contract for captured text gh calls, with failure evidence."""
+    """Buffer each attempt; a caller timeout bounds the entire GET operation."""
     env = dict(kwargs.pop("env", os.environ))
     env.update(GH_DEBUG="api", NO_COLOR="1")
-    try:
-        result = subprocess.run(command, env=env, **kwargs)
-    except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as error:
-        ordinary, record = record_failure(command, error, error.stdout, error.stderr)
-        error.stderr = ordinary + "\n" + record["summary"]
-        # Preserve the original return code/timeout and stdout for callers.
-        raise
-    except OSError as error:
-        record_failure(command, error)
-        raise
-    response, ordinary = split_trace(result.stderr)
-    result.stderr = ordinary
-    result.github_response = response
-    return result
+    retry_get = (is_read_only_get(command) and kwargs.get("check") is True
+                 and kwargs.get("capture_output") is True and kwargs.get("text") is True
+                 and kwargs.get("input") is None and kwargs.get("stdin") is None)
+    if retry_get:
+        # Fit within the runner's two-minute input-resolution step. Existing
+        # shorter submission/watchdog budgets include all attempts and sleeps.
+        budget = min(90, kwargs.get("timeout") if kwargs.get("timeout") is not None else 90)
+        deadline = time.monotonic() + budget
+    for attempt in range(1, 4):
+        options = dict(kwargs)
+        if retry_get:
+            options["timeout"] = min(30, max(0, deadline - time.monotonic()))
+        try:
+            result = subprocess.run(command, env=env, **options)
+        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as error:
+            ordinary, record = record_failure(command, error, error.stdout, error.stderr)
+            error.stderr = ordinary + "\n" + record["summary"]
+            if retry_get:
+                # gh --paginate may emit incomplete JSON before a later page
+                # fails. Retain it only in diagnostics, never in caller output.
+                error.stdout = ""
+                response = record["http"]
+                if (isinstance(error, subprocess.CalledProcessError)
+                        and error.returncode > 0 and response["method"] == "GET"
+                        and response["status"] in (502, 503, 504) and attempt < 3):
+                    delay = retry_delay(response, attempt)
+                    if delay is not None and delay < deadline - time.monotonic():
+                        print(f"Retrying GitHub GET after HTTP {response['status']} in {delay:g}s "
+                              f"(attempt {attempt + 1}/3)", file=sys.stderr, flush=True)
+                        time.sleep(delay)
+                        if time.monotonic() < deadline:
+                            continue
+            raise
+        except OSError as error:
+            record_failure(command, error)
+            raise
+        response, ordinary = split_trace(result.stderr)
+        result.stderr = ordinary
+        result.github_response = response
+        return result
 
 
 def decode_json(result, command):
@@ -201,6 +270,9 @@ def main():
         sys.stdout.write(error.stdout or "")
         print(error.stderr, file=sys.stderr)
         return error.returncode if error.returncode >= 0 else 128 - error.returncode
+    except subprocess.TimeoutExpired as error:
+        print(error.stderr, file=sys.stderr)
+        return 124
     except OSError as error:
         print(f"Could not execute gh ({type(error).__name__})", file=sys.stderr)
         return 1

.github/scripts/test_review_get_retry.py

@@ -0,0 +1,297 @@
+#!/usr/bin/env python3
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import http.server
+import io
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+from contextlib import redirect_stderr
+from datetime import datetime, timedelta, timezone
+from email.utils import format_datetime
+from pathlib import Path
+from unittest import mock
+from urllib.parse import parse_qs, urlsplit
+
+import review_github_api as api
+
+
+def trace(status, *, method="GET", retry_after=None):
+    header = "" if retry_after is None else f"< Retry-After: {retry_after}\n"
+    return ("* Request at now\n> " + method + " /repos/a/b/pulls HTTP/2.0\n"
+            + f"< HTTP/2.0 {status} Result\n< X-GitHub-Request-Id: retry-test\n"
+            + header + "\nresponse\n* Request took 1ms\n")
+
+
+class ReadRetryTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.enterContext(mock.patch.dict(os.environ, {"REVIEW_GITHUB_DIAGNOSTICS_DIR": self.tmp.name}))
+        self.enterContext(redirect_stderr(io.StringIO()))
+        self.now = 0
+        self.sleeps = []
+        self.enterContext(mock.patch.object(api.time, "monotonic", side_effect=lambda: self.now))
+        self.enterContext(mock.patch.object(api.time, "sleep", side_effect=self.sleep))
+        self.command = ["gh", "api", "repos/a/b/pulls"]
+
+    def sleep(self, seconds):
+        self.sleeps.append(seconds)
+        self.now += seconds
+
+    def failure(self, status=503, *, method="GET", retry_after=None, stdout="partial"):
+        return subprocess.CalledProcessError(1, self.command, stdout,
+                                             trace(status, method=method, retry_after=retry_after))
+
+    def success(self, stdout='[{"id":2}]'):
+        return subprocess.CompletedProcess(self.command, 0, stdout, trace(200))
+
+    def run_get(self, command=None, **kwargs):
+        return api.run_gh(command or self.command, check=True, capture_output=True, text=True, **kwargs)
+
+    def records(self):
+        return [json.loads(p.read_text()) for p in Path(self.tmp.name).glob("*.json")]
+
+    def test_known_get_forms_only(self):
+        for args in (["repos/a/b/pulls"], ["--paginate", "--slurp", "repos/a/b/pulls"],
+                     ["repos/a/b/pulls", "-H", "Accept: application/json"],
+                     ["-X", "GET", "repos/a/b/pulls"], ["--method=GET", "repos/a/b/pulls"],
+                     ["-XGET", "repos/a/b/pulls"]):
+            with self.subTest(args=args):
+                self.assertTrue(api.is_read_only_get(["gh", "api", *args]))
+        for args in (["graphql"], ["https://api.github.com/graphql"],
+                     ["repos/a/b/pulls", "-f", "title=x"], ["repos/a/b/pulls", "-ftitle=x"],
+                     ["repos/a/b/pulls", "--field=title=x"], ["repos/a/b/pulls", "-Ftitle=x"],
+                     ["repos/a/b/pulls", "--input", "-"], ["repos/a/b/pulls", "--input=-"],
+                     ["repos/a/b/pulls", "-X", "POST"], ["repos/a/b/pulls", "-XPATCH"],
+                     ["repos/a/b/pulls", "--method=DELETE"],
+                     ["repos/a/b/pulls", "-XGET", "-XPOST"],
+                     ["repos/a/b/pulls", "--unknown"], ["repos/a/b/pulls", "-H"], []):
+            with self.subTest(args=args):
+                self.assertFalse(api.is_read_only_get(["gh", "api", *args]))
+        self.assertFalse(api.is_read_only_get(["gh", "pr", "comment", "1"]))
+
+    def test_transient_statuses_retry_and_return_only_complete_success(self):
+        for status in (502, 503, 504):
+            with self.subTest(status=status):
+                with mock.patch.object(api.subprocess, "run", side_effect=[self.failure(status), self.success()]) as run:
+                    result = self.run_get()
+                self.assertEqual(result.stdout, '[{"id":2}]')
+                self.assertEqual(run.call_count, 2)
+                self.assertEqual(result.stderr, "")
+        self.assertEqual(self.sleeps, [2, 2, 2])
+        self.assertEqual(len(self.records()), 3)
+        self.assertTrue(all(r["stdout_excerpt"] == "partial" for r in self.records()))
+
+    def test_retry_after_seconds_are_respected(self):
+        with mock.patch.object(api.subprocess, "run", side_effect=[self.failure(retry_after="7"), self.success()]):
+            self.run_get()
+        self.assertEqual(self.sleeps, [7])
+
+    def test_retry_after_dates_and_invalid_values(self):
+        now = datetime(2026, 10, 23, tzinfo=timezone.utc)
+        with mock.patch.object(api, "datetime") as clock:
+            clock.now.return_value = now
+            for value, expected in ((format_datetime(now + timedelta(seconds=9)), 9),
+                                    (format_datetime(now - timedelta(seconds=9)), 2),
+                                    ("invalid", None), ("-1", None), ("9" * 5000, None)):
+                with self.subTest(value=value[:40]):
+                    self.assertEqual(api.retry_delay({"headers": {"retry-after": value}}, 1), expected)
+
+    def test_long_or_invalid_retry_after_does_not_retry_early(self):
+        for value in ("1000", "bad-date"):
+            with self.subTest(value=value), mock.patch.object(api.subprocess, "run", side_effect=self.failure(retry_after=value)) as run:
+                with self.assertRaises(subprocess.CalledProcessError):
+                    self.run_get()
+                self.assertEqual(run.call_count, 1)
+        self.assertEqual(self.sleeps, [])
+
+    def test_caller_budget_includes_attempts_and_sleep(self):
+        def first(*args, **kwargs):
+            self.now += 3
+            raise self.failure()
+        with mock.patch.object(api.subprocess, "run", side_effect=[self.failure(), self.success()]) as run:
+            self.run_get(timeout=10)
+        self.assertEqual([c.kwargs["timeout"] for c in run.call_args_list], [10, 8])
+        self.assertEqual(self.sleeps, [2])
+        self.now = 0
+        with mock.patch.object(api.subprocess, "run", side_effect=first) as run:
+            with self.assertRaises(subprocess.CalledProcessError):
+                self.run_get(timeout=5)
+        self.assertEqual(run.call_count, 1)  # No room for the server/backoff wait.
+
+    def test_default_budget_caps_each_attempt_and_whole_operation(self):
+        def fail(*args, **kwargs):
+            self.now += 29
+            raise self.failure()
+        with mock.patch.object(api.subprocess, "run", side_effect=fail) as run:
+            with self.assertRaises(subprocess.CalledProcessError):
+                self.run_get()
+        self.assertEqual([c.kwargs["timeout"] for c in run.call_args_list], [30, 30, 26])
+        self.assertEqual(self.sleeps, [2, 4])
+
+    def test_exhaustion_keeps_all_diagnostics_but_no_partial_stdout(self):
+        with mock.patch.object(api.subprocess, "run", side_effect=[self.failure() for _ in range(3)]) as run:
+            with self.assertRaises(subprocess.CalledProcessError) as caught:
+                self.run_get()
+        self.assertEqual(run.call_count, 3)
+        self.assertEqual(caught.exception.stdout, "")
+        self.assertEqual(caught.exception.returncode, 1)
+        self.assertIn("retry-test", caught.exception.stderr)
+        self.assertEqual(len(self.records()), 3)
+
+    def test_non_transient_statuses_are_not_retried(self):
+        for status in (400, 401, 403, 404, 422, 429, 500):
+            with self.subTest(status=status), mock.patch.object(api.subprocess, "run", side_effect=self.failure(status)) as run:
+                with self.assertRaises(subprocess.CalledProcessError):
+                    self.run_get()
+                self.assertEqual(run.call_count, 1)
+        self.assertEqual(self.sleeps, [])
+
+    def test_writes_never_retry_even_if_redirect_ends_in_get(self):
+        commands = [["gh", "pr", "comment", "1", "--body", "x"],
+                    [*self.command, "-X", "POST"], [*self.command, "-f", "state=pending"],
+                    [*self.command, "--input", "-"], ["gh", "api", "graphql"]]
+        for command in commands:
+            with self.subTest(command=command), mock.patch.object(api.subprocess, "run", side_effect=self.failure()) as run:
+                with self.assertRaises(subprocess.CalledProcessError) as caught:
+                    self.run_get(command)
+                self.assertEqual(run.call_count, 1)
+                self.assertEqual(caught.exception.stdout, "partial")
+                self.assertNotIn("timeout", run.call_args.kwargs)
+        self.assertEqual(self.sleeps, [])
+
+    def test_input_and_uncaptured_calls_do_not_retry(self):
+        for options in ({"input": "body"}, {"capture_output": False}):
+            kwargs = dict(check=True, capture_output=True, text=True)
+            kwargs.update(options)
+            with self.subTest(options=options), mock.patch.object(api.subprocess, "run", side_effect=self.failure()) as run:
+                with self.assertRaises(subprocess.CalledProcessError):
+                    api.run_gh(self.command, **kwargs)
+                self.assertEqual(run.call_count, 1)
+
+    def test_timeouts_transport_errors_and_signals_do_not_retry(self):
+        signal = self.failure()
+        signal.returncode = -15
+        for error in (subprocess.TimeoutExpired(self.command, 3, output="partial", stderr=trace(503)),
+                      OSError("transport"), signal,
+                      subprocess.CalledProcessError(1, self.command, "", "connection reset")):
+            with self.subTest(error=type(error)), mock.patch.object(api.subprocess, "run", side_effect=error) as run:
+                with self.assertRaises(type(error)):
+                    self.run_get()
+                self.assertEqual(run.call_count, 1)
+        self.assertEqual(self.sleeps, [])
+
+    def test_sleep_overrun_does_not_start_another_request(self):
+        with mock.patch.object(api.time, "sleep", side_effect=lambda _: setattr(self, "now", 90)), \
+             mock.patch.object(api.subprocess, "run", side_effect=self.failure()) as run:
+            with self.assertRaises(subprocess.CalledProcessError):
+                self.run_get()
+        self.assertEqual(run.call_count, 1)
+
+    def test_json_decode_failure_is_not_retried(self):
+        with mock.patch.object(api.subprocess, "run", return_value=self.success("{")) as run:
+            result = self.run_get()
+            with self.assertRaises(json.JSONDecodeError):
+                api.decode_json(result, self.command)
+        self.assertEqual(run.call_count, 1)
+
+    def test_cli_timeout_returns_124(self):
+        with mock.patch.object(api, "run_gh", side_effect=subprocess.TimeoutExpired(self.command, 30, stderr="timed out")):
+            self.assertEqual(api.main(), 124)
+
+
+@unittest.skipUnless(os.environ.get("RUN_GH_HTTP_TESTS") == "1" and shutil.which("gh"), "opt-in real gh loopback test")
+class GetRetryHTTPTest(unittest.TestCase):
+    def test_real_get_recovery_pagination_exhaustion_and_write_isolation(self):
+        requests = []
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def log_message(self, *_):
+                pass
+
+            def do_GET(self):
+                parsed = urlsplit(self.path)
+                page = parse_qs(parsed.query).get("page", [None])[0]
+                route = parsed.path + (f"?page={page}" if page is not None else "")
+                requests.append(("GET", route))
+                count = requests.count(("GET", route))
+                failed = route == "/always" or (count == 1 and route in ("/once", "/pages?page=2"))
+                status = 504 if failed and "page=2" in self.path else 503 if failed else 200
+                data = b"Service Unavailable" if failed else b'[{"id":2}]' if "page=2" in self.path else b'[{"id":1}]'
+                self.send_response(status)
+                self.send_header("Content-Type", "text/plain" if failed else "application/json")
+                self.send_header("Content-Length", str(len(data)))
+                self.send_header("Retry-After", "0")
+                if route == "/pages?page=1":
+                    self.send_header("Link", f'<http://127.0.0.1:{self.server.server_port}/pages?page=2>; rel="next"')
+                self.end_headers()
+                self.wfile.write(data)
+
+            def do_POST(self):
+                requests.append(("POST", self.path))
+                self.send_response(503)
+                self.send_header("Content-Length", "0")
+                self.end_headers()
+
+        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            with tempfile.TemporaryDirectory() as tmp:
+                env = {"PATH": os.environ["PATH"], "GH_TOKEN": "fake-only", "GH_ENTERPRISE_TOKEN": "fake-only",
+                       "GH_CONFIG_DIR": tmp, "GH_NO_UPDATE_NOTIFIER": "1", "GH_PROMPT_DISABLED": "1",
+                       "REVIEW_GITHUB_DIAGNOSTICS_DIR": tmp}
+                base = f"http://127.0.0.1:{server.server_port}"
+                helper = Path(__file__).with_name("review_github_api.py")
+                def call(path, *args):
+                    return subprocess.run([sys.executable, str(helper), "api", base + path, *args],
+                                          env=env, capture_output=True, text=True, timeout=20)
+                result = call("/once")
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(json.loads(result.stdout), [{"id": 1}])
+                self.assertEqual(requests, [("GET", "/once")] * 2)
+                requests.clear()
+                result = call("/pages?page=1", "--paginate", "--slurp")
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(json.loads(result.stdout), [[{"id": 1}], [{"id": 2}]])
+                self.assertEqual(requests, [("GET", "/pages?page=1"), ("GET", "/pages?page=2")] * 2)
+                requests.clear()
+                result = call("/always")
+                self.assertEqual(result.returncode, 1)
+                self.assertEqual(result.stdout, "")
+                self.assertEqual(requests, [("GET", "/always")] * 3)
+                requests.clear()
+                result = call("/write", "-X", "POST", "-f", "state=pending")
+                self.assertEqual(result.returncode, 1)
+                self.assertEqual(requests, [("POST", "/write")])
+                records = [json.loads(p.read_text()) for p in Path(tmp).glob("*.json")]
+                self.assertEqual(len(records), 6)
+                self.assertNotIn("fake-only", json.dumps(records))
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/scripts/test_review_input_failure.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Run actual input/failure steps without a checkout or real GitHub writes."""
+
+import json
+import os
+import re
+import subprocess
+import sys
+import tempfile
+import textwrap
+import unittest
+from pathlib import Path
+
+WORKFLOW = Path(__file__).resolve().parents[1] / "workflows/code-review-runner.yml"
+COMMENT = "Comment PR on review failure"
+FAIL = "Fail workflow if review failed"
+
+
+def step(name):
+    return re.search(rf"^      - name: {re.escape(name)}\n(.*?)(?=^      - name:|\Z)",
+                     WORKFLOW.read_text(), re.MULTILINE | re.DOTALL).group(1)
+
+
+class InputFailureTest(unittest.TestCase):
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.root = Path(tmp.name)
+        self.calls = self.root / "calls.jsonl"
+        self.output = self.root / "output"
+        self.output.touch()
+        tool = self.root / "gh"
+        tool.write_text(f"#!{sys.executable}\n" + textwrap.dedent("""\
+            import json, os, sys
+            from pathlib import Path
+            with Path(os.environ["GH_CALLS"]).open("a") as handle:
+                handle.write(json.dumps(sys.argv[1:]) + "\\n")
+            if sys.argv[1] == "api":
+                print("gh: HTTP 503", file=sys.stderr)
+                sys.exit(1)
+            sys.exit(int(os.environ.get("COMMENT_EXIT", "0")))
+        """))
+        tool.chmod(0o700)
+        self.context = {"github.repository": "example/repo", "github.server_url": "https://github.example.test",
+                        "github.run_id": "1", "github.event_name": "issue_comment",
+                        "github.event.issue.number": "69322", "github.event.comment.body": "/review",
+                        "steps.review_inputs.outcome": "failure", "steps.review_context.outcome": "skipped",
+                        "steps.review.outcome": "skipped", "secrets.GITHUB_TOKEN": "fake-only"}
+
+    def run_step(self, name, comment_exit=0):
+        block = step(name)
+        env = dict(os.environ, PATH=f"{self.root}:{os.environ['PATH']}",
+                   GH_CALLS=str(self.calls), COMMENT_EXIT=str(comment_exit),
+                   GITHUB_OUTPUT=str(self.output),
+                   REVIEW_GH_HELPER=str(Path(__file__).with_name("review_github_api.py")),
+                   REVIEW_GITHUB_DIAGNOSTICS_DIR=str(self.root / "diagnostics"))
+        bindings = block.split("        env:\n", 1)[1].split("        run:", 1)[0]
+        for line in bindings.splitlines():
+            key, value = line.strip().split(": ", 1)
+            env[key] = re.sub(r"\$\{\{\s*(.*?)\s*\}\}", lambda m: self.context.get(m[1], ""), value)
+        script = textwrap.dedent(block.split("        run: |\n", 1)[1])
+        return subprocess.run(["bash", "--noprofile", "--norc", "-eo", "pipefail", "-c", script],
+                              env=env, cwd=self.root, text=True, capture_output=True, timeout=10)
+
+    def last_comment(self):
+        args = json.loads(self.calls.read_text().splitlines()[-1])
+        self.assertEqual(args[:2], ["pr", "comment"])
+        self.assertEqual(args[args.index("--repo") + 1], "example/repo")
+        return args[2], args[args.index("--body") + 1]
+
+    def test_actual_failed_input_step_still_reports_event_pr(self):
+        result = self.run_step("Resolve review inputs")
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(self.output.read_text(), "")
+        self.assertFalse((self.root / ".git").exists())
+        result = self.run_step(COMMENT)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        number, body = self.last_comment()
+        self.assertEqual(number, "69322")
+        self.assertIn("'Resolve review inputs' ended with failure", body)
+        result = self.run_step(FAIL)
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("'Resolve review inputs' ended with failure", result.stdout)
+        sync = step("Sync Code Review check for current head")
+        self.assertIn("steps.review_inputs.outcome == 'success'", sync)
+        self.assertIn("needs.code-review.outputs.head_sha != ''", WORKFLOW.read_text())
+
+    def test_dispatch_uses_input_pr_without_issue_event(self):
+        self.context.update({"inputs.pr_number": "42", "github.event.issue.number": ""})
+        self.assertEqual(self.run_step(COMMENT).returncode, 0)
+        self.assertEqual(self.last_comment()[0], "42")
+
+    def test_resolved_pr_takes_precedence_and_existing_failure_reason_survives(self):
+        self.context.update({"steps.review_inputs.outputs.pr_number": "42",
+                             "inputs.pr_number": "43", "steps.review_inputs.outcome": "success",
+                             "steps.auth.outputs.failure_reason": "No eligible credential"})
+        self.assertEqual(self.run_step(COMMENT).returncode, 0)
+        number, body = self.last_comment()
+        self.assertEqual(number, "42")
+        self.assertIn("No eligible credential", body)
+
+    def test_cancelled_inputs_are_identified(self):
+        self.context["steps.review_inputs.outcome"] = "cancelled"
+        self.assertEqual(self.run_step(COMMENT).returncode, 0)
+        self.assertIn("'Resolve review inputs' ended with cancelled", self.last_comment()[1])
+        result = self.run_step(FAIL)
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("'Resolve review inputs' ended with cancelled", result.stdout)
+
+    def test_missing_or_invalid_pr_never_posts(self):
+        for value in ("", "0", "-1", "branch-name", "42\n43"):
+            with self.subTest(value=value):
+                self.context["github.event.issue.number"] = value
+                result = self.run_step(COMMENT)
+                self.assertEqual(result.returncode, 1)
+                self.assertFalse(self.calls.exists())
+                self.assertIn("No valid PR number", result.stdout)
+
+    def test_failed_notification_is_not_retried_and_terminal_cause_is_preserved(self):
+        result = self.run_step(COMMENT, comment_exit=1)
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(len(self.calls.read_text().splitlines()), 1)
+        result = self.run_step(FAIL)
+        self.assertEqual(result.returncode, 1)
+        self.assertIn("'Resolve review inputs' ended with failure", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/scripts/test_submit_review.py

@@ -906,7 +906,9 @@ class GitHubAPITest(unittest.TestCase):
             stderr="", stdout='[[{"id":1}], [{"id":2}]]'
         )) as command:
             self.assertEqual([{"id": 1}, {"id": 2}], submitter.api("reviews", lambda: 7, paginated=True))
-        self.assertEqual(7, command.call_args.kwargs["timeout"])
+        # The shared GET deadline also accounts for helper setup time.
+        self.assertGreater(command.call_args.kwargs["timeout"], 0)
+        self.assertLessEqual(command.call_args.kwargs["timeout"], 7)
         self.assertIn("--paginate", command.call_args.args[0])
         self.assertIn("--slurp", command.call_args.args[0])
 

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

@@ -1088,6 +1088,7 @@ jobs:
         timeout-minutes: 2
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          REVIEW_INPUTS_OUTCOME: ${{ steps.review_inputs.outcome }}
           AUTH_FAILURE_REASON: ${{ steps.auth.outputs.failure_reason }}
           AUTH_INVALID_REASON: ${{ steps.review.outputs.auth_invalid_reason }}
           AUTH_INVALID_RECORD_OUTCOME: ${{ steps.auth_invalid_record.outcome }}
@@ -1096,11 +1097,23 @@ jobs:
           REVIEW_CONTEXT_OUTCOME: ${{ steps.review_context.outcome }}
           REVIEW_FAILURE_REASON: ${{ steps.review.outputs.failure_reason }}
           REVIEW_OUTCOME: ${{ steps.review.outcome }}
+          REPO: ${{ github.repository }}
           PR_NUMBER: ${{ steps.review_inputs.outputs.pr_number }}
+          INPUT_PR_NUMBER: ${{ inputs.pr_number }}
+          ISSUE_NUMBER: ${{ github.event.issue.number }}
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
         run: |
+          # Input resolution can fail before publishing the PR number. Use only
+          # the authorized trigger's numeric PR identity for the failure notice.
+          PR_NUMBER="${PR_NUMBER:-${INPUT_PR_NUMBER:-$ISSUE_NUMBER}}"
+          if ! [[ "$PR_NUMBER" =~ ^[1-9][0-9]*$ ]]; then
+            echo "::error::No valid PR number is available for the failure notice."
+            exit 1
+          fi
           recovery_msg="Please inspect the workflow logs and rerun the review after the underlying issue is resolved."
-          if [ -n "$AUTH_FAILURE_REASON" ]; then
+          if [ "$REVIEW_INPUTS_OUTCOME" = "failure" ] || [ "$REVIEW_INPUTS_OUTCOME" = "cancelled" ]; then
+            error_msg="Step 'Resolve review inputs' ended with $REVIEW_INPUTS_OUTCOME; inspect its logs for the original error."
+          elif [ -n "$AUTH_FAILURE_REASON" ]; then
             error_msg="$AUTH_FAILURE_REASON"
           elif [ "$AUTH_INVALID_REASON" = "refresh_token_reused" ] || [ "$AUTH_INVALID_REASON" = "refresh_token_invalidated" ]; then
             error_msg="${REVIEW_FAILURE_REASON:-The selected Codex refresh token is invalid.}"
@@ -1122,7 +1135,7 @@ jobs:
           else
             error_msg="${REVIEW_FAILURE_REASON:-Review step was $REVIEW_OUTCOME (possibly timeout or cancelled)}"
           fi
-          python3 "$REVIEW_GH_HELPER" pr comment "$PR_NUMBER" --body "$(cat <<EOF
+          python3 "$REVIEW_GH_HELPER" pr comment "$PR_NUMBER" --repo "$REPO" --body "$(cat <<EOF
           Codex automated review failed and did not complete.
 
           Error: ${error_msg}
@@ -1136,12 +1149,15 @@ jobs:
         if: ${{ always() && (steps.review_context.outcome != 'success' || steps.review.outcome != 'success') }}
         timeout-minutes: 1
         env:
+          REVIEW_INPUTS_OUTCOME: ${{ steps.review_inputs.outcome }}
           AUTH_FAILURE_REASON: ${{ steps.auth.outputs.failure_reason }}
           REVIEW_CONTEXT_OUTCOME: ${{ steps.review_context.outcome }}
           REVIEW_FAILURE_REASON: ${{ steps.review.outputs.failure_reason }}
           REVIEW_OUTCOME: ${{ steps.review.outcome }}
         run: |
-          if [ -n "$AUTH_FAILURE_REASON" ]; then
+          if [ "$REVIEW_INPUTS_OUTCOME" = "failure" ] || [ "$REVIEW_INPUTS_OUTCOME" = "cancelled" ]; then
+            error_msg="Step 'Resolve review inputs' ended with $REVIEW_INPUTS_OUTCOME; inspect its logs for the original error."
+          elif [ -n "$AUTH_FAILURE_REASON" ]; then
             error_msg="$AUTH_FAILURE_REASON"
           elif [ "$REVIEW_CONTEXT_OUTCOME" != "success" ]; then
             error_msg="Review context preparation failed before Codex ran; inspect the 'Prepare authoritative PR context and required AGENTS guides' step."