doris/.github/scripts/test_run_review_with_resume.py

#!/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 shutil
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}}


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()
        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",
            fallback_model=None,
            capacity_fallback_model="gpt-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.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={
                "state": "success", "p0": 0, "p1": 0, "review_id": 123,
                "converged": True, "rounds": 1, "incomplete_reason": "",
            }
        ))

    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, watchdog=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 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}])
        )
        self.help.assert_not_called()
        self.target_check.assert_not_called()
        self.assertEqual([], self.sleeps)

    def test_completed_turn_without_verified_delivery_fails(self):
        self.verification.side_effect = ValueError("No final review submission was declared")
        self.assertEqual(1, self.execute([{"events": [thread_event(), completed()], "status": 0}]))
        self.assertIn("No final review submission", self.last_error())
        self.assertFalse((self.context / "review-result.json").exists())

    def test_partial_submission_at_capacity_does_not_resume_or_pass(self):
        (self.context / "review-submission.json").write_text("{}")
        self.verification.side_effect = ValueError("Final review inline comments were not completely verified")
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed()]}]))
        self.assertEqual([], self.sleeps)
        self.target_check.assert_not_called()
        self.assertFalse((self.context / "review-result.json").exists())

    def test_only_capacity_can_recover_after_submission(self):
        (self.context / "review-submission.json").write_text("{}")
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed("Other failure")]}]))
        self.verification.assert_not_called()

    def test_cancellation_after_submission_is_still_failure(self):
        (self.context / "review-submission.json").write_text("{}")
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed()], "status": 143}]))
        self.verification.assert_not_called()

    def test_unsupported_model_falls_back_before_review_work(self):
        self.args.model = "gpt-6-sol"
        self.args.fallback_model = "gpt-5.6-sol"
        rejection = (
            "The 'gpt-6-sol' model is not supported when using Codex "
            "with a ChatGPT account."
        )
        server_error = json.dumps(
            {
                "type": "error",
                "status": 400,
                "error": {"type": "invalid_request_error", "message": rejection},
            }
        )
        self.assertEqual(
            0,
            self.execute(
                [
                    {
                        "events": [
                            thread_event(),
                            {"type": "turn.started"},
                            failed(server_error),
                        ]
                    },
                    {"events": [thread_event(OTHER), completed()], "status": 0},
                ]
            ),
        )
        self.assertEqual(
            ["gpt-6-sol", "gpt-5.6-sol"],
            [command[command.index("--model") + 1] for command in self.commands],
        )
        self.assertNotIn("resume", self.commands[1])
        self.assertEqual(
            "gpt-5.6-sol\n", (self.context / "codex-review-model.txt").read_text()
        )
        self.assertEqual([], self.sleeps)
        self.target_check.assert_called_once()

    def test_unsupported_model_does_not_restart_after_item(self):
        self.args.model = "gpt-6-sol"
        self.args.fallback_model = "gpt-5.6-sol"
        rejection = (
            "The 'gpt-6-sol' model is not supported when using Codex "
            "with a ChatGPT account."
        )
        self.assertEqual(
            1,
            self.execute(
                [
                    {
                        "events": [
                            thread_event(),
                            {"type": "item.started"},
                            failed(rejection),
                        ]
                    }
                ]
            ),
        )
        self.assertEqual(1, len(self.commands))
        self.target_check.assert_not_called()

    def test_both_models_rejected_stops_after_one_fallback(self):
        self.args.model = "gpt-6-sol"
        self.args.fallback_model = "gpt-5.6-sol"

        def rejection(model):
            return (
                f"The '{model}' model is not supported when using Codex "
                "with a ChatGPT account."
            )

        self.assertEqual(
            1,
            self.execute(
                [
                    {"events": [thread_event(), failed(rejection("gpt-6-sol"))]},
                    {"events": [thread_event(OTHER), failed(rejection("gpt-5.6-sol"))]},
                ]
            ),
        )
        self.assertEqual(2, len(self.commands))
        self.assertEqual(rejection("gpt-5.6-sol"), self.last_error())

    def test_capacity_resume_uses_the_fallback_model(self):
        self.args.model = "gpt-6-sol"
        self.args.fallback_model = "gpt-5.6-sol"
        self.write_rollout(thread_id=OTHER)
        rejection = (
            "The 'gpt-6-sol' model is not supported when using Codex "
            "with a ChatGPT account."
        )
        self.assertEqual(
            0,
            self.execute(
                [
                    {"events": [thread_event(), failed(rejection)]},
                    {"events": [thread_event(OTHER), failed()]},
                    {"events": [thread_event(OTHER), completed()], "status": 0},
                ]
            ),
        )
        self.assertEqual([300], self.sleeps)
        self.assertEqual(
            ["gpt-6-sol", "gpt-5.6-sol", "gpt-5.6-sol"],
            [command[command.index("--model") + 1] for command in self.commands],
        )
        self.assertEqual(["resume", OTHER], self.commands[2][-3:-1])
        self.assertEqual(
            "gpt-5.6-sol\n", (self.context / "codex-review-model.txt").read_text()
        )

    def test_unsupported_model_during_resume_never_starts_another_review(self):
        self.args.model = "gpt-6-sol"
        self.args.fallback_model = "gpt-5.6-sol"
        rejection = (
            "The 'gpt-6-sol' model is not supported when using Codex "
            "with a ChatGPT account."
        )
        self.assertEqual(
            1,
            self.execute(
                [
                    {"events": [thread_event(), failed()]},
                    {"events": [thread_event(), failed(rejection)]},
                ]
            ),
        )
        self.assertEqual(2, len(self.commands))
        self.assertEqual([300], self.sleeps)
        self.assertEqual(rejection, self.last_error())

    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([300], self.sleeps)
        self.assertEqual([1000, 695], 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_capacity_fallback_resumes_same_session_after_five_failures(self):
        self.args.model = "gpt-6.1-sol"
        self.args.budget_seconds = 3600
        before = self.ledger.read_text()
        self.assertEqual(0, self.execute(
            [{"events": [thread_event(), failed()], "elapsed": 5}] * 5
            + [{"events": [thread_event(), completed()], "status": 0}]
        ))
        self.assertEqual(
            ["gpt-6.1-sol"] * 5 + ["gpt-6-sol"],
            [c[c.index("--model") + 1] for c in self.commands],
        )
        self.assertEqual([3600, 3295, 2990, 2685, 2380, 2075], self.timeouts)
        self.assertEqual([300] * 5, self.sleeps)
        for command in self.commands[1:]:
            self.assertEqual(["resume", THREAD], command[-3:-1])
            self.assertIn("model_reasoning_effort=xhigh", command)
        self.assertNotIn("Capacity recovery switched", self.commands[4][-1])
        self.assertIn("Capacity recovery switched to gpt-6-sol", self.commands[5][-1])
        self.assertIn("including child agents", self.commands[5][-1])
        self.assertIn("cumulative round count", self.commands[5][-1])
        self.assertEqual(before, self.ledger.read_text())
        self.assertEqual("gpt-6-sol\n", (self.context / "codex-review-model.txt").read_text())
        self.verification.assert_called_once()

    def test_policy_capacity_resumes_same_session_with_shared_budget(self):
        before = self.ledger.read_text()
        message = "Policy-restricted capacity is unavailable. Please try again later."
        self.assertEqual(0, self.execute([
            {"events": [thread_event(), failed(message)], "elapsed": 5},
            {"events": [thread_event(), completed()], "status": 0},
        ]))
        self.assertEqual([300], self.sleeps)
        self.assertEqual([1000, 695], self.timeouts)
        self.assertEqual(["resume", THREAD], self.commands[1][-3:-1])
        self.assertEqual(before, self.ledger.read_text())
        self.target_check.assert_called_once()

    def test_mixed_capacity_errors_share_fallback_and_retry_limit(self):
        self.args.budget_seconds = 3600
        policy_message = "Policy-restricted capacity is unavailable. Please try again later."
        messages = [policy_message, runner.CAPACITY_MESSAGE] * 3 + [policy_message]
        self.assertEqual(1, self.execute([
            {"events": [thread_event(), failed(message)]} for message in messages
        ]))
        self.assertEqual([300] * 6, self.sleeps)
        self.assertEqual(
            [self.args.model] * 5 + [self.args.capacity_fallback_model] * 2,
            [command[command.index("--model") + 1] for command in self.commands],
        )
        for command in self.commands[1:]:
            self.assertEqual(["resume", THREAD], command[-3:-1])
        self.assertEqual(policy_message, self.last_error())

    def test_policy_capacity_error_event_with_zero_exit_resumes(self):
        message = "Policy-restricted capacity is unavailable. Please try again later."
        self.assertEqual(0, self.execute([
            {"events": [thread_event(), {"type": "error", "message": message}], "status": 0},
            {"events": [thread_event(), completed()], "status": 0},
        ]))
        self.assertEqual([300], self.sleeps)

    def test_policy_capacity_after_submission_requires_verified_delivery(self):
        (self.context / runner.SUBMISSION_FILE).write_text("{}")
        message = "Policy-restricted capacity is unavailable. Please try again later."
        self.assertEqual(0, self.execute([{"events": [thread_event(), failed(message)]}]))
        self.verification.assert_called_once()
        self.assertTrue(json.loads((self.context / runner.RESULT_FILE).read_text())["recovered_after_capacity"])
        self.assertEqual([], self.sleeps)
        self.target_check.assert_not_called()

    def test_policy_capacity_with_partial_submission_does_not_resume(self):
        (self.context / runner.SUBMISSION_FILE).write_text("{}")
        self.verification.side_effect = ValueError("Final review inline comments were not completely verified")
        message = "Policy-restricted capacity is unavailable. Please try again later."
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed(message)]}]))
        self.verification.assert_called_once()
        self.assertIn("not completely verified", self.last_error())
        self.assertEqual([], self.sleeps)
        self.assertFalse((self.context / runner.RESULT_FILE).exists())

    def test_policy_capacity_does_not_override_cancellation(self):
        message = "Policy-restricted capacity is unavailable. Please try again later."
        self.assertEqual(1, self.execute([
            {"events": [thread_event(), failed(message)], "status": 143},
        ]))
        self.assertEqual([], self.sleeps)
        self.verification.assert_not_called()

    def test_policy_capacity_does_not_exceed_shared_budget(self):
        self.args.budget_seconds = 300
        message = "Policy-restricted capacity is unavailable. Please try again later."
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed(message)]}]))
        self.assertIn("Insufficient shared review budget", self.last_error())
        self.assertEqual([], self.sleeps)

    def test_policy_capacity_text_in_tool_output_does_not_override_safety_failure(self):
        event = {"type": "item.completed", "item": {
            "type": "command_execution",
            "aggregated_output": "Policy-restricted capacity is unavailable. Please try again later.",
        }}
        self.assertEqual(1, self.execute([
            {"events": [thread_event(), event, failed("Request blocked by safety policy")]},
        ]))
        self.assertEqual("Request blocked by safety policy", self.last_error())
        self.assertEqual([], self.sleeps)

    def test_capacity_fallback_is_optional(self):
        self.args.capacity_fallback_model = None
        self.args.budget_seconds = 3600
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed()]}] * 7))
        self.assertEqual(
            [self.args.model] * 7,
            [c[c.index("--model") + 1] for c in self.commands],
        )

    def test_fifth_attempt_non_capacity_failure_does_not_switch(self):
        self.args.budget_seconds = 3600
        self.assertEqual(1, self.execute(
            [{"events": [thread_event(), failed()]}] * 4
            + [{"events": [thread_event(), failed("HTTP 422")]}]
        ))
        self.assertEqual(5, len(self.commands))
        self.assertEqual("HTTP 422", self.last_error())
        self.assertEqual(self.args.model + "\n", (self.context / "codex-review-model.txt").read_text())

    def test_capacity_fallback_preserves_budget_and_target_checks(self):
        self.args.budget_seconds = 1500
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed()]}] * 5))
        self.assertEqual(5, len(self.commands))
        self.assertEqual([300] * 4, self.sleeps)
        self.assertIn("Insufficient shared review budget", self.last_error())
        self.assertEqual(self.args.model + "\n", (self.context / "codex-review-model.txt").read_text())

    def test_target_change_before_capacity_switch_stops_same_review(self):
        self.args.budget_seconds = 3600
        self.target_check.side_effect = [None] * 4 + [runner.ReviewSuperseded("c" * 40)]
        self.assertEqual(75, self.execute([{"events": [thread_event(), failed()]}] * 5))
        self.assertEqual(5, len(self.commands))
        self.assertEqual(self.args.model + "\n", (self.context / "codex-review-model.txt").read_text())
        self.assert_superseded()

    def test_submission_at_fifth_capacity_is_verified_without_switch(self):
        self.args.budget_seconds = 3600
        read_events = runner.read_events
        def submitted(path):
            if path.name == "5.jsonl":
                (self.context / runner.SUBMISSION_FILE).write_text("{}")
            return read_events(path)
        with mock.patch.object(runner, "read_events", side_effect=submitted):
            self.assertEqual(0, self.execute([{"events": [thread_event(), failed()]}] * 5))
        self.assertEqual(5, len(self.commands))
        self.assertEqual([300] * 4, self.sleeps)
        self.verification.assert_called_once()
        self.assertTrue(json.loads((self.context / runner.RESULT_FILE).read_text())["recovered_after_capacity"])

    def test_unsupported_capacity_fallback_does_not_restart_or_switch_back(self):
        self.args.budget_seconds = 3600
        rejection = "The 'gpt-6-sol' " + runner.UNSUPPORTED_CHATGPT_MODEL
        self.assertEqual(1, self.execute(
            [{"events": [thread_event(), failed()]}] * 5
            + [{"events": [thread_event(), failed(rejection)]}]
        ))
        self.assertEqual(6, len(self.commands))
        self.assertEqual(["resume", THREAD], self.commands[-1][-3:-1])
        self.assertEqual(rejection, self.last_error())

    def test_zero_exit_capacity_can_complete_on_the_sixth_resume(self):
        self.args.budget_seconds = 3600
        self.assertEqual(
            0,
            self.execute(
                [{"events": [thread_event(), failed()]}]
                + [{"events": [thread_event(), failed()], "status": 0}] * 5
                + [{"events": [thread_event(), completed()], "status": 0}]
            ),
        )
        self.assertEqual(7, len(self.commands))
        self.assertEqual([300] * 6, self.sleeps)
        self.assertEqual([3600, 3300, 3000, 2700, 2400, 2100, 1800], self.timeouts)
        self.assertEqual(6, self.target_check.call_count)
        for command in self.commands[1:]:
            self.assertEqual(["resume", THREAD], command[-3:-1])
        retries = [e for e in self.events() if e["type"] == "review.capacity_retry"]
        self.assertEqual([2, 3, 4, 5, 6, 7], [e["next_attempt"] for e in retries])
        self.assertEqual([300] * 6, [e["delay_seconds"] for e in retries])
        self.assertEqual("completed", exporter.latest_turn_result(self.events())[0])

    def test_zero_exit_capacity_stops_at_the_retry_limit(self):
        self.args.budget_seconds = 3600
        self.assertEqual(
            1, self.execute([{"events": [thread_event(), failed()], "status": 0}] * 7)
        )
        self.assertEqual([300] * 6, self.sleeps)
        self.assertEqual(7, len(self.commands))
        self.assertEqual(runner.CAPACITY_MESSAGE, self.last_error())

    def test_zero_exit_error_event_can_identify_capacity(self):
        self.assertEqual(
            0,
            self.execute(
                [
                    {
                        "events": [
                            thread_event(),
                            {"type": "error", "message": runner.CAPACITY_MESSAGE},
                        ],
                        "status": 0,
                    },
                    {"events": [thread_event(), completed()], "status": 0},
                ]
            ),
        )
        self.assertEqual([300], self.sleeps)

    def test_zero_exit_auth_usage_and_generic_failures_do_not_retry(self):
        for message in (
            "refresh_token_reused",
            "You've hit your usage limit.",
            "HTTP 500",
        ):
            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": [thread_event(), failed(message)], "status": 0}]
                    ),
                )
                events = runner.read_events(
                    self.args.context_dir / "codex-events.jsonl"
                )
                self.assertEqual(message, events[-1]["error"]["message"])
        self.assertEqual([], self.sleeps)
        self.target_check.assert_not_called()

    def test_zero_exit_without_terminal_event_fails_closed(self):
        self.assertEqual(1, self.execute([{"events": [thread_event()], "status": 0}]))
        self.assertIn("without a terminal turn event", self.last_error())
        self.assertEqual([], self.sleeps)

    def test_empty_resume_does_not_reuse_the_previous_capacity_error(self):
        self.assertEqual(
            1,
            self.execute(
                [
                    {"events": [thread_event(), failed()]},
                    {"events": [thread_event()], "status": 0},
                ]
            ),
        )
        self.assertEqual(2, len(self.commands))
        self.assertEqual([300], self.sleeps)
        self.assertIn("without a terminal turn event", self.last_error())
        self.assertNotEqual(runner.CAPACITY_MESSAGE, self.last_error())

    def test_later_completion_supersedes_earlier_error_in_the_same_attempt(self):
        self.assertEqual(
            0,
            self.execute(
                [{"events": [thread_event(), failed(), completed()], "status": 0}]
            ),
        )
        self.assertEqual([], self.sleeps)

    def test_later_failure_supersedes_earlier_completion(self):
        self.assertEqual(
            1,
            self.execute(
                [
                    {
                        "events": [thread_event(), completed(), failed("auth failed")],
                        "status": 0,
                    }
                ]
            ),
        )
        self.assertEqual("auth failed", self.last_error())
        self.assertEqual([], self.sleeps)

    def test_unfinished_new_turn_does_not_reuse_an_earlier_completion(self):
        self.assertEqual(
            1,
            self.execute(
                [
                    {
                        "events": [
                            thread_event(),
                            completed(),
                            {"type": "turn.started"},
                        ],
                        "status": 0,
                    }
                ]
            ),
        )
        self.assertIn("without a terminal turn event", self.last_error())
        self.assertEqual([], self.sleeps)

    def test_completion_with_nonzero_exit_does_not_retry_old_stderr_capacity(self):
        self.assertEqual(
            1,
            self.execute(
                [
                    {
                        "events": [thread_event(), completed()],
                        "status": 1,
                        "stderr": runner.CAPACITY_MESSAGE,
                    }
                ]
            ),
        )
        self.assertIn("status 1 after turn.completed", self.last_error())
        self.assertEqual([], self.sleeps)

    def test_retry_count_is_bounded(self):
        self.args.budget_seconds = 3600
        self.assertEqual(1, self.execute([{"events": [thread_event(), failed()]}] * 7))
        self.assertEqual([300] * 6, self.sleeps)
        self.assertEqual(7, len(self.commands))
        self.assertEqual(
            [self.args.model] * 5 + ["gpt-6-sol"] * 2,
            [c[c.index("--model") + 1] for c in 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 = 400
        self.assertEqual(
            1,
            self.execute(
                [
                    {"events": [thread_event(), failed()], "elapsed": 10},
                    {"events": [thread_event(), failed()], "elapsed": 10},
                ]
            ),
        )
        self.assertEqual([400, 90], self.timeouts)
        self.assertEqual([300], 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 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(stderr="", 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(stderr="", stdout=json.dumps({"head": {"sha": "a" * 40}})),
            SimpleNamespace(stderr="", 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(stderr="", 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()


@unittest.skipUnless(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 (
            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 = """python3() {
  if [ "$1" != "$REVIEW_GH_HELPER" ]; then command python3 "$@"; return; fi
  shift
  SECONDS=$((SECONDS + SETUP_SECONDS))
  cat <<'HELPER'
import json, sys
print(json.dumps(sys.argv[1:]))
HELPER
}
"""
        for minutes, setup, expected in (
            (default_minutes, 12, 7056),
            ("150", 30, 8820),
            ("120", 7201, -7322),
        ):
            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)
                self.assertEqual("gpt-6-sol", args[args.index("--capacity-fallback-model") + 1])
                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(stderr="", 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_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(
                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):
        self.assert_real_process_capacity_then_resume(runner.CAPACITY_MESSAGE)

    def test_real_process_policy_capacity_then_resume_preserves_state(self):
        self.assert_real_process_capacity_then_resume(
            "Policy-restricted capacity is unavailable. Please try again later."
        )

    def assert_real_process_capacity_then_resume(self, message):
        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")
            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": os.environ["FAKE_CAPACITY_MESSAGE"]
    }}))
    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,
                        "FAKE_CAPACITY_MESSAGE": message,
                    },
                ),
                mock.patch.object(runner, "RETRY_DELAYS", (0, 0, 0)),
                mock.patch.object(runner, "check_resume_target"),
                mock.patch.object(runner, "verify_completion", return_value={
                    "state": "success", "p0": 0, "p1": 0, "review_id": 123,
                    "converged": True, "rounds": 1, "incomplete_reason": "",
                }),
            ):
                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)
            write_fake_gh(root)
            (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", "superseded"):
            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)
            write_fake_gh(root)
            (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; r.HEAD_POLL_SECONDS = 0.05; "
                    + (
                        """
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 == "superseded":
                        (root / "new-head").write_text("c" * 40)
                    if stop.startswith("cancel"):
                        process.send_signal(signal.SIGTERM)
                    _, stderr = process.communicate(timeout=10)
                    self.assertEqual(
                        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):
                            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 == "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:
                        process.kill()
                    process.communicate(timeout=5)
                    unrelated.kill()


if __name__ == "__main__":
    unittest.main()