mirror of
https://github.com/apache/doris.git
synced 2026-11-04 07:31:17 +00:00
shuke
173a0e2064
### What problem does this PR solve? Related PR: https://github.com/shuke987/codex/pull/6 Upgrade the Code Review Runner and its runtime canary from Codex 0.162.1 to the validated [0.166.0 fork release](https://github.com/shuke987/codex/releases/tag/exec-goal-v0.166.0), retaining goal completion and same-session resume support. Codex 0.166.0 distinguishes `policy_capacity_unavailable` from ordinary overloads. Accept both canonical terminal capacity messages so the existing bounded recovery policy continues to apply. Preserve the original diagnostic and exact submission verification; unrelated failures remain non-retryable. Update the archive, version and CLI checksum pins together. The workflow changes are limited to seven package pin replacements. Existing model selection and delivery checks are preserved.
680 lines
28 KiB
Python
680 lines
28 KiB
Python
#!/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 re
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from review_github_api import decode_json, run_gh
|
|
from submit_review import RUN_FILE, RESULT_FILE, SUBMISSION_FILE, verify_completion, write_json
|
|
|
|
# Six same-session capacity retries, each after a fixed five-minute wait.
|
|
RETRY_DELAYS = (300,) * 6
|
|
CAPACITY_FALLBACK_AFTER = 5
|
|
CAPACITY_MESSAGE = "Selected model is at capacity. Please try a different model."
|
|
# Codex 0.166 distinguishes policy capacity from ordinary overloads. Both
|
|
# retain the same bounded recovery; other policy/safety failures must not retry.
|
|
CAPACITY_MESSAGES = (
|
|
CAPACITY_MESSAGE,
|
|
"Policy-restricted capacity is unavailable. Please try again later.",
|
|
)
|
|
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:
|
|
command = ["gh", "api", f"repos/{self.args.repository}/pulls/{self.args.pr_number}"]
|
|
response = run_gh(
|
|
command,
|
|
check=True, capture_output=True, text=True, timeout=timeout,
|
|
)
|
|
check_head(decode_json(response, command), 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:
|
|
"""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; synchronous gh/help calls have returned.
|
|
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 attempt_result(events, status, stderr_path):
|
|
"""Use the latest turn in this attempt, never a previous attempt's failure."""
|
|
terminal = next(
|
|
(
|
|
event
|
|
for event in reversed(events)
|
|
if event.get("type")
|
|
in ("turn.started", "turn.completed", "turn.failed", "error")
|
|
),
|
|
{},
|
|
)
|
|
event_type = terminal.get("type", "missing")
|
|
if event_type == "turn.completed":
|
|
message = (
|
|
None
|
|
if status == 0
|
|
else f"Codex exited with status {status} after turn.completed"
|
|
)
|
|
return event_type, message
|
|
if event_type in ("turn.failed", "error"):
|
|
error = terminal.get("error") or terminal
|
|
message = (
|
|
error.get("message")
|
|
or f"Codex reported {event_type} (exit status {status})"
|
|
)
|
|
return event_type, message
|
|
# A zero exit after only thread.started (or an unfinished new turn) does
|
|
# not prove recovery. Do not reuse an older capacity event to retry it.
|
|
if status == 0:
|
|
return (
|
|
event_type,
|
|
"Codex exited with status 0 without a terminal turn event; review is incomplete",
|
|
)
|
|
lines = stderr_path.read_text(errors="replace").splitlines()
|
|
return event_type, next(
|
|
(line for line in reversed(lines) if line.strip()),
|
|
f"Codex exited with status {status}",
|
|
)
|
|
|
|
|
|
def can_fallback_model(events, message, model):
|
|
"""Only a rejected first model request may start a new review session."""
|
|
return (
|
|
message is not None
|
|
and f"The '{model}' {UNSUPPORTED_CHATGPT_MODEL}" in message
|
|
and not any(
|
|
event.get("type") == "turn.completed"
|
|
or event.get("type", "").startswith("item.")
|
|
for event in events
|
|
)
|
|
)
|
|
|
|
|
|
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, watchdog=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)
|
|
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.
|
|
# 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}")
|
|
check_head(pr, args.head_sha)
|
|
if (
|
|
pr["state"] != "open"
|
|
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")
|
|
run = {
|
|
"repository": args.repository, "pr_number": args.pr_number,
|
|
"head_sha": args.head_sha, "base_sha": args.base_sha,
|
|
"started_at": started_at, "token": str(uuid.uuid4()), "deadline": deadline,
|
|
}
|
|
write_json(context / RUN_FILE, run)
|
|
thread_id = None
|
|
model = args.model
|
|
fallback_model = getattr(args, "fallback_model", None)
|
|
capacity_fallback_model = getattr(args, "capacity_fallback_model", None)
|
|
capacity_retry = 0
|
|
attempt_number = 0
|
|
|
|
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
|
|
|
|
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"
|
|
final_message.write_text("")
|
|
command = [
|
|
"codex",
|
|
"exec",
|
|
"--goal",
|
|
"--cd",
|
|
str(args.cwd),
|
|
"--model",
|
|
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_number} "
|
|
f"(model={model}, 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, watchdog=watchdog
|
|
)
|
|
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)
|
|
|
|
# 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(
|
|
f"Codex was interrupted or timed out (status {status}); not resuming"
|
|
)
|
|
terminal, message = attempt_result(events, status, stderr_path)
|
|
print(
|
|
f"Finished Codex review attempt {attempt_number} "
|
|
f"(exit_status={status}, terminal_event={terminal}, events={len(events)})",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
if (
|
|
fallback_model
|
|
and fallback_model != model
|
|
and thread_id is None
|
|
and can_fallback_model(events, message, model)
|
|
):
|
|
check_resume_target(args, started_at, remaining)
|
|
print(
|
|
f"Model {model} is unavailable for this ChatGPT account; "
|
|
f"retrying with {fallback_model} before review work began",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
model = fallback_model
|
|
continue
|
|
if message is not None and message not in CAPACITY_MESSAGES:
|
|
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 message is None or (context / SUBMISSION_FILE).exists():
|
|
# 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.
|
|
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 in CAPACITY_MESSAGES
|
|
write_json(context / RESULT_FILE, result)
|
|
with aggregate.open("a") as handle:
|
|
handle.write(json.dumps({"type": "review.completed", **result}) + "\n")
|
|
print(
|
|
f"Verified final review {result['review_id']}: "
|
|
f"P0={result['p0']}, P1={result['p1']}, "
|
|
f"converged={result['converged']}, rounds={result['rounds']}, "
|
|
f"recovered_after_capacity={result['recovered_after_capacity']}",
|
|
file=sys.stderr, flush=True,
|
|
)
|
|
return 0
|
|
if capacity_retry == len(RETRY_DELAYS):
|
|
return fail(message)
|
|
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[capacity_retry]
|
|
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_number + 1,
|
|
"delay_seconds": delay,
|
|
}
|
|
)
|
|
+ "\n"
|
|
)
|
|
watchdog.wait(delay)
|
|
check_resume_target(args, started_at, remaining)
|
|
capacity_retry += 1
|
|
if capacity_retry == CAPACITY_FALLBACK_AFTER and capacity_fallback_model:
|
|
print(
|
|
f"Switching capacity recovery from {model} to {capacity_fallback_model} "
|
|
f"after {CAPACITY_FALLBACK_AFTER} capacity failures; keeping session {thread_id}",
|
|
file=sys.stderr, flush=True,
|
|
)
|
|
model = capacity_fallback_model
|
|
goal_prompt += (
|
|
f"\n\nCapacity recovery switched to {model}. Use this model for remaining "
|
|
"review work, including child agents. If an unfinished child cannot resume "
|
|
"with this model, replace only its unfinished work using its recorded ledger "
|
|
"section; preserve completed findings and the cumulative round count."
|
|
)
|
|
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
|
|
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("--fallback-model")
|
|
parser.add_argument("--capacity-fallback-model", help="Resume with this model after five capacity failures")
|
|
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())
|