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 d2e18b9425d [fix](ci) Resume automated reviews after transient Codex
capacity failures (#67624)
d2e18b9425d is described below
commit d2e18b9425dee17f737122b85796dba9eb8fb684
Author: shuke <[email protected]>
AuthorDate: Mon Sep 21 15:59:37 2026 +0800
[fix](ci) Resume automated reviews after transient Codex capacity failures
(#67624)
### What problem does this PR solve?
Problem Summary:
A transient Codex capacity error currently ends the entire automated
review:
`Selected model is at capacity. Please try a different model.`
This PR adds bounded recovery of the persisted main goal session,
retaining the checkout, CODEX_HOME and shared review ledger.
- Resume the exact main UUID with `codex exec --goal ... resume <UUID>
<recovery objective>`. Require a matching persisted rollout and CLI
argument support.
- Allow at most three resumes with 30 / 60 / 120 second backoff.
Authentication, usage limits, unknown errors, cancellation and timeout
are not retried.
- Use one `REVIEW_TIMEOUT_MINUTES: 120` setting for the review step.
Deduct elapsed helper-download time and a two-minute
cleanup/verification reserve, then pass the remaining seconds explicitly
to the helper. All attempts, backoffs and pre-resume checks share one
monotonic deadline; the old 89-minute default is removed.
- Before resuming, verify that the PR remains open at the expected
base/head and that this run has not already submitted a bot review for
that head. Recovery instructions preserve completed work and
review-round limits, account for interrupted subagents and require fresh
GitHub checks before writes.
- Finish process shutdown, Linux orphan reaping and stderr cleanup
before delivering pending SIGINT/SIGTERM. Repeated cancellation cannot
interrupt this cleanup; cancellation still exits with status 130 without
another attempt. pidfd ownership checks avoid signalling unrelated
processes.
- Preserve raw per-attempt output and merge parseable events for
Litefuse, including failed-attempt items and child IDs. Clear stale
final messages between attempts.
The helper is loaded from the workflow's pinned ref. This branch
includes current master workflow changes, retaining authentication
quarantine and final GitHub review verification.
### Validation
- Linux Python 3.12: `python -m unittest discover -s .github/scripts -p
'test_*.py'` — **135 tests passed, no skips**, including 45
recovery-helper tests.
- Actual Linux process tests cover timeout, detached descendants,
graceful/forced shutdown, cancellation during reaping, repeated mixed
SIGINT/SIGTERM and repeated cancellation after an initial interrupt.
They verify no surviving owned descendants, no retry and no effect on
unrelated processes.
- The three new cancellation scenarios fail against the previous helper
because descendants survive, and pass with this fix.
- Budget tests execute the workflow shell with a fake download, checking
elapsed setup deduction, changed total budget and exhaustion. A
simulated 95-minute successful review verifies removal of the old limit;
retry tests verify that the deadline is not reset.
- Existing authentication-quarantine tests now run through the real
helper with fake Codex/GitHub/OSS endpoints.
- macOS suite completed with Linux-dependent cases skipped.
- actionlint passed. YAML parsed and all 22 shell blocks passed `bash
-n`. Ruff checks, helper/test formatting and `git diff --check` passed.
**Validation limit:** the actual deployed OSS Codex binary has not been
exercised with a real capacity-failure/recovery model request. The
[linked goal/resume
implementation](https://github.com/zclllyybb/codex/blob/8072a457fad974d6e856b319e225541ee1bde59d/codex-rs/exec/src/lib.rs#L769)
supports exact-UUID goal resume, but the deployed binary reports only
`0.0.0` and its source SHA remains unverified. Real model/subagent
recovery and final Litefuse linkage still need a controlled deployment
smoke test. No production review, credential change or OSS replacement
was performed during validation.
### Release note
None
### Check List (For Author)
- Test
- [x] Unit Test
- [x] Manual test: Linux subprocess fault injection and workflow
validation
- Behavior changed:
- [x] Yes. Bounded recovery of transient capacity failures within the
existing review job.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test coverage and plan the deployed-binary recovery smoke
test
- [ ] Confirm document
- [ ] Add branch pick label
---
.github/scripts/run_review_with_resume.py | 467 ++++++++++
.github/scripts/test_review_auth_quarantine.py | 20 +-
.github/scripts/test_run_review_with_resume.py | 1087 ++++++++++++++++++++++++
.github/workflows/code-review-runner.yml | 61 +-
4 files changed, 1596 insertions(+), 39 deletions(-)
diff --git a/.github/scripts/run_review_with_resume.py
b/.github/scripts/run_review_with_resume.py
new file mode 100755
index 00000000000..0803ad4c147
--- /dev/null
+++ b/.github/scripts/run_review_with_resume.py
@@ -0,0 +1,467 @@
+#!/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.
+
+"""Resume the same review after capacity failures, never restart the
workflow."""
+
+import argparse
+import ctypes
+import json
+import os
+import shutil
+import signal
+import subprocess
+import sys
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+RETRY_DELAYS = (30, 60, 120)
+CAPACITY_MESSAGE = "Selected model is at capacity. Please try a different
model."
+PROCESS_EXIT_GRACE_SECONDS = 5
+
+
+class ChildReaper:
+ """Own orphaned commands in the standalone Linux helper, not the whole
runner."""
+
+ def __init__(self):
+ if sys.platform != "linux":
+ raise OSError("Review process supervision requires Linux")
+ # Fail before starting Codex if the runner cannot provide safe cleanup.
+ for module, name in (
+ (os, "pidfd_open"),
+ (os, "P_PIDFD"),
+ (signal, "pidfd_send_signal"),
+ ):
+ if not hasattr(module, name):
+ raise OSError(f"Review process supervision requires {name}")
+ fd = os.pidfd_open(os.getpid())
+ try:
+ signal.pidfd_send_signal(fd, 0)
+ try:
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG | os.WNOWAIT)
+ except ChildProcessError:
+ pass # Expected: this process is not its own child.
+ finally:
+ os.close(fd)
+ self.children = Path(f"/proc/self/task/{os.getpid()}/children")
+ self.children.read_text()
+ prctl = ctypes.CDLL(None, use_errno=True).prctl
+ prctl.argtypes = [ctypes.c_int] + [ctypes.c_ulong] * 4
+ prctl.restype = ctypes.c_int
+ # PR_SET_CHILD_SUBREAPER: descendants that outlive Codex are reparented
+ # to this helper, even if shell/PTY commands created new sessions.
+ if prctl(36, 1, 0, 0, 0) != 0:
+ error = ctypes.get_errno()
+ raise OSError(error, os.strerror(error))
+
+ def reap(self):
+ # Called only after Popen.wait() reaps Codex. This dedicated helper has
+ # no other concurrent subprocesses; gh/help commands run between
attempts.
+ deadline = time.monotonic() + PROCESS_EXIT_GRACE_SECONDS
+ while children := self.children.read_text().split():
+ if time.monotonic() >= deadline:
+ raise OSError("Codex descendant cleanup did not finish; not
resuming")
+ for child in children:
+ try:
+ fd = os.pidfd_open(int(child))
+ except ProcessLookupError:
+ continue
+ try:
+ # Kernel-verified parenthood plus a stable pidfd prevents
+ # signalling an unrelated process if a PID was recycled.
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG |
os.WNOWAIT)
+ signal.pidfd_send_signal(fd, signal.SIGKILL)
+ os.waitid(os.P_PIDFD, fd, os.WEXITED | os.WNOHANG)
+ except (ChildProcessError, ProcessLookupError):
+ pass
+ finally:
+ os.close(fd)
+ # Killing one orphan can reparent its children to us on the next
pass.
+ time.sleep(0.01)
+
+
+def read_events(path):
+ with path.open() as handle:
+ events = [json.loads(line) for line in handle if line.strip()]
+ if any(not isinstance(event, dict) for event in events):
+ raise ValueError("Codex JSONL contains a non-object event")
+ return events
+
+
+def failure(events, status, stderr_path):
+ for event_type in ("turn.failed", "error"):
+ for event in reversed(events):
+ if event.get("type") == event_type:
+ error = event.get("error") or event
+ return error.get("message") or f"Codex exited with status
{status}"
+ lines = stderr_path.read_text(errors="replace").splitlines()
+ return next(
+ (line for line in reversed(lines) if line.strip()),
+ f"Codex exited with status {status}",
+ )
+
+
+def session_id(events):
+ ids = [
+ event.get("thread_id")
+ for event in events
+ if event.get("type") == "thread.started"
+ ]
+ if len(ids) != 1 or not isinstance(ids[0], str):
+ raise ValueError("Expected exactly one main thread.started event")
+ # Names and --last can fall back to a new thread in codex exec. A UUID uses
+ # thread/resume directly and fails if the persisted thread cannot be
loaded.
+ if str(uuid.UUID(ids[0])) != ids[0]:
+ raise ValueError("Main session ID is not a canonical UUID")
+ return ids[0]
+
+
+def require_rollout(codex_home, thread_id, cwd):
+ for path in (codex_home / "sessions").rglob(f"*{thread_id}.jsonl"):
+ with path.open() as handle:
+ meta = json.loads(handle.readline())
+ payload = meta.get("payload") or {}
+ if (
+ meta.get("type") == "session_meta"
+ and payload.get("id") == thread_id
+ and payload.get("cwd")
+ and Path(payload["cwd"]).resolve() == cwd.resolve()
+ ):
+ return
+ raise ValueError("Main session rollout is missing; refusing to start a new
review")
+
+
+def stop_process(process):
+ try:
+ # Codex handles SIGINT through its graceful turn-interrupt/shutdown
path.
+ process.send_signal(signal.SIGINT)
+ process.wait(timeout=PROCESS_EXIT_GRACE_SECONDS)
+ except subprocess.TimeoutExpired:
+ pass
+ finally:
+ process.kill()
+ process.wait()
+
+
+def run_attempt(command, events_path, stderr_path, timeout, reaper=None):
+ with events_path.open("w") as stdout, stderr_path.open("w") as stderr:
+ process = None
+ copier = None
+ try:
+ process = subprocess.Popen(
+ command,
+ stdout=stdout,
+ stderr=subprocess.PIPE,
+ text=True,
+ encoding="utf-8",
+ errors="replace",
+ start_new_session=True,
+ )
+
+ def copy_stderr():
+ for line in process.stderr:
+ stderr.write(line)
+ stderr.flush()
+ print(line, end="", file=sys.stderr, flush=True)
+
+ copier = threading.Thread(target=copy_stderr, daemon=True)
+ # Do not interrupt Thread.start() between the native thread being
+ # created and its ident being published. The copier inherits this
+ # mask; pending cancellation reaches the main thread on
restoration,
+ # inside the cleanup-protected region. Codex was spawned unmasked.
+ previous_mask = signal.pthread_sigmask(
+ signal.SIG_BLOCK, {signal.SIGINT, signal.SIGTERM}
+ )
+ try:
+ copier.start()
+ finally:
+ signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
+ return process.wait(timeout=timeout)
+ finally:
+ # Finish bounded cleanup before delivering cancellation, including
a
+ # second signal while an earlier cancellation is already unwinding.
+ # The stderr copier inherited a blocked mask at startup, so pending
+ # SIGINT/SIGTERM can only reach this thread after everything is
closed.
+ previous_mask = signal.pthread_sigmask(
+ signal.SIG_BLOCK, {signal.SIGINT, signal.SIGTERM}
+ )
+ try:
+ try:
+ try:
+ if process is not None:
+ stop_process(process)
+ finally:
+ # Popen itself may be interrupted before returning a
handle.
+ if reaper is not None:
+ reaper.reap()
+ finally:
+ if copier is not None and copier.ident is not None:
+ copier.join(timeout=5)
+ if copier.is_alive():
+ raise OSError(
+ "Codex stderr remained open after descendant
cleanup"
+ )
+ if process is not None:
+ process.stderr.close()
+ finally:
+ signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
+
+
+def append_events(source, target):
+ # Interrupted writes can leave a partial JSON line. Retain raw bytes in the
+ # attempt file, but keep the aggregate parseable by jq and the trace
uploader.
+ with source.open() as src, target.open("a") as dst:
+ for line in src:
+ if not line.strip():
+ continue
+ try:
+ event = json.loads(line)
+ except json.JSONDecodeError:
+ continue
+ if isinstance(event, dict):
+ dst.write(json.dumps(event) + "\n")
+
+
+def check_resume_target(args, started_at, remaining):
+ def api(path, paginated=False):
+ command = ["gh", "api", path]
+ if paginated:
+ command += ["--paginate", "--slurp"]
+ result = subprocess.run(
+ command,
+ check=True,
+ capture_output=True,
+ text=True,
+ timeout=min(30, remaining()),
+ )
+ return json.loads(result.stdout)
+
+ pr = api(f"repos/{args.repository}/pulls/{args.pr_number}")
+ if (
+ pr["state"] != "open"
+ or pr["head"]["sha"] != args.head_sha
+ or pr["base"]["sha"] != args.base_sha
+ ):
+ raise ValueError(
+ "PR base/head or open state changed; refusing to resume stale
context"
+ )
+ pages = api(
+ f"repos/{args.repository}/pulls/{args.pr_number}/reviews",
paginated=True
+ )
+ for page in pages:
+ for review in page:
+ if (
+ review.get("user", {}).get("login") == "github-actions[bot]"
+ and review.get("commit_id") == args.head_sha
+ and (review.get("submitted_at") or "") >= started_at
+ ):
+ # A capacity error can arrive after the final GitHub write.
+ # Keep the failure visible for manual verification instead of
+ # replaying a possibly already-completed external side effect.
+ raise ValueError(
+ "A bot review was already submitted during this run; "
+ "refusing automatic resume to avoid duplicate submission"
+ )
+
+
+def resume_prompt(goal_prompt):
+ return (
+ goal_prompt
+ + "\n\n"
+ + (
+ "Recovery after a temporary model-capacity error in this SAME
review session. "
+ "Continue unfinished work using the existing conversation and
shared subagent review ledger; "
+ "do not restart the review, reset the ledger, or reset the
three-round limit. "
+ "The previous process exited: do not assume child agents or shell
processes are still running. "
+ "Recover unfinished child-agent work using their recorded IDs and
ledger sections where possible. "
+ "Keep completed findings and replace only work that could not be
recovered. "
+ "Before any GitHub write, fetch the current PR reviews and inline
comments again; "
+ "the initial snapshots may be stale. Verify already-submitted work
and never submit it twice. "
+ "Mark the goal complete only after the original review completion
criteria are satisfied."
+ )
+ )
+
+
+def run_review(args, reaper=None):
+ context = args.context_dir
+ aggregate = context / "codex-events.jsonl"
+ stderr_log = context / "codex-stderr.log"
+ final_message = context / "codex-final-message.txt"
+ attempts = context / "codex-attempts"
+ attempts.mkdir()
+ aggregate.touch()
+ stderr_log.touch()
+ goal_prompt = (context / "codex_goal_prompt.txt").read_text()
+ deadline = time.monotonic() + args.budget_seconds
+ started_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+ thread_id = None
+
+ def remaining():
+ seconds = deadline - time.monotonic()
+ if seconds <= 0:
+ raise TimeoutError("Review's shared time budget was exhausted")
+ return seconds
+
+ def fail(message):
+ print(message, file=sys.stderr, flush=True)
+ with aggregate.open("a") as handle:
+ handle.write(
+ json.dumps(
+ {
+ "type": "turn.failed",
+ "source": "review-resume-runner",
+ "error": {"message": message},
+ }
+ )
+ + "\n"
+ )
+ return 1
+
+ try:
+ for attempt in range(len(RETRY_DELAYS) + 1):
+ events_path = attempts / f"{attempt + 1}.jsonl"
+ stderr_path = attempts / f"{attempt + 1}.stderr.log"
+ output_path = attempts / f"{attempt + 1}.final.txt"
+ final_message.write_text("")
+ command = [
+ "codex",
+ "exec",
+ "--goal",
+ "--cd",
+ str(args.cwd),
+ "--model",
+ args.model,
+ "--config",
+ f"model_reasoning_effort={args.effort}",
+ "--sandbox",
+ "danger-full-access",
+ "--color",
+ "never",
+ "--json",
+ "--output-last-message",
+ str(output_path),
+ ]
+ if thread_id:
+ command += ["resume", thread_id, resume_prompt(goal_prompt)]
+ else:
+ command += [goal_prompt]
+ print(
+ f"Starting Codex review attempt {attempt + 1}/4 "
+ f"(session={thread_id or 'new'},
remaining={remaining():.0f}s)",
+ file=sys.stderr,
+ flush=True,
+ )
+ try:
+ status = run_attempt(
+ command, events_path, stderr_path, remaining(),
reaper=reaper
+ )
+ finally:
+ # Preserve failed attempts and their child-thread IDs for
Litefuse.
+ if events_path.exists():
+ append_events(events_path, aggregate)
+ if stderr_path.exists():
+ with stderr_path.open("rb") as src, stderr_log.open("ab")
as dst:
+ shutil.copyfileobj(src, dst)
+ dst.write(b"\n")
+ if output_path.exists():
+ shutil.copyfile(output_path, final_message)
+
+ events = read_events(events_path)
+ if status < 0 or status in (124, 130, 137, 143):
+ return fail(
+ f"Codex was interrupted or timed out (status {status});
not resuming"
+ )
+ message = failure(events, status, stderr_path)
+ if status != 0 and (
+ message != CAPACITY_MESSAGE or attempt == len(RETRY_DELAYS)
+ ):
+ return fail(message)
+ current_id = session_id(events)
+ if thread_id and current_id != thread_id:
+ return fail(
+ "Codex resumed a different session; refusing further
attempts"
+ )
+ thread_id = current_id
+ if status == 0:
+ return 0
+ require_rollout(Path(os.environ["CODEX_HOME"]), thread_id,
args.cwd)
+ # Check parser support without authenticating or starting a model
request.
+ subprocess.run(
+ ["codex", "exec", "--goal", "resume", "--help"],
+ check=True,
+ capture_output=True,
+ timeout=min(10, remaining()),
+ )
+ delay = RETRY_DELAYS[attempt]
+ if remaining() <= delay:
+ return fail("Insufficient shared review budget for capacity
backoff")
+ print(
+ f"Capacity unavailable; waiting {delay}s before resuming
{thread_id}",
+ file=sys.stderr,
+ flush=True,
+ )
+ with aggregate.open("a") as handle:
+ handle.write(
+ json.dumps(
+ {
+ "type": "review.capacity_retry",
+ "thread_id": thread_id,
+ "next_attempt": attempt + 2,
+ "delay_seconds": delay,
+ }
+ )
+ + "\n"
+ )
+ time.sleep(delay)
+ check_resume_target(args, started_at, remaining)
+ except KeyboardInterrupt:
+ fail("Review cancelled; not resuming")
+ return 130
+ except subprocess.TimeoutExpired:
+ return fail(
+ "Review recovery stopped: shared time budget or helper timeout
exhausted"
+ )
+ except (ValueError, OSError, subprocess.SubprocessError) as exc:
+ return fail(f"Review recovery stopped: {exc}")
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--context-dir", type=Path, required=True)
+ parser.add_argument("--cwd", type=Path, required=True)
+ parser.add_argument("--repository", required=True)
+ parser.add_argument("--pr-number", required=True)
+ parser.add_argument("--head-sha", required=True)
+ parser.add_argument("--base-sha", required=True)
+ parser.add_argument("--model", required=True)
+ parser.add_argument("--effort", required=True)
+ # The workflow owns the total timeout and deducts setup/finalization time.
+ parser.add_argument("--budget-seconds", type=int, required=True)
+ args = parser.parse_args()
+
+ def cancelled(_signum, _frame):
+ raise KeyboardInterrupt
+
+ signal.signal(signal.SIGTERM, cancelled)
+ return run_review(args, reaper=ChildReaper())
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/.github/scripts/test_review_auth_quarantine.py
b/.github/scripts/test_review_auth_quarantine.py
old mode 100644
new mode 100755
index 035b78b75cc..2f414d97391
--- a/.github/scripts/test_review_auth_quarantine.py
+++ b/.github/scripts/test_review_auth_quarantine.py
@@ -19,6 +19,7 @@
"""Run the actual workflow shell steps against fake OSS, Codex and GitHub.
Requires bash, jq and GNU coreutils (gdate is accepted on macOS).
+Tests invoking the review helper require Linux process supervision.
Run: python3 -m unittest discover -s .github/scripts -p
test_review_auth_quarantine.py
No credentials or network access are used.
"""
@@ -26,7 +27,6 @@ No credentials or network access are used.
import hashlib
import json
import os
-from pathlib import Path
import re
import shutil
import subprocess
@@ -35,7 +35,7 @@ import tempfile
import textwrap
import time
import unittest
-
+from pathlib import Path
WORKFLOW = Path(__file__).resolve().parents[1] /
"workflows/code-review-runner.yml"
PREFIX = "oss://doris-community-ci/codex/"
@@ -103,6 +103,7 @@ import os
from pathlib import Path
import sys
+print('{"type":"thread.started","thread_id":"0199a213-81c0-7800-8aa1-bbab2a035a53"}')
print(os.environ.get("FAKE_CODEX_EVENTS", ""))
print(os.environ.get("FAKE_CODEX_STDERR", ""), file=sys.stderr, flush=True)
# Rotation during a run must not change the identity used for the failure
marker.
@@ -117,7 +118,9 @@ import os
from pathlib import Path
import sys
-if sys.argv[1] == "api":
+if sys.argv[1] == "api" and
any("/contents/.github/scripts/run_review_with_resume.py?ref=" in arg for arg
in sys.argv):
+ sys.stdout.write(Path(os.environ["FAKE_REVIEW_HELPER"]).read_text())
+elif sys.argv[1] == "api":
print(json.dumps([[{"submitted_at": "2099-01-01T00:00:00Z", "commit_id":
os.environ["HEAD_SHA"]}]]))
else:
Path(os.environ["FAKE_COMMENT_FILE"]).write_text(sys.argv[-1])
@@ -151,6 +154,8 @@ class ReviewAuthQuarantineTest(unittest.TestCase):
"OSS_AK": "fake", "OSS_SK": "fake", "OSS_ENDPOINT": "unused",
"REPO": "example/repo", "PR_NUMBER": "1", "HEAD_SHA": "a" * 40,
"GITHUB_WORKSPACE": str(self.root),
+ "BASE_SHA": "b" * 40, "HELPER_REF": "test-pin",
"REVIEW_TIMEOUT_MINUTES": "120",
+ "FAKE_REVIEW_HELPER":
str(Path(__file__).resolve().with_name("run_review_with_resume.py")),
})
self.new_runner()
@@ -167,11 +172,18 @@ class ReviewAuthQuarantineTest(unittest.TestCase):
self.env.pop("CODEX_AUTH_OSS_OBJECT", None)
def run_step(self, name, expected=0, **env):
+ if name == "Run automated code review":
+ if sys.platform != "linux":
+ self.skipTest("review helper process supervision requires
Linux")
+ # Each invocation models a separate review step with its own
attempt files.
+ context = Path(tempfile.mkdtemp(dir=self.env["RUNNER_TEMP"]))
+ (context / "codex_goal_prompt.txt").write_text("Fake review")
+ self.env["REVIEW_CONTEXT_DIR"] = str(context)
output = Path(self.env["GITHUB_OUTPUT"])
output.write_text("")
result = subprocess.run(
["bash", "--noprofile", "--norc", "-eo", "pipefail", "-c",
step_script(name)],
- env={**self.env, **env}, cwd=self.root, capture_output=True,
text=True, timeout=15,
+ env={**self.env, **env}, cwd=self.root, capture_output=True,
text=True, timeout=15, check=False,
)
self.assertEqual(result.returncode, expected, result.stdout +
result.stderr)
env_file = Path(self.env["GITHUB_ENV"])
diff --git a/.github/scripts/test_run_review_with_resume.py
b/.github/scripts/test_run_review_with_resume.py
new file mode 100755
index 00000000000..24f91723f4b
--- /dev/null
+++ b/.github/scripts/test_run_review_with_resume.py
@@ -0,0 +1,1087 @@
+#!/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 io
+import json
+import os
+import signal
+import subprocess
+import sys
+import tempfile
+import textwrap
+import time
+import unittest
+from contextlib import redirect_stderr
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+import emit_litefuse_otel_io as exporter
+import run_review_with_resume as runner
+
+THREAD = "0199a213-81c0-7800-8aa1-bbab2a035a53"
+CHILD = "0199a213-81c0-7800-8aa1-bbab2a035a54"
+OTHER = "0199a213-81c0-7800-8aa1-bbab2a035a55"
+
+
+def thread_event(thread_id=THREAD):
+ return {"type": "thread.started", "thread_id": thread_id}
+
+
+def failed(message=runner.CAPACITY_MESSAGE):
+ return {"type": "turn.failed", "error": {"message": message}}
+
+
+def completed():
+ return {"type": "turn.completed", "usage": {"input_tokens": 20,
"output_tokens": 3}}
+
+
+class ResumeReviewTest(unittest.TestCase):
+ def setUp(self):
+ self.temp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temp.cleanup)
+ root = Path(self.temp.name)
+ self.context = root / "context"
+ self.context.mkdir()
+ self.cwd = root / "checkout"
+ self.cwd.mkdir()
+ self.codex_home = root / "codex-home"
+ self.sessions = self.codex_home / "sessions" / "2026" / "09" / "07"
+ self.sessions.mkdir(parents=True)
+ self.args = SimpleNamespace(
+ context_dir=self.context,
+ cwd=self.cwd,
+ repository="apache/doris",
+ pr_number="123",
+ head_sha="a" * 40,
+ base_sha="b" * 40,
+ model="gpt-5.6-sol",
+ effort="xhigh",
+ budget_seconds=1000,
+ )
+ (self.context / "codex_goal_prompt.txt").write_text(
+ "Read review_prompt.txt; complete the review."
+ )
+ self.ledger = self.context / "subagent_review_findings.md"
+ self.ledger.write_text("Completed risk scan and round 1; keep these
findings.")
+ self.write_rollout()
+ self.clock = 0
+ self.commands = []
+ self.timeouts = []
+ self.sleeps = []
+ self.enterContext(
+ mock.patch.dict(os.environ, {"CODEX_HOME": str(self.codex_home)})
+ )
+ self.enterContext(redirect_stderr(io.StringIO()))
+ 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.target_check = self.enterContext(
+ mock.patch.object(runner, "check_resume_target")
+ )
+ self.help = self.enterContext(mock.patch.object(runner.subprocess,
"run"))
+
+ def write_rollout(self, *, thread_id=THREAD, cwd=None):
+ (self.sessions / f"rollout-2026-09-07-{thread_id}.jsonl").write_text(
+ json.dumps(
+ {
+ "type": "session_meta",
+ "payload": {"id": thread_id, "cwd": str(cwd or self.cwd)},
+ }
+ )
+ + "\n"
+ )
+
+ def advance(self, delay):
+ self.sleeps.append(delay)
+ self.clock += delay
+
+ def execute(self, attempts):
+ pending = iter(attempts)
+
+ def fake_attempt(command, events_path, stderr_path, timeout,
reaper=None):
+ self.commands.append(command)
+ self.timeouts.append(timeout)
+ spec = next(pending)
+ elapsed = spec.get("elapsed", 0)
+ if elapsed > timeout:
+ raise subprocess.TimeoutExpired(command, timeout)
+ self.clock += elapsed
+ events_path.write_text(
+ spec.get("raw", "".join(json.dumps(e) + "\n" for e in
spec["events"]))
+ )
+ stderr_path.write_text(spec.get("stderr", ""))
+ if "output" in spec:
+ Path(command[command.index("--output-last-message") +
1]).write_text(
+ spec["output"]
+ )
+ if "raises" in spec:
+ raise spec["raises"]
+ return spec.get("status", 1)
+
+ with mock.patch.object(runner, "run_attempt",
side_effect=fake_attempt):
+ return runner.run_review(self.args)
+
+ def events(self):
+ return runner.read_events(self.context / "codex-events.jsonl")
+
+ def last_error(self):
+ return self.events()[-1]["error"]["message"]
+
+ def test_success_does_not_retry(self):
+ self.assertEqual(
+ 0, self.execute([{"events": [thread_event(), completed()],
"status": 0}])
+ )
+ self.help.assert_not_called()
+ self.target_check.assert_not_called()
+ self.assertEqual([], self.sleeps)
+
+ def
test_capacity_resumes_exact_session_with_same_settings_and_ledger(self):
+ before = self.ledger.read_text()
+ self.assertEqual(
+ 0,
+ self.execute(
+ [
+ {
+ "events": [thread_event(), failed()],
+ "elapsed": 5,
+ "output": "partial",
+ },
+ {
+ "events": [thread_event(), completed()],
+ "status": 0,
+ "output": "done",
+ },
+ ]
+ ),
+ )
+ self.assertEqual([30], self.sleeps)
+ self.assertEqual([1000, 965], self.timeouts)
+ command = self.commands[1]
+ self.assertEqual(["resume", THREAD], command[-3:-1])
+ self.assertNotIn("--last", command)
+ for option in ("--goal", "--cd", "--model", "--config", "--sandbox",
"--json"):
+ self.assertIn(option, command)
+ self.assertEqual("gpt-5.6-sol", command[command.index("--model") + 1])
+ self.assertIn("do not restart the review", command[-1])
+ self.assertIn("initial snapshots may be stale", command[-1])
+ self.assertIn("do not assume child agents", command[-1])
+ self.assertEqual(before, self.ledger.read_text())
+ self.assertEqual("done", (self.context /
"codex-final-message.txt").read_text())
+ self.target_check.assert_called_once()
+
+ def test_retry_count_is_bounded(self):
+ self.assertEqual(1, self.execute([{"events": [thread_event(),
failed()]}] * 4))
+ self.assertEqual([30, 60, 120], self.sleeps)
+ self.assertEqual(4, len(self.commands))
+ self.assertEqual(runner.CAPACITY_MESSAGE, self.last_error())
+
+ def
test_no_retry_for_auth_usage_or_generic_errors_even_before_session_starts(self):
+ for message in (
+ "You've hit your usage limit. Try again at 9:00 PM.",
+ "refresh_token_reused",
+ "HTTP 500",
+ "permission denied",
+ ):
+ with self.subTest(message=message), tempfile.TemporaryDirectory()
as tmp:
+ self.args.context_dir = Path(tmp)
+ (self.args.context_dir /
"codex_goal_prompt.txt").write_text("review")
+ self.assertEqual(1, self.execute([{"events":
[failed(message)]}]))
+ self.assertEqual(
+ message,
+ runner.read_events(self.args.context_dir /
"codex-events.jsonl")[
+ -1
+ ]["error"]["message"],
+ )
+ self.assertEqual([], self.sleeps)
+
+ def test_stderr_auth_error_is_preserved(self):
+ message = "You've hit your usage limit. Try again at 9:00 PM."
+ self.assertEqual(1, self.execute([{"events": [], "stderr": message}]))
+ self.assertEqual(message, self.last_error())
+
+ def test_capacity_text_inside_tool_output_does_not_trigger_retry(self):
+ event = {
+ "type": "item.completed",
+ "item": {
+ "type": "command_execution",
+ "aggregated_output": runner.CAPACITY_MESSAGE,
+ },
+ }
+ self.assertEqual(
+ 1,
+ self.execute([{"events": [thread_event(), event, failed("auth
failed")]}]),
+ )
+ self.assertEqual([], self.sleeps)
+
+ def test_missing_session_id_fails_closed(self):
+ self.assertEqual(1, self.execute([{"events": [failed()]}]))
+ self.assertIn("thread.started", self.last_error())
+ self.assertEqual([], self.sleeps)
+
+ def test_thread_name_is_not_accepted_as_resume_id(self):
+ self.assertEqual(
+ 1, self.execute([{"events": [thread_event("latest-review"),
failed()]}])
+ )
+ self.assertEqual([], self.sleeps)
+
+ def test_multiple_bootstrap_ids_fail_closed(self):
+ self.assertEqual(
+ 1,
+ self.execute([{"events": [thread_event(), thread_event(CHILD),
failed()]}]),
+ )
+ self.assertEqual([], self.sleeps)
+
+ def test_missing_or_wrong_workspace_rollout_fails_closed(self):
+ self.write_rollout(cwd=self.context)
+ self.assertEqual(1, self.execute([{"events": [thread_event(),
failed()]}]))
+ self.assertIn("rollout is missing", self.last_error())
+ self.assertEqual([], self.sleeps)
+
+ def test_resuming_another_session_is_rejected(self):
+ self.assertEqual(
+ 1,
+ self.execute(
+ [
+ {"events": [thread_event(), failed()]},
+ {"events": [thread_event(OTHER), completed()], "status":
0},
+ ]
+ ),
+ )
+ self.assertIn("different session", self.last_error())
+
+ def test_unsupported_goal_resume_fails_without_new_request(self):
+ self.help.side_effect = subprocess.CalledProcessError(
+ 2, ["codex", "exec", "--help"]
+ )
+ self.assertEqual(1, self.execute([{"events": [thread_event(),
failed()]}]))
+ self.assertEqual(1, len(self.commands))
+ self.assertEqual([], self.sleeps)
+
+ def test_budget_is_not_reset_between_attempts(self):
+ self.args.budget_seconds = 100
+ self.assertEqual(
+ 1,
+ self.execute(
+ [
+ {"events": [thread_event(), failed()], "elapsed": 10},
+ {"events": [thread_event(), failed()], "elapsed": 10},
+ ]
+ ),
+ )
+ self.assertEqual([100, 60], self.timeouts)
+ self.assertEqual([30], self.sleeps)
+ self.assertIn("Insufficient shared review budget", self.last_error())
+
+ def test_normal_review_can_run_past_the_old_89_minute_limit(self):
+ self.args.budget_seconds = 120 * 60 - 120
+ self.assertEqual(
+ 0,
+ self.execute(
+ [
+ {
+ "events": [thread_event(), completed()],
+ "status": 0,
+ "elapsed": 95 * 60,
+ }
+ ]
+ ),
+ )
+ self.assertEqual([7080], self.timeouts)
+ self.help.assert_not_called()
+
+ def test_exhausted_setup_budget_does_not_start_codex(self):
+ self.args.budget_seconds = -1
+ self.assertEqual(1, self.execute([]))
+ self.assertEqual([], self.commands)
+ self.assertIn("shared time budget was exhausted", self.last_error())
+
+ def
test_timeout_preserves_raw_partial_json_but_aggregate_stays_parseable(self):
+ raw = json.dumps(thread_event()) + '\n{"type":"item.'
+ self.assertEqual(
+ 1,
+ self.execute(
+ [
+ {
+ "events": [],
+ "raw": raw,
+ "raises": subprocess.TimeoutExpired(["codex"], 1),
+ }
+ ]
+ ),
+ )
+ self.assertIn("timeout exhausted", self.last_error())
+ self.assertEqual(raw, (self.context /
"codex-attempts/1.jsonl").read_text())
+ self.assertEqual([], self.sleeps)
+
+ def test_non_object_json_is_rejected(self):
+ self.assertEqual(1, self.execute([{"events": [], "raw": "[]\n"}]))
+ self.assertIn("non-object event", self.last_error())
+
+ def test_cancellation_while_waiting_never_resumes(self):
+ self.sleep.side_effect = KeyboardInterrupt
+ self.assertEqual(130, self.execute([{"events": [thread_event(),
failed()]}]))
+ self.assertEqual(1, len(self.commands))
+ self.target_check.assert_not_called()
+
+ def test_interrupted_process_is_not_retried_even_with_capacity_event(self):
+ self.assertEqual(
+ 1, self.execute([{"events": [thread_event(), failed()], "status":
-15}])
+ )
+ self.assertEqual([], self.sleeps)
+
+ def test_existing_review_or_stale_pr_stops_resume_and_keeps_failure(self):
+ self.target_check.side_effect = ValueError("review already submitted")
+ self.assertEqual(1, self.execute([{"events": [thread_event(),
failed()]}]))
+ self.assertEqual(1, len(self.commands))
+ self.assertIn("review already submitted", self.last_error())
+
+ def test_failed_attempt_output_is_not_reused_as_final_response(self):
+ self.assertEqual(
+ 1,
+ self.execute(
+ [
+ {"events": [thread_event(), failed()], "output": "old
partial"},
+ {"events": [thread_event(), failed("auth failed")]},
+ ]
+ ),
+ )
+ self.assertEqual("", (self.context /
"codex-final-message.txt").read_text())
+ self.assertEqual(
+ "old partial", (self.context /
"codex-attempts/1.final.txt").read_text()
+ )
+
+ def
test_litefuse_retains_failed_attempt_items_and_child_ids_after_recovery(self):
+ first_item = {
+ "type": "item.completed",
+ "item": {
+ "type": "collab_tool_call",
+ "id": "item_0",
+ "tool": "spawn_agent",
+ "receiver_thread_ids": [CHILD],
+ },
+ }
+ last_item = {
+ "type": "item.completed",
+ "item": {"type": "agent_message", "id": "item_0", "text": "done"},
+ }
+ self.assertEqual(
+ 0,
+ self.execute(
+ [
+ {"events": [thread_event(), first_item, failed()]},
+ {"events": [thread_event(), last_item, completed()],
"status": 0},
+ ]
+ ),
+ )
+ events = exporter.load_jsonl(str(self.context / "codex-events.jsonl"))
+ self.assertEqual({CHILD}, exporter.receiver_thread_ids(events))
+ self.assertEqual("completed", exporter.latest_turn_result(events)[0])
+ args = SimpleNamespace(
+ **vars(self.args),
+ reasoning_effort="xhigh",
+ run_id="123",
+ workflow="review",
+ trace_name="review",
+ session_id="123",
+ environment="test",
+ max_json_chars=4000,
+ max_context_json_chars=4000,
+ )
+ _, payload, _ = exporter.build_ingestion_payload(args, "review",
"done", events)
+ items = [
+ event["body"]
+ for event in payload["batch"]
+ if event["body"].get("metadata", {}).get("item_id") == "item_0"
+ ]
+ self.assertEqual(2, len(items))
+ self.assertNotEqual(items[0]["id"], items[1]["id"])
+ self.assertNotEqual(
+ items[0]["metadata"]["event_line"],
items[1]["metadata"]["event_line"]
+ )
+
+
+class WorkflowBudgetTest(unittest.TestCase):
+ def test_cli_requires_the_workflow_budget(self):
+ with (
+ mock.patch.object(
+ sys,
+ "argv",
+ [
+ "runner",
+ "--context-dir",
+ "/tmp",
+ "--cwd",
+ "/tmp",
+ "--repository",
+ "apache/doris",
+ "--pr-number",
+ "123",
+ "--head-sha",
+ "a",
+ "--base-sha",
+ "b",
+ "--model",
+ "test",
+ "--effort",
+ "xhigh",
+ ],
+ ),
+ mock.patch.object(runner, "ChildReaper") as reaper,
+ redirect_stderr(io.StringIO()) as stderr,
+ self.assertRaises(SystemExit) as error,
+ ):
+ runner.main()
+ self.assertEqual(2, error.exception.code)
+ self.assertIn("--budget-seconds", stderr.getvalue())
+ reaper.assert_not_called()
+
+ def test_workflow_passes_remaining_budget_after_helper_download(self):
+ workflow = (
+ Path(__file__).resolve().parents[1] / "workflows" /
"code-review-runner.yml"
+ ).read_text()
+ review = workflow.split(" - name: Run automated code review\n",
1)[1]
+ self.assertIn(
+ "timeout-minutes: ${{ fromJSON(env.REVIEW_TIMEOUT_MINUTES) }}",
+ review.split(" run: |", 1)[0],
+ )
+ default_minutes = next(
+ line.split(":", 1)[1].strip()
+ for line in workflow.splitlines()
+ if line.strip().startswith("REVIEW_TIMEOUT_MINUTES:")
+ )
+ self.assertEqual("120", default_minutes)
+ script = textwrap.dedent(
+ review.split(" run: |\n", 1)[1].split("
status=$?", 1)[0]
+ )
+ # A shell gh stub writes the downloaded helper and advances Bash's
elapsed
+ # clock without waiting minutes. The helper reports the actual CLI
args.
+ gh = """gh() {
+ SECONDS=$((SECONDS + SETUP_SECONDS))
+ cat <<'HELPER'
+import json, sys
+print(json.dumps(sys.argv[1:]))
+HELPER
+}
+"""
+ for minutes, setup, expected in (
+ (default_minutes, 12, 7068),
+ ("150", 30, 8850),
+ ("120", 7201, -121),
+ ):
+ with (
+ self.subTest(minutes=minutes, setup=setup),
+ tempfile.TemporaryDirectory() as tmp,
+ ):
+ result = subprocess.run(
+ ["bash", "-e", "-c", gh + script],
+ env={
+ **os.environ,
+ "RUNNER_TEMP": tmp,
+ "REVIEW_CONTEXT_DIR": tmp,
+ "GITHUB_WORKSPACE": tmp,
+ "REPO": "apache/doris",
+ "PR_NUMBER": "123",
+ "HEAD_SHA": "a",
+ "BASE_SHA": "b",
+ "HELPER_REF": "pinned",
+ "REVIEW_TIMEOUT_MINUTES": minutes,
+ "SETUP_SECONDS": str(setup),
+ },
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ args = json.loads(result.stdout)
+ actual = int(args[args.index("--budget-seconds") + 1])
+ self.assertLessEqual(actual, expected)
+ self.assertGreaterEqual(actual, expected - 2)
+
+
+class ResumeTargetTest(unittest.TestCase):
+ def setUp(self):
+ self.args = SimpleNamespace(
+ repository="apache/doris",
+ pr_number="123",
+ head_sha="a" * 40,
+ base_sha="b" * 40,
+ )
+ self.pr = {
+ "state": "open",
+ "head": {"sha": self.args.head_sha},
+ "base": {"sha": self.args.base_sha},
+ }
+ self.review = {
+ "user": {"login": "github-actions[bot]"},
+ "commit_id": self.args.head_sha,
+ "submitted_at": "2026-09-07T11:30:00Z",
+ }
+
+ def check(self, pages):
+ def api(command, **kwargs):
+ data = pages if "--paginate" in command else self.pr
+ return SimpleNamespace(stdout=json.dumps(data))
+
+ with mock.patch.object(runner.subprocess, "run", side_effect=api):
+ runner.check_resume_target(self.args, "2026-09-07T11:00:00Z",
lambda: 500)
+
+ def test_published_review_on_later_page_prevents_duplicate_resume(self):
+ with self.assertRaisesRegex(ValueError, "already submitted"):
+ self.check([[], [self.review]])
+
+ def test_other_heads_humans_and_old_reviews_do_not_block(self):
+ old = {**self.review, "submitted_at": "2026-09-07T10:00:00Z"}
+ human = {**self.review, "user": {"login": "reviewer"}}
+ other = {**self.review, "commit_id": "c" * 40}
+ self.check([[old, human, other]])
+
+ def test_changed_pr_prevents_resume(self):
+ self.pr["base"]["sha"] = "c" * 40
+ with self.assertRaisesRegex(ValueError, "changed"):
+ self.check([[]])
+
+ def test_unavailable_api_does_not_blindly_resume(self):
+ with (
+ mock.patch.object(
+ runner.subprocess,
+ "run",
+ side_effect=subprocess.CalledProcessError(1, ["gh"]),
+ ),
+ self.assertRaises(subprocess.CalledProcessError),
+ ):
+ runner.check_resume_target(self.args, "2026-09-07T11:00:00Z",
lambda: 500)
+
+
+class ChildReaperTest(unittest.TestCase):
+ def setUp(self):
+ self.reaper = runner.ChildReaper.__new__(runner.ChildReaper)
+ self.reaper.children = mock.Mock()
+ self.reaper.children.read_text.side_effect = ["10", "20", ""]
+ self.open = self.enterContext(mock.patch.object(os, "pidfd_open",
create=True))
+ self.open.side_effect = [110, 120]
+ self.close = self.enterContext(mock.patch.object(os, "close"))
+ self.wait = self.enterContext(mock.patch.object(os, "waitid",
create=True))
+ self.enterContext(mock.patch.object(os, "P_PIDFD", 3, create=True))
+ self.send = self.enterContext(
+ mock.patch.object(signal, "pidfd_send_signal", create=True)
+ )
+ self.enterContext(mock.patch.object(runner.time, "sleep"))
+
+ def test_unsupported_platform_fails_before_starting_codex(self):
+ with (
+ mock.patch.object(runner.sys, "platform", "darwin"),
+ self.assertRaisesRegex(OSError, "requires Linux"),
+ ):
+ runner.ChildReaper()
+ self.open.assert_not_called()
+
+ def test_subreaper_setup_failure_is_not_silenced(self):
+ with (
+ mock.patch.object(runner.sys, "platform", "linux"),
+ mock.patch.object(Path, "read_text", return_value=""),
+ mock.patch.object(runner.ctypes, "CDLL") as libc,
+ mock.patch.object(runner.ctypes, "get_errno", return_value=1),
+ ):
+ libc.return_value.prctl.return_value = -1
+ with self.assertRaises(OSError):
+ runner.ChildReaper()
+ libc.return_value.prctl.assert_called_once_with(36, 1, 0, 0, 0)
+ self.close.assert_called_once_with(110)
+
+ def test_reaps_newly_adopted_descendants_with_stable_handles(self):
+ self.reaper.reap()
+ self.assertEqual([mock.call(10), mock.call(20)],
self.open.call_args_list)
+ self.assertEqual(
+ [mock.call(110, signal.SIGKILL), mock.call(120, signal.SIGKILL)],
+ self.send.call_args_list,
+ )
+ self.assertEqual([mock.call(110), mock.call(120)],
self.close.call_args_list)
+ self.assertEqual(
+ mock.call(3, 110, os.WEXITED | os.WNOHANG | os.WNOWAIT),
+ self.wait.call_args_list[0],
+ )
+
+ def test_non_child_pid_is_never_signalled(self):
+ self.reaper.children.read_text.side_effect = ["10", ""]
+ self.wait.side_effect = ChildProcessError()
+ self.reaper.reap()
+ self.send.assert_not_called()
+ self.close.assert_called_once_with(110)
+
+ def test_already_exited_child_is_safe(self):
+ self.reaper.children.read_text.side_effect = ["10", ""]
+ self.open.side_effect = ProcessLookupError()
+ self.reaper.reap()
+ self.send.assert_not_called()
+ self.close.assert_not_called()
+
+ def test_cleanup_deadline_fails_closed(self):
+ with (
+ mock.patch.object(runner.time, "monotonic", side_effect=[0, 6]),
+ self.assertRaisesRegex(OSError, "cleanup did not finish"),
+ ):
+ self.reaper.reap()
+ self.send.assert_not_called()
+
+
+class ProcessLifecycleTest(unittest.TestCase):
+ def test_copier_construction_failure_cleans_process(self):
+ self.check_startup_cleanup("construct")
+
+ def test_copier_start_failure_cleans_process(self):
+ self.check_startup_cleanup("start")
+
+ def test_cancellation_before_copier_starts_cleans_process(self):
+ self.check_startup_cleanup("cancel_before")
+
+ def test_cancellation_after_copier_starts_cleans_process(self):
+ self.check_startup_cleanup("cancel_after")
+
+ def check_startup_cleanup(self, stage):
+ processes = []
+ copiers = []
+ original_mask = signal.pthread_sigmask(signal.SIG_BLOCK, set())
+ real_popen = subprocess.Popen
+ real_thread = runner.threading.Thread
+ real_start = real_thread.start
+ reaper = mock.Mock()
+
+ def create_process(*args, **kwargs):
+ process = real_popen(*args, **kwargs)
+ processes.append(process)
+ return process
+
+ def create_thread(*args, **kwargs):
+ if stage == "construct":
+ raise RuntimeError("copier construction failed")
+ thread = real_thread(*args, **kwargs)
+ copiers.append(thread)
+ return thread
+
+ def start_thread(thread):
+ if stage == "start":
+ raise RuntimeError("copier start failed")
+ if stage == "cancel_before":
+ os.kill(os.getpid(), signal.SIGTERM)
+ real_start(thread)
+ if stage == "cancel_after":
+ os.kill(os.getpid(), signal.SIGTERM)
+
+ def cancelled(_signum, _frame):
+ raise KeyboardInterrupt
+
+ original_handler = signal.signal(signal.SIGTERM, cancelled)
+ try:
+ with (
+ tempfile.TemporaryDirectory() as tmp,
+ redirect_stderr(io.StringIO()),
+ mock.patch.object(
+ runner.subprocess, "Popen", side_effect=create_process
+ ),
+ mock.patch.object(
+ runner.threading, "Thread", side_effect=create_thread
+ ),
+ mock.patch.object(real_thread, "start", start_thread),
+ ):
+ expected = (
+ KeyboardInterrupt if stage.startswith("cancel") else
RuntimeError
+ )
+ with self.assertRaises(expected) as error:
+ runner.run_attempt(
+ [sys.executable, "-c", "import time; time.sleep(20)"],
+ Path(tmp) / "events",
+ Path(tmp) / "stderr",
+ 5,
+ reaper=reaper,
+ )
+ if expected is RuntimeError:
+ self.assertIn("copier", str(error.exception))
+ self.assertIsNotNone(
+ processes[0].poll(), "Codex survived startup failure"
+ )
+ self.assertTrue(processes[0].stderr.closed)
+ reaper.reap.assert_called_once_with()
+ self.assertTrue(all(not thread.is_alive() for thread in
copiers))
+ self.assertEqual(
+ original_mask, signal.pthread_sigmask(signal.SIG_BLOCK,
set())
+ )
+ finally:
+ signal.signal(signal.SIGTERM, original_handler)
+ signal.pthread_sigmask(signal.SIG_SETMASK, original_mask)
+ for process in processes:
+ if process.poll() is None:
+ process.kill()
+ process.wait(timeout=5)
+ for thread in copiers:
+ if thread.ident is not None:
+ thread.join(timeout=5)
+ for process in processes:
+ process.stderr.close()
+
+ def test_interrupted_popen_still_calls_reaper(self):
+ reaper = mock.Mock()
+ with (
+ tempfile.TemporaryDirectory() as tmp,
+ mock.patch.object(
+ runner.subprocess, "Popen", side_effect=KeyboardInterrupt
+ ),
+ self.assertRaises(KeyboardInterrupt),
+ ):
+ runner.run_attempt(
+ ["codex"], Path(tmp) / "events", Path(tmp) / "stderr", 5,
reaper=reaper
+ )
+ reaper.reap.assert_called_once_with()
+
+ def test_real_process_capacity_then_resume_preserves_state(self):
+ with tempfile.TemporaryDirectory() as tmp,
redirect_stderr(io.StringIO()):
+ root = Path(tmp)
+ (root / "codex_goal_prompt.txt").write_text("test review")
+ ledger = root / "subagent_review_findings.md"
+ ledger.write_text("completed round 1")
+ fake_codex = root / "codex"
+ fake_codex.write_text(
+ f"#!{sys.executable}\n"
+ + r"""
+import json, os, sys
+from pathlib import Path
+if "--help" in sys.argv:
+ print("exec --goal resume SESSION_ID PROMPT")
+ sys.exit(0)
+root = Path(os.environ["FAKE_REVIEW_ROOT"])
+thread_id = os.environ["FAKE_THREAD_ID"]
+print(json.dumps({"type": "thread.started", "thread_id": thread_id}),
flush=True)
+if "resume" in sys.argv:
+ assert sys.argv[sys.argv.index("resume") + 1] == thread_id
+ assert (root / "subagent_review_findings.md").read_text() == "completed
round 1"
+ assert (root / "request-count").read_text() == "1"
+ (root / "request-count").write_text("2")
+ Path(sys.argv[sys.argv.index("--output-last-message") +
1]).write_text("review complete")
+ print(json.dumps({"type": "turn.completed", "usage": {"input_tokens":
123}}))
+else:
+ sessions = Path(os.environ["CODEX_HOME"]) / "sessions"
+ sessions.mkdir(parents=True)
+ (sessions / f"rollout-{thread_id}.jsonl").write_text(json.dumps({
+ "type": "session_meta", "payload": {"id": thread_id, "cwd": str(root)}
+ }) + "\n")
+ (root / "request-count").write_text("1")
+ print(json.dumps({"type": "turn.failed", "error": {
+ "message": "Selected model is at capacity. Please try a different
model."
+ }}))
+ sys.exit(1)
+"""
+ )
+ fake_codex.chmod(0o700)
+ args = SimpleNamespace(
+ context_dir=root,
+ cwd=root,
+ repository="apache/doris",
+ pr_number="123",
+ head_sha="a" * 40,
+ base_sha="b" * 40,
+ model="gpt-5.6-sol",
+ effort="xhigh",
+ budget_seconds=30,
+ )
+ with (
+ mock.patch.dict(
+ os.environ,
+ {
+ "PATH": str(root) + os.pathsep + os.environ["PATH"],
+ "CODEX_HOME": str(root / "isolated-home"),
+ "FAKE_REVIEW_ROOT": str(root),
+ "FAKE_THREAD_ID": THREAD,
+ },
+ ),
+ mock.patch.object(runner, "RETRY_DELAYS", (0, 0, 0)),
+ mock.patch.object(runner, "check_resume_target"),
+ ):
+ self.assertEqual(0, runner.run_review(args))
+ self.assertEqual("2", (root / "request-count").read_text())
+ self.assertEqual(
+ "review complete", (root /
"codex-final-message.txt").read_text()
+ )
+ self.assertEqual(2, len(list((root /
"codex-attempts").glob("*.jsonl"))))
+
+ def test_actual_process_captures_stdout_and_stderr(self):
+ with tempfile.TemporaryDirectory() as tmp,
redirect_stderr(io.StringIO()):
+ root = Path(tmp)
+ status = runner.run_attempt(
+ [
+ sys.executable,
+ "-c",
+ "import sys; print('output'); print('error',
file=sys.stderr)",
+ ],
+ root / "events",
+ root / "stderr",
+ 5,
+ )
+ self.assertEqual(0, status)
+ self.assertEqual("output\n", (root / "events").read_text())
+ self.assertEqual("error\n", (root / "stderr").read_text())
+
+ def test_timeout_terminates_codex_process(self):
+ with tempfile.TemporaryDirectory() as tmp,
redirect_stderr(io.StringIO()):
+ root = Path(tmp)
+ program = "import os,time; print(os.getpid(), flush=True);
time.sleep(60)"
+ with self.assertRaises(subprocess.TimeoutExpired):
+ runner.run_attempt(
+ [sys.executable, "-c", program],
+ root / "events",
+ root / "stderr",
+ 0.2,
+ )
+ pid = int((root / "events").read_text())
+ with self.assertRaises(ProcessLookupError):
+ os.kill(pid, 0)
+
+ @unittest.skipUnless(sys.platform == "linux", "CLI supervision requires
Linux")
+ def test_cli_sigterm_cleans_up_child_and_does_not_restart(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ (root / "codex_goal_prompt.txt").write_text(
+ "test only; do not call a service"
+ )
+ fake_codex = root / "codex"
+ fake_codex.write_text(
+ f"#!{sys.executable}\n"
+ + "import json,os,time\nfrom pathlib import Path\n"
+ + f"print(json.dumps({thread_event()!r}), flush=True)\n"
+ +
"Path(os.environ['FAKE_CHILD_PID']).write_text(str(os.getpid()))\n"
+ + "time.sleep(60)\n"
+ )
+ fake_codex.chmod(0o700)
+ pid_file = root / "child.pid"
+ command = [
+ sys.executable,
+ str(Path(runner.__file__).resolve()),
+ "--context-dir",
+ str(root),
+ "--cwd",
+ str(root),
+ "--repository",
+ "apache/doris",
+ "--pr-number",
+ "123",
+ "--head-sha",
+ "a" * 40,
+ "--base-sha",
+ "b" * 40,
+ "--model",
+ "gpt-5.6-sol",
+ "--effort",
+ "xhigh",
+ "--budget-seconds",
+ "30",
+ ]
+ process = subprocess.Popen(
+ command,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ env={
+ **os.environ,
+ "PATH": str(root) + os.pathsep + os.environ["PATH"],
+ "CODEX_HOME": str(root / "isolated-home"),
+ "FAKE_CHILD_PID": str(pid_file),
+ },
+ )
+ try:
+ deadline = time.monotonic() + 5
+ while not pid_file.exists() and time.monotonic() < deadline:
+ time.sleep(0.01)
+ self.assertTrue(pid_file.exists(), "fake Codex did not start")
+ process.send_signal(signal.SIGTERM)
+ process.communicate(timeout=10)
+ self.assertEqual(130, process.returncode)
+ self.assertEqual(
+ 1, len(list((root / "codex-attempts").glob("*.jsonl")))
+ )
+ events = runner.read_events(root / "codex-events.jsonl")
+ self.assertEqual(
+ "Review cancelled; not resuming",
events[-1]["error"]["message"]
+ )
+ with self.assertRaises(ProcessLookupError):
+ os.kill(int(pid_file.read_text()), 0)
+ finally:
+ if process.poll() is None:
+ process.kill()
+ process.communicate(timeout=10)
+
+ @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 graceful in (True, False):
+ with self.subTest(stop=stop, graceful=graceful):
+ self.check_detached_descendants(stop, graceful)
+
+ @unittest.skipUnless(sys.platform == "linux", "requires Linux
subreaper/pidfd")
+ def test_cancellation_during_reaping_finishes_cleanup_without_retry(self):
+ for stop in ("reap_cancel", "reap_cancel_repeat",
"cancel_reap_repeat"):
+ with self.subTest(stop=stop):
+ self.check_detached_descendants(stop, True)
+
+ def check_detached_descendants(self, stop, graceful):
+ with tempfile.TemporaryDirectory() as tmp:
+ root = Path(tmp)
+ (root / "codex_goal_prompt.txt").write_text("local process test
only")
+ fake_codex = root / "codex"
+ fake_codex.write_text(
+ f"#!{sys.executable}\n"
+ + r"""
+import json, os, signal, subprocess, sys, time
+from pathlib import Path
+root = Path(os.environ["TREE_ROOT"])
+role = sys.argv[1]
+if role == "worker":
+ signal.signal(signal.SIGINT, signal.SIG_IGN)
+ signal.signal(signal.SIGTERM, signal.SIG_IGN)
+elif role == "shell":
+ worker = subprocess.Popen([sys.executable, __file__, "worker"],
start_new_session=True)
+ print(worker.pid, flush=True)
+else:
+ shell = subprocess.Popen(
+ [sys.executable, __file__, "shell"], start_new_session=True,
+ stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True,
+ )
+ worker_pid = int(shell.stdout.readline())
+ def interrupted(_signum, _frame):
+ (root / "interrupted").touch()
+ shell.terminate()
+ shell.wait()
+ sys.exit(0)
+ signal.signal(signal.SIGINT, interrupted if os.environ["TREE_GRACEFUL"] ==
"1" else signal.SIG_IGN)
+ (root / "pids.json").write_text(json.dumps([shell.pid, worker_pid]))
+ print(json.dumps({"type": "thread.started", "thread_id":
os.environ["TREE_THREAD"]}), flush=True)
+ if os.environ["TREE_STOP"] in ("exit", "reap_cancel",
"reap_cancel_repeat"):
+ print(json.dumps({"type": "turn.failed", "error": {"message": "test
failure"}}), flush=True)
+ sys.exit(1)
+deadline = time.monotonic() + 20
+while time.monotonic() < deadline:
+ time.sleep(1)
+"""
+ )
+ fake_codex.chmod(0o700)
+ command = [
+ sys.executable,
+ "-c",
+ (
+ "import sys; sys.path.insert(0, sys.argv.pop(1)); "
+ "import run_review_with_resume as r; "
+ "r.PROCESS_EXIT_GRACE_SECONDS = 0.5; "
+ + (
+ """
+import os, signal
+original_open = r.os.pidfd_open
+def interrupt_cleanup(pid):
+ fd = original_open(pid)
+ if pid != os.getpid():
+ os.kill(os.getpid(), signal.SIGTERM)
+ if os.environ["TREE_STOP"].endswith("repeat"):
+ for _ in range(3):
+ os.kill(os.getpid(), signal.SIGINT)
+ os.kill(os.getpid(), signal.SIGTERM)
+ return fd
+r.os.pidfd_open = interrupt_cleanup
+"""
+ if "reap" in stop
+ else ""
+ )
+ + "sys.exit(r.main())"
+ ),
+ str(Path(runner.__file__).parent),
+ "--context-dir",
+ str(root),
+ "--cwd",
+ str(root),
+ "--repository",
+ "apache/doris",
+ "--pr-number",
+ "123",
+ "--head-sha",
+ "a" * 40,
+ "--base-sha",
+ "b" * 40,
+ "--model",
+ "test",
+ "--effort",
+ "xhigh",
+ "--budget-seconds",
+ "2" if stop == "timeout" else "30",
+ ]
+ with subprocess.Popen(
+ [sys.executable, "-c", "import time; time.sleep(30)"],
+ start_new_session=True,
+ ) as unrelated:
+ process = subprocess.Popen(
+ command,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ env={
+ **os.environ,
+ "PATH": str(root) + os.pathsep + os.environ["PATH"],
+ "TREE_ROOT": str(root),
+ "TREE_THREAD": THREAD,
+ "TREE_STOP": stop,
+ "TREE_GRACEFUL": str(int(graceful)),
+ "CODEX_HOME": str(root / "isolated-home"),
+ },
+ )
+ try:
+ deadline = time.monotonic() + 5
+ while (
+ not (root / "pids.json").exists()
+ and time.monotonic() < deadline
+ ):
+ time.sleep(0.01)
+ self.assertTrue(
+ (root / "pids.json").exists(), "fake Codex did not
start"
+ )
+ 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
+ )
+ for pid in json.loads((root / "pids.json").read_text()):
+ with self.assertRaises(ProcessLookupError):
+ os.kill(pid, 0)
+ self.assertIsNone(
+ unrelated.poll(), "cleanup killed an unrelated process"
+ )
+ self.assertEqual(
+ 1, len(list((root / "codex-attempts").glob("*.jsonl")))
+ )
+ if "cancel" in stop:
+ events = runner.read_events(root /
"codex-events.jsonl")
+ self.assertEqual(
+ "Review cancelled; not resuming",
+ events[-1]["error"]["message"],
+ )
+ if stop in ("cancel", "timeout", "cancel_reap_repeat"):
+ self.assertEqual(graceful, (root /
"interrupted").exists())
+ finally:
+ if process.poll() is None:
+ process.kill()
+ process.communicate(timeout=5)
+ unrelated.kill()
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/.github/workflows/code-review-runner.yml
b/.github/workflows/code-review-runner.yml
index ba5278465ff..75b168d4d96 100644
--- a/.github/workflows/code-review-runner.yml
+++ b/.github/workflows/code-review-runner.yml
@@ -23,28 +23,6 @@ on:
required: false
type: boolean
default: true
- workflow_call:
- inputs:
- pr_number:
- required: true
- type: string
- head_sha:
- required: true
- type: string
- base_sha:
- required: true
- type: string
- review_focus:
- required: false
- type: string
- default: ''
- manage_status:
- description: >-
- Update the code-review commit status. Callers that enable this must
- grant statuses: write.
- required: false
- type: boolean
- default: false
permissions:
statuses: write
@@ -52,6 +30,10 @@ permissions:
contents: read
issues: write
+# One review-step budget for helper download, all attempts, and final
verification.
+env:
+ REVIEW_TIMEOUT_MINUTES: 120
+
jobs:
code-review:
runs-on: ubuntu-latest
@@ -711,33 +693,42 @@ jobs:
- name: Run automated code review
id: review
- timeout-minutes: 120
+ timeout-minutes: ${{ fromJSON(env.REVIEW_TIMEOUT_MINUTES) }}
continue-on-error: true
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
REPO: ${{ github.repository }}
PR_NUMBER: ${{ steps.review_inputs.outputs.pr_number }}
HEAD_SHA: ${{ steps.review_inputs.outputs.head_sha }}
+ BASE_SHA: ${{ steps.review_inputs.outputs.base_sha }}
+ HELPER_REF: ${{ github.workflow_sha || github.sha }}
run: |
- GOAL_PROMPT="$(cat "$REVIEW_CONTEXT_DIR/codex_goal_prompt.txt")"
+ review_step_started_at=$SECONDS
review_started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+ helper="$RUNNER_TEMP/run_review_with_resume.py"
+ gh api \
+ -H "Accept: application/vnd.github.raw" \
+
"repos/${REPO}/contents/.github/scripts/run_review_with_resume.py?ref=${HELPER_REF}"
\
+ > "$helper"
+
+ # Share the step's budget; reserve two minutes for process cleanup
and
+ # the existing GitHub review verification. Helper download counts
too.
+ budget_seconds=$((REVIEW_TIMEOUT_MINUTES * 60 - (SECONDS -
review_step_started_at) - 120))
set +e
# GitHub-hosted runners are ephemeral. Avoid workspace-write here
because
# Codex uses bubblewrap for that mode and uid maps can be
unavailable.
- codex exec --goal "$GOAL_PROMPT" \
- --cd "$GITHUB_WORKSPACE" \
+ python3 "$helper" \
+ --context-dir "$REVIEW_CONTEXT_DIR" \
+ --cwd "$GITHUB_WORKSPACE" \
+ --repository "$REPO" \
+ --pr-number "$PR_NUMBER" \
+ --head-sha "$HEAD_SHA" \
+ --base-sha "$BASE_SHA" \
--model "gpt-5.6-sol" \
- --config "model_reasoning_effort=xhigh" \
- --sandbox danger-full-access \
- --color never \
- --json \
- --output-last-message
"$REVIEW_CONTEXT_DIR/codex-final-message.txt" \
- > "$REVIEW_CONTEXT_DIR/codex-events.jsonl" \
- 2> >(tee "$REVIEW_CONTEXT_DIR/codex-stderr.log" >&2)
+ --effort xhigh \
+ --budget-seconds "$budget_seconds"
status=$?
- # Drain the stderr tee before inspecting it for refresh-token
failures.
- wait
set -e
failure_reason=""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
