This is an automated email from the ASF dual-hosted git repository.
hello-stephen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 05a280e0860 [fix](ci) Stop superseded reviews when the PR head changes
(#68981)
05a280e0860 is described below
commit 05a280e08601ad4707a2e564b66b3061485e4b95
Author: shuke <[email protected]>
AuthorDate: Tue Oct 13 11:44:24 2026 +0800
[fix](ci) Stop superseded reviews when the PR head changes (#68981)
When a PR receives another commit, its existing AI review can keep
consuming model capacity until Codex notices the stale head. It then
fails final submission verification with `No final review submission was
declared`. This also happens while the outer runner waits five minutes
to resume after a capacity error.
Poll the authoritative PR head every 60 seconds during execution and
capacity backoff, and check at attempt boundaries. A confirmed different
SHA stops the old attempt through the existing
SIGINT/grace-period/descendant-reaping path. API errors, timeouts, and
malformed responses only warn; they do not establish supersession.
Existing strict resume and submission checks remain in place.
Record `review.superseded`, both SHAs, and a final diagnostic. The
workflow validates the distinct helper exit code/result, treats
supersession as a handled outcome, skips verdict/status publication and
obsolete status aggregation, and retains Litefuse, session, and auth
finalization. A successful workflow in this case does **not** mean a
passed review. Litefuse verification also permits a review stopped
before its first model/tool request. New commits still require the
existing `/review` trigger; this does not add automatic reviews on
pushes.
---
.github/scripts/emit_litefuse_otel_io.py | 8 +
.github/scripts/run_review_with_resume.py | 110 ++++++++++-
.github/scripts/test_emit_litefuse_otel_io.py | 17 ++
.github/scripts/test_run_review_with_resume.py | 255 ++++++++++++++++++++++++-
.github/scripts/test_submit_review.py | 3 +-
.github/workflows/code-review-runner.yml | 34 +++-
6 files changed, 409 insertions(+), 18 deletions(-)
diff --git a/.github/scripts/emit_litefuse_otel_io.py
b/.github/scripts/emit_litefuse_otel_io.py
index 4da5db39950..3c6cf3846cb 100644
--- a/.github/scripts/emit_litefuse_otel_io.py
+++ b/.github/scripts/emit_litefuse_otel_io.py
@@ -389,6 +389,14 @@ def build_ingestion_payload(args, input_text, output_text,
events):
if key not in ("type", "_line_number")
}
+ superseded = next((event for event in reversed(events)
+ if event.get("type") == "review.superseded"), None)
+ if superseded is not None:
+ trace_metadata["review_superseded"] = {
+ key: value for key, value in superseded.items()
+ if key not in ("type", "_line_number")
+ }
+
completed_items = [
event
for event in events
diff --git a/.github/scripts/run_review_with_resume.py
b/.github/scripts/run_review_with_resume.py
old mode 100755
new mode 100644
index 43fe87a4079..b0dfd51ee36
--- a/.github/scripts/run_review_with_resume.py
+++ b/.github/scripts/run_review_with_resume.py
@@ -22,6 +22,7 @@ import argparse
import ctypes
import json
import os
+import re
import shutil
import signal
import subprocess
@@ -39,6 +40,65 @@ RETRY_DELAYS = (300,) * 6
CAPACITY_MESSAGE = "Selected model is at capacity. Please try a different
model."
UNSUPPORTED_CHATGPT_MODEL = "model is not supported when using Codex with a
ChatGPT account."
PROCESS_EXIT_GRACE_SECONDS = 5
+HEAD_POLL_SECONDS = 60
+HEAD_API_TIMEOUT_SECONDS = 10
+SUPERSEDED_EXIT_CODE = 75
+
+
+class ReviewSuperseded(Exception):
+ def __init__(self, head_sha):
+ self.head_sha = head_sha
+ super().__init__(f"PR head changed to {head_sha}; stopping superseded
review")
+
+
+def check_head(pr, expected):
+ """Only a complete, valid API head can supersede the frozen review
target."""
+ head = pr.get("head") if isinstance(pr, dict) else None
+ sha = head.get("sha") if isinstance(head, dict) else None
+ if not isinstance(sha, str) or not re.fullmatch(r"[0-9a-f]{40}", sha):
+ raise ValueError("PR API returned no valid head SHA")
+ if sha != expected:
+ raise ReviewSuperseded(sha)
+
+
+class HeadWatchdog:
+ def __init__(self, args, remaining):
+ self.args = args
+ self.remaining = remaining
+ self.next_check = 0
+
+ def seconds_until_check(self):
+ return max(0, self.next_check - time.monotonic())
+
+ def check(self, *, force=False):
+ if not force and self.seconds_until_check() > 0:
+ return
+ timeout = min(HEAD_API_TIMEOUT_SECONDS, self.remaining())
+ # Poll synchronously: gh is reaped before Codex cleanup/ChildReaper
runs.
+ # Include API time in the existing shared budget and polling interval.
+ self.next_check = time.monotonic() + HEAD_POLL_SECONDS
+ try:
+ response = subprocess.run(
+ ["gh", "api",
f"repos/{self.args.repository}/pulls/{self.args.pr_number}"],
+ check=True, capture_output=True, text=True, timeout=timeout,
+ )
+ check_head(json.loads(response.stdout), self.args.head_sha)
+ except (ValueError, OSError, subprocess.SubprocessError) as exc:
+ # An unavailable API is not evidence of a new commit. Strict target
+ # checks before resume/submission still apply independently.
+ print(
+ f"PR head check unavailable ({type(exc).__name__}); keeping
current review",
+ file=sys.stderr, flush=True,
+ )
+
+ def wait(self, seconds):
+ deadline = time.monotonic() + seconds
+ while True:
+ self.check()
+ delay = min(deadline - time.monotonic(), self.remaining())
+ if delay <= 0:
+ return
+ time.sleep(min(delay, self.seconds_until_check()))
class ChildReaper:
@@ -77,7 +137,7 @@ class ChildReaper:
def reap(self):
# Called only after Popen.wait() reaps Codex. This dedicated helper has
- # no other concurrent subprocesses; gh/help commands run between
attempts.
+ # no other concurrent subprocesses; synchronous gh/help calls have
returned.
deadline = time.monotonic() + PROCESS_EXIT_GRACE_SECONDS
while children := self.children.read_text().split():
if time.monotonic() >= deadline:
@@ -204,7 +264,7 @@ def stop_process(process):
process.wait()
-def run_attempt(command, events_path, stderr_path, timeout, reaper=None):
+def run_attempt(command, events_path, stderr_path, timeout, reaper=None,
watchdog=None):
with events_path.open("w") as stdout, stderr_path.open("w") as stderr:
process = None
copier = None
@@ -237,7 +297,19 @@ def run_attempt(command, events_path, stderr_path,
timeout, reaper=None):
copier.start()
finally:
signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
- return process.wait(timeout=timeout)
+ if watchdog is None:
+ return process.wait(timeout=timeout)
+ deadline = time.monotonic() + timeout
+ while True:
+ watchdog.check()
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise subprocess.TimeoutExpired(command, timeout)
+ try:
+ return process.wait(timeout=min(remaining,
watchdog.seconds_until_check()))
+ except subprocess.TimeoutExpired:
+ if time.monotonic() >= deadline:
+ raise
finally:
# Finish bounded cleanup before delivering cancellation, including
a
# second signal while an earlier cancellation is already unwinding.
@@ -298,9 +370,9 @@ def check_resume_target(args, started_at, remaining):
return json.loads(result.stdout)
pr = api(f"repos/{args.repository}/pulls/{args.pr_number}")
+ check_head(pr, args.head_sha)
if (
pr["state"] != "open"
- or pr["head"]["sha"] != args.head_sha
or pr["base"]["sha"] != args.base_sha
):
raise ValueError(
@@ -388,13 +460,15 @@ def run_review(args, reaper=None):
)
return 1
+ watchdog = HeadWatchdog(args, remaining)
try:
while capacity_retry <= len(RETRY_DELAYS):
+ (context / "codex-review-model.txt").write_text(model + "\n")
+ watchdog.check(force=True)
attempt_number += 1
events_path = attempts / f"{attempt_number}.jsonl"
stderr_path = attempts / f"{attempt_number}.stderr.log"
output_path = attempts / f"{attempt_number}.final.txt"
- (context / "codex-review-model.txt").write_text(model + "\n")
final_message.write_text("")
command = [
"codex",
@@ -426,7 +500,7 @@ def run_review(args, reaper=None):
)
try:
status = run_attempt(
- command, events_path, stderr_path, remaining(),
reaper=reaper
+ command, events_path, stderr_path, remaining(),
reaper=reaper, watchdog=watchdog
)
finally:
# Preserve failed attempts and their child-thread IDs for
Litefuse.
@@ -439,6 +513,9 @@ def run_review(args, reaper=None):
if output_path.exists():
shutil.copyfile(output_path, final_message)
+ # Catch a head change even if Codex exited before the next poll,
+ # including its own stale-head refusal without a final submission.
+ watchdog.check(force=True)
events = read_events(events_path)
if status < 0 or status in (124, 130, 137, 143):
return fail(
@@ -478,7 +555,12 @@ def run_review(args, reaper=None):
# Final submission declares all review work complete. Only an
# exact GitHub readback can recover a subsequent capacity
error;
# another run's review or a partially posted review cannot
pass.
- result = verify_completion(context, run, remaining)
+ try:
+ result = verify_completion(context, run, remaining)
+ except ValueError:
+ # The head can move between the exit check and readback.
+ watchdog.check(force=True)
+ raise
result["recovered_after_capacity"] = message ==
CAPACITY_MESSAGE
write_json(context / RESULT_FILE, result)
with aggregate.open("a") as handle:
@@ -520,9 +602,21 @@ def run_review(args, reaper=None):
)
+ "\n"
)
- time.sleep(delay)
+ watchdog.wait(delay)
check_resume_target(args, started_at, remaining)
capacity_retry += 1
+ except ReviewSuperseded as exc:
+ result = {
+ "state": "superseded", "repository": args.repository,
+ "pr_number": args.pr_number, "head_sha": args.head_sha,
+ "superseding_head_sha": exc.head_sha, "message": str(exc),
+ }
+ write_json(context / RESULT_FILE, result)
+ with aggregate.open("a") as handle:
+ handle.write(json.dumps({"type": "review.superseded", **result}) +
"\n")
+ final_message.write_text(str(exc) + "\n")
+ print(str(exc), file=sys.stderr, flush=True)
+ return SUPERSEDED_EXIT_CODE
except KeyboardInterrupt:
fail("Review cancelled; not resuming")
return 130
diff --git a/.github/scripts/test_emit_litefuse_otel_io.py
b/.github/scripts/test_emit_litefuse_otel_io.py
index 866b3c09ead..6c7b31960ac 100644
--- a/.github/scripts/test_emit_litefuse_otel_io.py
+++ b/.github/scripts/test_emit_litefuse_otel_io.py
@@ -192,6 +192,23 @@ class LitefuseOtelExporterTest(unittest.TestCase):
self.assertEqual("failed", turn["output"]["status"])
self.assertEqual("ERROR", turn["level"])
+ def test_superseded_outcome_is_distinct_from_verified_completion(self):
+ args = SimpleNamespace(
+ repository="apache/doris", workflow="Code Review", run_id="123",
pr_number="123",
+ head_sha="a" * 40, base_sha="b" * 40, reasoning_effort="xhigh",
+ max_json_chars=20000, max_context_json_chars=0,
trace_name="review",
+ session_id="123", environment="test", model="gpt-6-sol",
+ )
+ outcome = {"state": "superseded", "head_sha": "a" * 40,
"superseding_head_sha": "c" * 40}
+ _, payload, _ = MODULE.build_ingestion_payload(args, "review",
"Superseded by new head", [
+ {"type": "turn.failed", "error": {"message": "capacity"}},
+ {"type": "review.superseded", **outcome},
+ ])
+ trace = payload["batch"][0]["body"]
+ self.assertEqual(outcome, trace["metadata"]["review_superseded"])
+ self.assertNotIn("review_completion", trace["metadata"])
+ self.assertIn("Superseded by new head", trace["output"])
+
def test_bounds_and_shrinks_failed_turn_status_message(self):
args = SimpleNamespace(
repository="apache/doris",
diff --git a/.github/scripts/test_run_review_with_resume.py
b/.github/scripts/test_run_review_with_resume.py
old mode 100755
new mode 100644
index ea06e64284e..e58d35d36c7
--- a/.github/scripts/test_run_review_with_resume.py
+++ b/.github/scripts/test_run_review_with_resume.py
@@ -20,6 +20,7 @@ import io
import json
import os
import signal
+import shutil
import subprocess
import sys
import tempfile
@@ -51,6 +52,19 @@ def completed():
return {"type": "turn.completed", "usage": {"input_tokens": 20,
"output_tokens": 3}}
+def write_fake_gh(root):
+ # CLI lifecycle tests must never use real GitHub credentials or the
network.
+ gh = root / "gh"
+ gh.write_text(
+ f"#!{sys.executable}\n"
+ "import json\nfrom pathlib import Path\n"
+ f"head_file = Path({str(root / 'new-head')!r})\n"
+ f"head = head_file.read_text() if head_file.exists() else {'a' *
40!r}\n"
+ "print(json.dumps({'head': {'sha': head}}))\n"
+ )
+ gh.chmod(0o700)
+
+
class ResumeReviewTest(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
@@ -98,6 +112,8 @@ class ResumeReviewTest(unittest.TestCase):
self.target_check = self.enterContext(
mock.patch.object(runner, "check_resume_target")
)
+ self.watchdog = self.enterContext(mock.patch.object(runner,
"HeadWatchdog")).return_value
+ self.watchdog.wait.side_effect = lambda seconds:
runner.time.sleep(seconds)
self.help = self.enterContext(mock.patch.object(runner.subprocess,
"run"))
self.verification = self.enterContext(mock.patch.object(
runner, "verify_completion", return_value={
@@ -123,7 +139,7 @@ class ResumeReviewTest(unittest.TestCase):
def execute(self, attempts):
pending = iter(attempts)
- def fake_attempt(command, events_path, stderr_path, timeout,
reaper=None):
+ def fake_attempt(command, events_path, stderr_path, timeout,
reaper=None, watchdog=None):
self.commands.append(command)
self.timeouts.append(timeout)
spec = next(pending)
@@ -152,6 +168,62 @@ class ResumeReviewTest(unittest.TestCase):
def last_error(self):
return self.events()[-1]["error"]["message"]
+ def assert_superseded(self):
+ result = json.loads((self.context / runner.RESULT_FILE).read_text())
+ self.assertEqual("superseded", result["state"])
+ self.assertEqual("c" * 40, result["superseding_head_sha"])
+ self.assertEqual(self.args.head_sha, result["head_sha"])
+ self.assertEqual("review.superseded", self.events()[-1]["type"])
+ self.assertIn("superseded", (self.context /
"codex-final-message.txt").read_text())
+
+ def test_stale_before_start_does_not_launch_codex(self):
+ self.watchdog.check.side_effect = runner.ReviewSuperseded("c" * 40)
+ self.assertEqual(runner.SUPERSEDED_EXIT_CODE, self.execute([]))
+ self.assertEqual([], self.commands)
+ self.assertEqual(self.args.model, (self.context /
"codex-review-model.txt").read_text().strip())
+ self.verification.assert_not_called()
+ self.assert_superseded()
+
+ def
test_superseded_attempt_preserves_logs_without_verification_or_resume(self):
+ self.assertEqual(runner.SUPERSEDED_EXIT_CODE, self.execute([{
+ "events": [thread_event()], "stderr": "attempt diagnostic",
+ "output": "interrupted partial review",
+ "raises": runner.ReviewSuperseded("c" * 40),
+ }]))
+ self.assertEqual(thread_event(), self.events()[0])
+ self.assertIn("attempt diagnostic", (self.context /
"codex-stderr.log").read_text())
+ self.assertEqual("interrupted partial review",
+ (self.context /
"codex-attempts/1.final.txt").read_text())
+ self.verification.assert_not_called()
+ self.target_check.assert_not_called()
+ self.assertEqual([], self.sleeps)
+ self.assert_superseded()
+
+ def test_stale_at_exit_does_not_report_missing_submission(self):
+ self.watchdog.check.side_effect = [None, runner.ReviewSuperseded("c" *
40)]
+ self.assertEqual(runner.SUPERSEDED_EXIT_CODE, self.execute([
+ {"events": [thread_event(), completed()], "status": 0},
+ ]))
+ self.verification.assert_not_called()
+ self.assert_superseded()
+
+ def test_stale_during_verification_is_superseded(self):
+ self.verification.side_effect = ValueError("PR head moved during
readback")
+ self.watchdog.check.side_effect = [None, None,
runner.ReviewSuperseded("c" * 40)]
+ self.assertEqual(runner.SUPERSEDED_EXIT_CODE, self.execute([
+ {"events": [thread_event(), completed()], "status": 0},
+ ]))
+ self.assert_superseded()
+
+ def test_superseded_in_backoff_never_resumes(self):
+ self.watchdog.wait.side_effect = runner.ReviewSuperseded("c" * 40)
+ self.assertEqual(runner.SUPERSEDED_EXIT_CODE, self.execute([
+ {"events": [thread_event(), failed()]},
+ ]))
+ self.assertEqual(1, len(self.commands))
+ self.target_check.assert_not_called()
+ self.assert_superseded()
+
def test_success_does_not_retry(self):
self.assertEqual(
0, self.execute([{"events": [thread_event(), completed()],
"status": 0}])
@@ -733,6 +805,162 @@ class ResumeReviewTest(unittest.TestCase):
)
+
+class HeadWatchdogTest(unittest.TestCase):
+ def setUp(self):
+ self.clock = 0
+ self.args = SimpleNamespace(repository="apache/doris",
pr_number="123", head_sha="a" * 40)
+ self.enterContext(mock.patch.object(runner.time, "monotonic",
side_effect=lambda: self.clock))
+ self.sleep = self.enterContext(mock.patch.object(runner.time, "sleep",
side_effect=self.advance))
+ self.api = self.enterContext(mock.patch.object(runner.subprocess,
"run"))
+ self.api.return_value = SimpleNamespace(stdout=json.dumps({"head":
{"sha": "a" * 40}}))
+ self.stderr = self.enterContext(redirect_stderr(io.StringIO()))
+ self.watchdog = runner.HeadWatchdog(self.args, lambda: 1000 -
self.clock)
+
+ def advance(self, seconds):
+ self.clock += seconds
+
+ def test_unchanged_head_is_polled_once_per_minute(self):
+ self.watchdog.check()
+ self.clock = 59
+ self.watchdog.check()
+ self.assertEqual(1, self.api.call_count)
+ self.clock = 60
+ self.watchdog.check()
+ self.assertEqual(2, self.api.call_count)
+ self.assertEqual(10, self.api.call_args.kwargs["timeout"])
+
+ def test_only_head_changes_cancel_not_base_or_open_state(self):
+ self.api.return_value.stdout = json.dumps({
+ "head": {"sha": "a" * 40}, "base": {"sha": "c" * 40}, "state":
"closed",
+ })
+ self.watchdog.check()
+ self.api.return_value.stdout = json.dumps({"head": {"sha": "c" * 40}})
+ with self.assertRaises(runner.ReviewSuperseded) as raised:
+ self.watchdog.check(force=True)
+ self.assertEqual("c" * 40, raised.exception.head_sha)
+
+ def test_invalid_response_does_not_cancel(self):
+ for payload in ("not json", "null", "[]", "{}", '{"head":null}',
+ '{"head":{"sha":null}}', '{"head":{"sha":"short"}}',
+ '{"head":{"sha":123}}'):
+ with self.subTest(payload=payload):
+ self.api.return_value.stdout = payload
+ self.watchdog.check(force=True)
+ self.assertIn("head check unavailable", self.stderr.getvalue())
+
+ def test_api_failures_keep_review_and_allow_later_detection(self):
+ for error in (subprocess.TimeoutExpired("gh", 10),
+ subprocess.CalledProcessError(1, "gh"),
OSError("unavailable")):
+ with self.subTest(error=type(error)):
+ self.api.side_effect = error
+ self.watchdog.check(force=True)
+ self.api.side_effect = None
+ self.api.return_value.stdout = json.dumps({"head": {"sha": "c" * 40}})
+ with self.assertRaises(runner.ReviewSuperseded):
+ self.watchdog.check(force=True)
+
+ def test_cancellation_is_not_swallowed_as_api_failure(self):
+ self.api.side_effect = KeyboardInterrupt
+ with self.assertRaises(KeyboardInterrupt):
+ self.watchdog.check()
+
+ def test_api_timeout_is_bounded_by_remaining_budget(self):
+ self.clock = 998
+ self.watchdog.check()
+ self.assertEqual(2, self.api.call_args.kwargs["timeout"])
+
+ def test_backoff_polls_and_stops_before_another_model_request(self):
+ self.api.side_effect = [
+ SimpleNamespace(stdout=json.dumps({"head": {"sha": "a" * 40}})),
+ SimpleNamespace(stdout=json.dumps({"head": {"sha": "c" * 40}})),
+ ]
+ with self.assertRaises(runner.ReviewSuperseded):
+ self.watchdog.wait(300)
+ self.assertEqual(60, self.clock)
+
+ def test_backoff_api_time_counts_toward_wait(self):
+ def slow_api(*args, **kwargs):
+ self.clock += 5
+ return SimpleNamespace(stdout=json.dumps({"head": {"sha": "a" *
40}}))
+ self.api.side_effect = slow_api
+ self.watchdog.wait(125)
+ self.assertEqual(125, self.clock)
+ self.assertEqual([mock.call(55), mock.call(55)],
self.sleep.call_args_list)
+
+ def test_running_process_keeps_shared_deadline_despite_poll_time(self):
+ self.watchdog.remaining = lambda: 65 - self.clock
+ process = mock.Mock(stderr=io.StringIO())
+ def wait(timeout=None):
+ if timeout is None or timeout == runner.PROCESS_EXIT_GRACE_SECONDS:
+ return 0 # cleanup
+ self.clock += timeout
+ raise subprocess.TimeoutExpired("codex", timeout)
+ process.wait.side_effect = wait
+ def slow_api(*args, **kwargs):
+ self.clock += min(10, kwargs["timeout"])
+ raise subprocess.TimeoutExpired("gh", kwargs["timeout"])
+ self.api.side_effect = slow_api
+ reaper = mock.Mock()
+ with tempfile.TemporaryDirectory() as tmp,
mock.patch.object(runner.subprocess, "Popen", return_value=process):
+ with self.assertRaises(subprocess.TimeoutExpired):
+ runner.run_attempt(["codex"], Path(tmp) / "events", Path(tmp)
/ "stderr", 65,
+ reaper=reaper, watchdog=self.watchdog)
+ self.assertEqual(65, self.clock)
+ process.send_signal.assert_called_once_with(signal.SIGINT)
+ reaper.reap.assert_called_once()
+
+
[email protected](shutil.which("jq"), "workflow shell test requires jq")
+class SupersededWorkflowTest(unittest.TestCase):
+ def test_shell_accepts_only_matching_superseded_result(self):
+ workflow = (Path(__file__).resolve().parents[1] /
"workflows/code-review-runner.yml").read_text()
+ step = workflow.split(" - name: Run automated code review\n",
1)[1].split("\n - name:", 1)[0]
+ script = textwrap.dedent(step.split(" set -e\n\n", 1)[1])
+ for case, status, state, head, expected in (
+ ("superseded", 75, "superseded", "a" * 40, 0),
+ ("wrong head", 75, "superseded", "b" * 40, 1),
+ ("wrong state", 75, "success", "a" * 40, 1),
+ ("generic failure", 1, "superseded", "a" * 40, 1),
+ ):
+ with self.subTest(case=case), tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ (root / "review-result.json").write_text(json.dumps({
+ "state": state, "repository": "apache/doris", "pr_number":
"123",
+ "head_sha": head, "superseding_head_sha": "c" * 40,
+ "message": "PR head changed; stopping superseded review",
+ }))
+ (root / "codex-events.jsonl").touch()
+ (root / "codex-stderr.log").touch()
+ result = subprocess.run(["bash", "-e", "-c", script],
capture_output=True, text=True, env={
+ **os.environ, "status": str(status), "REPO":
"apache/doris", "PR_NUMBER": "123",
+ "HEAD_SHA": "a" * 40, "REVIEW_CONTEXT_DIR": tmp,
+ "GITHUB_OUTPUT": str(root / "output"),
"GITHUB_STEP_SUMMARY": str(root / "summary"),
+ })
+ self.assertEqual(expected, result.returncode, result.stderr)
+ output = (root / "output").read_text() if (root /
"output").exists() else ""
+ if expected == 0:
+ self.assertEqual("review_state=superseded\n", output)
+ self.assertIn("superseded", (root / "summary").read_text())
+ else:
+ self.assertNotIn("review_state=superseded", output)
+
+
+class SupersededCleanupTest(unittest.TestCase):
+ def test_cleanup_failure_does_not_become_handled_supersession(self):
+ process = mock.Mock(stderr=io.StringIO())
+ watchdog = mock.Mock()
+ watchdog.check.side_effect = runner.ReviewSuperseded("c" * 40)
+ reaper = mock.Mock()
+ reaper.reap.side_effect = OSError("descendant cleanup failed")
+ with tempfile.TemporaryDirectory() as tmp,
mock.patch.object(runner.subprocess, "Popen", return_value=process):
+ with self.assertRaisesRegex(OSError, "cleanup failed"):
+ runner.run_attempt(["codex"], Path(tmp) / "events", Path(tmp)
/ "stderr", 60,
+ reaper=reaper, watchdog=watchdog)
+ process.send_signal.assert_called_once_with(signal.SIGINT)
+ process.kill.assert_called_once()
+
+
class WorkflowBudgetTest(unittest.TestCase):
def test_cli_requires_the_workflow_budget(self):
with (
@@ -872,6 +1100,11 @@ class ResumeTargetTest(unittest.TestCase):
with self.assertRaisesRegex(ValueError, "changed"):
self.check([[]])
+ def test_new_head_before_resume_has_explicit_outcome(self):
+ self.pr["head"]["sha"] = "c" * 40
+ with self.assertRaises(runner.ReviewSuperseded):
+ self.check([[]])
+
def test_unavailable_api_does_not_blindly_resume(self):
with (
mock.patch.object(
@@ -1067,6 +1300,7 @@ class ProcessLifecycleTest(unittest.TestCase):
def test_real_process_capacity_then_resume_preserves_state(self):
with tempfile.TemporaryDirectory() as tmp,
redirect_stderr(io.StringIO()):
root = Path(tmp)
+ write_fake_gh(root)
(root / "codex_goal_prompt.txt").write_text("test review")
ledger = root / "subagent_review_findings.md"
ledger.write_text("completed round 1")
@@ -1173,6 +1407,7 @@ else:
def test_cli_sigterm_cleans_up_child_and_does_not_restart(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
+ write_fake_gh(root)
(root / "codex_goal_prompt.txt").write_text(
"test only; do not call a service"
)
@@ -1243,7 +1478,7 @@ else:
@unittest.skipUnless(sys.platform == "linux", "requires Linux
subreaper/pidfd")
def
test_cli_reaps_detached_descendants_without_touching_other_processes(self):
- for stop in ("cancel", "timeout", "exit"):
+ for stop in ("cancel", "timeout", "exit", "superseded"):
for graceful in (True, False):
with self.subTest(stop=stop, graceful=graceful):
self.check_detached_descendants(stop, graceful)
@@ -1257,6 +1492,7 @@ else:
def check_detached_descendants(self, stop, graceful):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
+ write_fake_gh(root)
(root / "codex_goal_prompt.txt").write_text("local process test
only")
fake_codex = root / "codex"
fake_codex.write_text(
@@ -1301,7 +1537,7 @@ while time.monotonic() < deadline:
(
"import sys; sys.path.insert(0, sys.argv.pop(1)); "
"import run_review_with_resume as r; "
- "r.PROCESS_EXIT_GRACE_SECONDS = 0.5; "
+ "r.PROCESS_EXIT_GRACE_SECONDS = 0.5; r.HEAD_POLL_SECONDS =
0.05; "
+ (
"""
import os, signal
@@ -1370,11 +1606,14 @@ r.os.pidfd_open = interrupt_cleanup
self.assertTrue(
(root / "pids.json").exists(), "fake Codex did not
start"
)
+ if stop == "superseded":
+ (root / "new-head").write_text("c" * 40)
if stop.startswith("cancel"):
process.send_signal(signal.SIGTERM)
_, stderr = process.communicate(timeout=10)
self.assertEqual(
- 130 if "cancel" in stop else 1, process.returncode,
stderr
+ 75 if stop == "superseded" else 130 if "cancel" in
stop else 1,
+ process.returncode, stderr
)
for pid in json.loads((root / "pids.json").read_text()):
with self.assertRaises(ProcessLookupError):
@@ -1391,7 +1630,13 @@ r.os.pidfd_open = interrupt_cleanup
"Review cancelled; not resuming",
events[-1]["error"]["message"],
)
- if stop in ("cancel", "timeout", "cancel_reap_repeat"):
+ if stop == "superseded":
+ result = json.loads((root /
runner.RESULT_FILE).read_text())
+ self.assertEqual("superseded", result["state"])
+ self.assertEqual("c" * 40,
result["superseding_head_sha"])
+ self.assertEqual("review.superseded",
+ runner.read_events(root /
"codex-events.jsonl")[-1]["type"])
+ if stop in ("cancel", "timeout", "cancel_reap_repeat",
"superseded"):
self.assertEqual(graceful, (root /
"interrupted").exists())
finally:
if process.poll() is None:
diff --git a/.github/scripts/test_submit_review.py
b/.github/scripts/test_submit_review.py
index f40ad82d05b..702f2ceb558 100644
--- a/.github/scripts/test_submit_review.py
+++ b/.github/scripts/test_submit_review.py
@@ -247,7 +247,7 @@ class FinalReviewTest(unittest.TestCase):
}, model="gpt-6-sol", effort="xhigh", budget_seconds=60)
self.reviews.clear()
- def attempt(command, events, stderr, timeout, reaper=None):
+ def attempt(command, events, stderr, timeout, reaper=None,
watchdog=None):
run = json.loads((context /
submitter.RUN_FILE).read_text())
submitter.verify_completion(context, run, self.remaining,
submit=True,
submission=self.submission(level))
@@ -259,6 +259,7 @@ class FinalReviewTest(unittest.TestCase):
with mock.patch.object(runner, "run_attempt",
side_effect=attempt), \
mock.patch.object(runner, "RETRY_DELAYS", ()), \
+ mock.patch.object(runner, "HeadWatchdog"), \
mock.patch.object(runner, "check_resume_target") as
guard, \
mock.patch.object(runner.time, "sleep") as sleep,
redirect_stderr(io.StringIO()):
self.assertEqual(0, runner.run_review(args))
diff --git a/.github/workflows/code-review-runner.yml
b/.github/workflows/code-review-runner.yml
index 51b84852fb6..23ffcefa2c9 100644
--- a/.github/workflows/code-review-runner.yml
+++ b/.github/workflows/code-review-runner.yml
@@ -205,6 +205,7 @@ jobs:
base_sha: ${{ steps.review_inputs.outputs.base_sha }}
head_sha: ${{ steps.review_inputs.outputs.head_sha }}
manage_status: ${{ steps.review_inputs.outputs.manage_status }}
+ review_state: ${{ steps.review.outputs.review_state }}
pr_number: ${{ steps.review_inputs.outputs.pr_number }}
# Pre-finalization steps can use 218 minutes and auth sync can use 8 more,
# leaving 12 minutes for runner setup and post-job cleanup.
@@ -895,6 +896,20 @@ jobs:
status=$?
set -e
+ # The runner has stopped Codex and reaped descendants. Treat an
+ # obsolete head as a handled outcome, without publishing a verdict.
+ if [ "$status" -eq 75 ]; then
+ jq -e --arg repo "$REPO" --arg pr "$PR_NUMBER" --arg head
"$HEAD_SHA" '
+ .state == "superseded" and .repository == $repo and
+ (.pr_number | tostring) == $pr and .head_sha == $head and
+ (.superseding_head_sha | test("^[0-9a-f]{40}$")) and
+ .superseding_head_sha != $head' \
+ "$REVIEW_CONTEXT_DIR/review-result.json" > /dev/null
+ echo "review_state=superseded" >> "$GITHUB_OUTPUT"
+ jq -r '.message' "$REVIEW_CONTEXT_DIR/review-result.json" | tee -a
"$GITHUB_STEP_SUMMARY"
+ exit 0
+ fi
+
failure_reason=""
if [ "$status" -ne 0 ]; then
if [ -s "$REVIEW_CONTEXT_DIR/codex-events.jsonl" ]; then
@@ -1080,7 +1095,8 @@ jobs:
${{
always() &&
steps.review_inputs.outcome == 'success' &&
- steps.review_inputs.outputs.manage_status == 'true'
+ steps.review_inputs.outputs.manage_status == 'true' &&
+ steps.review.outputs.review_state != 'superseded'
}}
timeout-minutes: 2
env:
@@ -1127,6 +1143,7 @@ jobs:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PK }}
LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SK }}
+ REVIEW_STATE: ${{ steps.review.outputs.review_state }}
REPO: ${{ github.repository }}
PR_NUMBER: ${{ steps.review_inputs.outputs.pr_number }}
HEAD_SHA: ${{ steps.review_inputs.outputs.head_sha }}
@@ -1145,6 +1162,14 @@ jobs:
> "$helper"
chmod 700 "$helper"
+ min_observations=4
+ min_step_observations=2
+ if [ "$REVIEW_STATE" = "superseded" ]; then
+ # The head may have changed before the first model/tool request.
+ min_observations=1
+ min_step_observations=0
+ fi
+
python3 "$helper" \
--input-file "$REVIEW_CONTEXT_DIR/codex_goal_prompt.txt" \
--events-file "$REVIEW_CONTEXT_DIR/codex-events.jsonl" \
@@ -1163,8 +1188,8 @@ jobs:
--reasoning-effort "xhigh" \
--environment "github-actions" \
--max-payload-bytes 4000000 \
- --min-observations 4 \
- --min-step-observations 2 \
+ --min-observations "$min_observations" \
+ --min-step-observations "$min_step_observations" \
--verify-attempts 24 \
--verify-sleep-seconds 5 \
--verify-timeout-seconds 120 \
@@ -1255,7 +1280,8 @@ jobs:
if: >-
always() &&
needs.code-review.outputs.manage_status == 'true' &&
- needs.code-review.outputs.head_sha != ''
+ needs.code-review.outputs.head_sha != '' &&
+ needs.code-review.outputs.review_state != 'superseded'
uses: ./.github/workflows/code-review-aggregate-status.yml
with:
head_sha: ${{ needs.code-review.outputs.head_sha }}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
