codex-app-server/tests/test_v2cov_gaps_d3.py

"""Coverage batch v4-D3: host.py wait/bin/client/run-turn/restore/
merge/notify/pr/dispatch/main branches.

Stubbing discipline: the function under test is never stubbed. Patched
collaborators are: _summarize_trace_item / _append_trace / request /
_wait_turn / interrupt_turn / resume_thread / ensure_alive / _start /
git / run_gh / _github_remote_url / task_diff / open_review_comments /
merge_task / _cmd_verify_audit / daemonize / Server / subprocess.run /
os.remove (sinks, data providers and predicates). Every assertion reads
a real return value, a real file, or real stdout.
"""

import contextlib
import io
import json
import os
import queue
import sqlite3
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT))

import host
from host import (
    Server,
    _wait_turn,
    create_pr,
    default_codex_bin,
    dispatch_from_branch,
    merge_task,
    notify_failure,
    parse_pr_checks,
    restore_state,
)

GIT_ENV = {
    "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
    "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t",
    "GIT_CONFIG_NOSYSTEM": "1",
}


def _server():
    srv = object.__new__(Server)
    srv.proc = None
    srv.codex_bin = "codex"
    srv.codex_home = None
    srv.approval = "deny"
    srv.model_provider = None
    srv.progress_interval = 30
    srv.send_lock = threading.Lock()
    srv.next_id = 1
    srv.pending = {}
    srv.pending_lock = threading.Lock()
    srv.turn_queues = {}
    srv.events = []
    srv.events_lock = threading.Lock()
    srv.restart_count = 0
    srv.trace_path = None
    srv._trace_lines = 0
    return srv


class _FakeServer:
    def __init__(self, *args, **kwargs):
        self.closed = False

    def initialize(self):
        pass

    def close(self):
        self.closed = True

    def resume_thread(self, thread_id):
        pass

    def interrupt_turn(self, thread_id, turn_id):
        return {"ok": True}


class WaitAndBinTestCase(unittest.TestCase):
    def test_wait_turn_tiny_timeout(self):
        # 154: remaining <= 0 on the first slice.
        q = queue.Queue()
        self.assertEqual(_wait_turn(q, 1e-5), ("timeout", None))

    def test_default_codex_bin_paired(self):
        # 204
        with tempfile.TemporaryDirectory() as tmp:
            os.makedirs(os.path.join(tmp, "bin"))
            paired = os.path.join(tmp, "bin", "codex")
            with open(paired, "w", encoding="utf-8") as f:
                f.write("x")
            with mock.patch.object(host, "HERE", tmp):
                with mock.patch.dict(os.environ, {}, clear=False):
                    os.environ.pop("CODEX_BIN", None)
                    self.assertEqual(default_codex_bin(), paired)


class ServerClientTestCase(unittest.TestCase):
    def test_ensure_alive_true_and_false(self):
        # 456, 462-463
        srv = _server()
        srv.proc = mock.Mock()
        srv.proc.poll.return_value = None
        self.assertTrue(srv.ensure_alive())
        srv2 = _server()
        srv2.proc = None
        srv2._start = mock.Mock(side_effect=RuntimeError("start down"))
        self.assertFalse(srv2.ensure_alive())

    def test_read_loop_non_dict_stderr_failure(self):
        # 501-502
        srv = _server()
        srv.proc = SimpleNamespace(stdout=[b"123\n"])
        bad_err = mock.Mock()
        bad_err.write.side_effect = RuntimeError("stderr down")
        with mock.patch.object(sys, "stderr", bad_err):
            srv._read_loop()
        bad_err.write.assert_called()

    def test_read_loop_append_trace_failure(self):
        # 562-565
        srv = _server()
        srv.trace_path = os.path.join(
            tempfile.mkdtemp(), "trace.jsonl")
        srv.proc = SimpleNamespace(stdout=[
            b'{"jsonrpc":"2.0","method":"agent/message","params":{}}\n'])
        srv._append_trace = mock.Mock(
            side_effect=RuntimeError("trace down"))
        srv._read_loop()
        self.assertEqual(len(srv.events), 1)

    def test_read_loop_events_append_failure(self):
        # 570-571: outer except of the read loop.
        srv = _server()
        srv.proc = SimpleNamespace(stdout=[
            b'{"jsonrpc":"2.0","method":"agent/message","params":{}}\n'])
        srv.events = mock.Mock()
        srv.events.append.side_effect = RuntimeError("events down")
        srv._read_loop()

    def test_request_broken_pipe_retry_then_raise(self):
        # 596-600
        srv = _server()
        srv.proc = SimpleNamespace(
            stdin=mock.Mock())
        srv.proc.stdin.write.side_effect = BrokenPipeError("pipe down")
        srv.ensure_alive = mock.Mock(return_value=True)
        with self.assertRaises(BrokenPipeError):
            srv.request("initialize", {}, timeout=60)
        self.assertEqual(srv.ensure_alive.call_count, 1)

    def test_thread_method_error_branches(self):
        # 627, 639, 651, 662, 673
        srv = _server()
        srv.request = mock.Mock(return_value={"error": "boom"})
        with self.assertRaises(RuntimeError):
            srv.create_thread("/tmp")
        with self.assertRaises(RuntimeError):
            srv.list_threads()
        with self.assertRaises(RuntimeError):
            srv.resume_thread("th1")
        with self.assertRaises(RuntimeError):
            srv.start_review("th1", "target")
        with self.assertRaises(RuntimeError):
            srv.fork_thread("th1")
        with self.assertRaises(RuntimeError):
            srv.archive_thread("th1")

    def test_fork_thread_non_dict_result_none(self):
        # 668
        srv = _server()
        srv.request = mock.Mock(return_value={"result": "not-a-dict"})
        self.assertIsNone(srv.fork_thread("th1"))

    def test_append_trace_break_at_cap(self):
        # 430
        srv = _server()
        tmp = tempfile.mkdtemp()
        srv.trace_path = os.path.join(tmp, "trace.jsonl")
        srv._trace_lines = host.TRACE_MAX_LINES - 1
        with mock.patch.object(
                host, "_summarize_trace_item",
                return_value=[{"a": 1}, {"b": 2}]):
            srv._append_trace({"method": "x", "params": {}})
        with open(srv.trace_path, encoding="utf-8") as f:
            self.assertEqual(len(f.readlines()), 1)

    def test_turn_error_empty_message_none(self):
        # 891
        srv = _server()
        home = tempfile.mkdtemp()
        srv.codex_home = home
        db_path = os.path.join(home, "thread_history_1.sqlite")
        conn = sqlite3.connect(db_path)
        try:
            conn.execute(
                "CREATE TABLE thread_turns(thread_id TEXT, turn_id TEXT,"
                " error_json TEXT)")
            conn.execute(
                "INSERT INTO thread_turns VALUES(?, ?, ?)",
                ("th1", "turn1", '{"message": ""}'))
            conn.commit()
        finally:
            conn.close()
        self.assertIsNone(srv._turn_error_from_history("th1", "turn1"))

    def test_close_stdin_stdout_failures(self):
        # 899-900, 907-908
        srv = _server()
        proc = mock.Mock()
        proc.stdin.close.side_effect = RuntimeError("stdin down")
        proc.wait.return_value = 0
        proc.stdout.close.side_effect = RuntimeError("stdout down")
        srv.proc = proc
        srv.close()


class RunTurnTestCase(unittest.TestCase):
    def _client(self):
        srv = _server()
        srv.request = mock.Mock(return_value={
            "result": {"turn": {"id": "turn1"}}})
        return srv

    def test_resume_failure_swallowed(self):
        # 683-687
        srv = self._client()
        srv.proc = None
        srv.ensure_alive = mock.Mock(return_value=True)
        srv.resume_thread = mock.Mock(
            side_effect=RuntimeError("resume down"))
        with mock.patch.object(host, "_wait_turn",
                               return_value=("completed", {
                                   "params": {"turn": {"items": []}}})):
            out = srv.run_turn("th1", "do work")
        self.assertEqual(out["status"], "completed")

    def test_progress_success_and_failure(self):
        # 735-753
        srv = self._client()
        tmp = tempfile.mkdtemp()
        srv.trace_path = os.path.join(tmp, "trace.jsonl")
        srv._trace_lines = 0

        def _fake_wait(q, timeout, progress_interval=None,
                       on_progress=None, interrupt_check=None):
            on_progress(5)
            srv.trace_path = "/nonexistent-dir-xyz/trace.jsonl"
            on_progress(6)
            return ("completed", {"params": {"turn": {"items": []}}})

        with mock.patch.object(host, "_wait_turn",
                               side_effect=_fake_wait):
            out = srv.run_turn("th1", "do work")
        self.assertEqual(out["status"], "completed")
        with open(os.path.join(tmp, "trace.jsonl"),
                  encoding="utf-8") as f:
            self.assertEqual(len(f.readlines()), 1)

    def test_interrupted_error_and_remove_failure(self):
        # 773-780, 784-785
        srv = self._client()
        srv.interrupt_turn = mock.Mock(
            return_value={"error": "nope"})
        with tempfile.TemporaryDirectory() as tmp:
            interrupt_file = os.path.join(tmp, "req.json")
            with open(interrupt_file, "w", encoding="utf-8") as f:
                f.write("{}")
            with mock.patch.object(host, "_wait_turn",
                                   return_value=("interrupted-requested",
                                                 None)):
                with mock.patch("os.remove",
                                side_effect=OSError("remove down")):
                    out = srv.run_turn("th1", "do work",
                                       interrupt_file=interrupt_file)
        self.assertEqual(out["status"], "interrupted")

    def test_wait_raises_empty_timeout(self):
        # 799-802
        srv = self._client()
        with mock.patch.object(host, "_wait_turn",
                               side_effect=queue.Empty()):
            out = srv.run_turn("th1", "do work")
        self.assertEqual(out["status"], "timeout")

    def test_fallback_final_answer_from_events(self):
        # 822
        srv = self._client()

        def _fake_wait(q, timeout, progress_interval=None,
                       on_progress=None, interrupt_check=None):
            srv.events.append({"params": {"item": {
                "type": "agentMessage", "phase": "final_answer",
                "text": "hello"}}})
            return ("completed", {"params": {"turn": {"items": []}}})

        with mock.patch.object(host, "_wait_turn",
                               side_effect=_fake_wait):
            out = srv.run_turn("th1", "do work")
        self.assertEqual(out["text"], "hello")


class RestoreMergeNotifyTestCase(unittest.TestCase):
    def test_restore_corrupt_state_from_bak(self):
        # Corrupt main file -> renamed to .corrupt.* and restored from
        # .bak. (The missing-file else branch at 1046-1047 is
        # unreachable by construction: load_state_strict returns
        # (empty state, None) for a missing file, which takes the
        # early "state 完好" return; recorded in the gap ledger.)
        with tempfile.TemporaryDirectory() as tmp:
            state_path = os.path.join(tmp, "host-state.json")
            with open(state_path, "w", encoding="utf-8") as f:
                f.write("{not json")
            with open(state_path + ".bak", "w", encoding="utf-8") as f:
                json.dump({"version": 3, "tasks": {}}, f)
            ok, msg = restore_state(state_path)
            self.assertTrue(ok)
            self.assertIn(".bak", msg)
            with open(state_path, encoding="utf-8") as f:
                self.assertEqual(json.load(f)["version"], 3)

    def _git_result(self, returncode=0, stdout="", stderr=""):
        return SimpleNamespace(returncode=returncode, stdout=stdout,
                               stderr=stderr)

    def test_merge_commit_failure(self):
        # 2019
        def _git(args, cwd):
            if args[0] == "commit":
                return self._git_result(1, "", "boom")
            if args[0] == "rev-parse":
                return self._git_result(0, "abc\n", "")
            return self._git_result(0, "", "")

        with mock.patch.object(host, "git", side_effect=_git):
            ok, msg = merge_task("/tmp", "/tmp", "b", "t",
                                 verify_argv=[])
        self.assertFalse(ok)
        self.assertIn("提交失败", msg)

    def test_merge_retry_abort(self):
        # 2032-2033
        calls = []

        def _git(args, cwd):
            calls.append(args)
            if args[0] == "rev-parse":
                return self._git_result(0, "abc\n", "")
            if args[0] == "merge" and "-X" not in args:
                return self._git_result(1, "", "conflict")
            if args[0] == "merge" and "-X" in args:
                return self._git_result(1, "", "still conflict")
            return self._git_result(0, "", "")

        with mock.patch.object(host, "git", side_effect=_git):
            ok, msg = merge_task("/tmp", "/tmp", "b", "t",
                                 verify_argv=[])
        self.assertFalse(ok)
        self.assertIn("已 abort", msg)

    def test_merge_missing_lines_pyc_and_broken_link(self):
        # 2048, 2055-2056
        with tempfile.TemporaryDirectory() as tmp:
            repo = os.path.join(tmp, "repo")
            os.makedirs(repo)
            with open(os.path.join(repo, "skip.pyc"), "wb") as f:
                f.write(b"\x00")
            os.symlink("/nonexistent-target-xyz",
                       os.path.join(repo, "broken.txt"))
            with open(os.path.join(repo, "real.txt"), "w",
                      encoding="utf-8") as f:
                f.write("hello\n")

            def _git(args, cwd):
                if args[0] == "rev-parse":
                    return self._git_result(0, "abc\n", "")
                if args[0] == "merge" and "-X" not in args:
                    return self._git_result(1, "", "conflict")
                if args[0] == "merge" and "-X" in args:
                    return self._git_result(0, "", "")
                if args[0] == "merge-base":
                    return self._git_result(0, "base\n", "")
                if args[0] == "diff":
                    return self._git_result(0, "+zzz-not-present-xyz\n",
                                            "")
                return self._git_result(0, "", "")

            with mock.patch.object(host, "git", side_effect=_git):
                ok, msg = merge_task(repo, os.path.join(tmp, "wt"),
                                     "b", "t", verify_argv=[])
        self.assertFalse(ok)
        self.assertIn("改动被丢弃", msg)

    def test_notify_timeout_and_generic_failure(self):
        # 2191, 2197-2198
        with tempfile.TemporaryDirectory() as tmp:
            state_path = os.path.join(tmp, "host-state.json")
            cfg = {"notify_command": "echo {task} {status} {text}"}
            with mock.patch("subprocess.run",
                           side_effect=subprocess.TimeoutExpired(
                               "echo", 30)):
                msg = notify_failure(state_path, cfg, "t1",
                                     {"status": "failed", "text": "x"})
            self.assertTrue(msg.startswith("FAILED t1"))
            with mock.patch("subprocess.run",
                           side_effect=RuntimeError("run down")):
                msg = notify_failure(state_path, cfg, "t1",
                                     {"status": "failed", "text": "x"})
            self.assertTrue(msg.startswith("FAILED t1"))

    def test_create_pr_push_and_gh_failures(self):
        # 2322, 2331
        task = {"name": "t1", "branch": "host/t1",
                "status": "completed"}
        with mock.patch.object(host, "_github_remote_url",
                               return_value="https://example.com/r.git"):
            with mock.patch.object(
                    host, "git",
                    return_value=self._git_result(1, "", "push boom")):
                with self.assertRaises(RuntimeError):
                    create_pr("/tmp", dict(task))
            with mock.patch.object(
                    host, "git",
                    return_value=self._git_result(0, "", "")):
                with mock.patch.object(
                        host, "run_gh",
                        return_value=self._git_result(1, "", "gh boom")):
                    with self.assertRaises(RuntimeError):
                        create_pr("/tmp", dict(task))

    def test_parse_pr_checks_short_tab_line(self):
        # 2356
        checks, all_pass = parse_pr_checks("build\t\n")
        self.assertEqual(checks, [])
        self.assertFalse(all_pass)

    def test_dispatch_from_branch_worktree_failure(self):
        # 2453
        with tempfile.TemporaryDirectory() as tmp:
            with mock.patch.object(host, "HERE", tmp):
                def _git(args, cwd):
                    if args[0] == "worktree":
                        return self._git_result(1, "", "wt fail")
                    return self._git_result(0, "", "")

                with mock.patch.object(host, "git", side_effect=_git):
                    out, err = dispatch_from_branch(
                        SimpleNamespace(), {"tasks": {}}, "t1",
                        "do work", tmp, "main", tmp, None, None,
                        "workspace-write", "never")
        self.assertIsNone(out)
        self.assertIn("失败", err)


class MainBranchTestCase(unittest.TestCase):
    def _state_cfg(self, tmp, tasks):
        state_path = os.path.join(tmp, "host-state.json")
        cfg_path = os.path.join(tmp, "host-config.json")
        with open(state_path, "w", encoding="utf-8") as f:
            json.dump({"tasks": tasks, "automations": []}, f)
        with open(cfg_path, "w", encoding="utf-8") as f:
            json.dump({}, f)
        return state_path, cfg_path

    def _run_main(self, argv, patches=()):
        with contextlib.ExitStack() as stack:
            for p in patches:
                stack.enter_context(p)
            buf = io.StringIO()
            with contextlib.redirect_stdout(buf):
                with mock.patch.object(sys, "argv", argv):
                    host.main()
            return buf.getvalue()

    def test_pr_missing_repo_exits(self):
        # 3292-3293
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(
                tmp, {"t1": {"name": "t1"}})
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "pr", "--task", "t1"]
            with self.assertRaises(SystemExit) as cm:
                self._run_main(argv)
            self.assertEqual(cm.exception.code, 1)

    def test_daemon_status_bad_pid(self):
        # 3376-3377
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(tmp, {})
            with open(os.path.join(tmp, "host-daemon.pid"), "w",
                      encoding="utf-8") as f:
                f.write("abc")
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "daemon-status"]
            out = self._run_main(argv)
            self.assertFalse(json.loads(out)["alive"])

    def test_daemon_status_log_is_dir(self):
        # 3383-3384
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(tmp, {})
            with open(os.path.join(tmp, "host-daemon.pid"), "w",
                      encoding="utf-8") as f:
                f.write(str(os.getpid()))
            os.makedirs(os.path.join(tmp, "host-daemon.log"))
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "daemon-status"]
            out = self._run_main(argv)
            data = json.loads(out)
            self.assertTrue(data["alive"])
            self.assertEqual(data["log_tail"], "")

    def test_daemon_stop_inner_break(self):
        # 3407-3408
        proc = subprocess.Popen(["sleep", "30"])
        try:
            with tempfile.TemporaryDirectory() as tmp:
                state_path, cfg_path = self._state_cfg(tmp, {})
                with open(os.path.join(tmp, "host-daemon.pid"), "w",
                          encoding="utf-8") as f:
                    f.write(str(proc.pid))
                argv = ["host.py", "--state", state_path,
                        "--config", cfg_path, "daemon-stop"]
                out = self._run_main(argv)
                self.assertIn("daemon 已停", out)
                self.assertFalse(os.path.exists(
                    os.path.join(tmp, "host-daemon.pid")))
        finally:
            try:
                proc.kill()
            except Exception:
                pass
            proc.wait()

    def test_daemon_stop_outer_except(self):
        # 3411-3412
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(tmp, {})
            pidfile = os.path.join(tmp, "host-daemon.pid")
            with open(pidfile, "w", encoding="utf-8") as f:
                f.write("999999")
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "daemon-stop"]
            out = self._run_main(argv)
            self.assertIn("daemon 已停", out)
            self.assertFalse(os.path.exists(pidfile))

    def test_daemon_stop_remove_failure(self):
        # 3415-3416
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(tmp, {})
            with open(os.path.join(tmp, "host-daemon.pid"), "w",
                      encoding="utf-8") as f:
                f.write("999999")
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "daemon-stop"]
            out = self._run_main(
                argv, patches=[mock.patch(
                    "os.remove", side_effect=OSError("remove down"))])
            self.assertIn("daemon 已停", out)

    def test_daemon_stale_pid_daemonize_false(self):
        # 3431-3435
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(tmp, {})
            with open(os.path.join(tmp, "host-daemon.pid"), "w",
                      encoding="utf-8") as f:
                f.write("999999")
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "daemon"]
            out = self._run_main(
                argv, patches=[mock.patch.object(
                    host, "daemonize", return_value=False)])
            self.assertIn("daemon 已在后台启动", out)

    def test_verify_argv_empty_errors(self):
        # 3524
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(
                tmp, {"t1": {"name": "t1", "repo": "/tmp",
                             "branch": "b", "cwd": tmp}})
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "merge", "--task", "t1",
                    "--verify-argv"]
            with self.assertRaises(SystemExit) as cm:
                self._run_main(argv)
            self.assertEqual(cm.exception.code, 2)

    def test_verify_audit_dispatch(self):
        # 3526-3527
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(tmp, {})
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "verify-audit"]
            audit_mock = mock.Mock()
            self._run_main(
                argv, patches=[mock.patch.object(
                    host, "_cmd_verify_audit", audit_mock)])
            audit_mock.assert_called_once()

    def test_status_untracked_contents(self):
        # 3601
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(
                tmp, {"t1": {"name": "t1", "cwd": tmp}})
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "review", "--task", "t1"]
            out = self._run_main(argv, patches=[
                mock.patch.object(host, "Server",
                                  lambda **kw: _FakeServer()),
                mock.patch.object(host, "task_diff", return_value={
                    "status": "S", "untracked": "U", "diff": "D",
                    "untracked_contents": {"f.txt": "hello"}}),
                mock.patch.object(host, "open_review_comments",
                                  return_value=[]),
            ])
            self.assertIn("--- untracked file: f.txt ---", out)
            self.assertIn("hello", out)

    def test_merge_verify_argv_warning(self):
        # 3620
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(
                tmp, {"t1": {"name": "t1", "repo": "/tmp",
                             "branch": "b", "cwd": tmp}})
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "merge", "--task", "t1",
                    "--verify", "echo hi", "--verify-argv", "echo", "hi"]
            out = self._run_main(argv, patches=[
                mock.patch.object(host, "Server",
                                  lambda **kw: _FakeServer()),
                mock.patch.object(host, "merge_task",
                                  return_value=(True, "ok")),
            ])
            self.assertIn("verify_argv 已指定", out)
            with open(state_path, encoding="utf-8") as f:
                self.assertEqual(json.load(f)["tasks"]["t1"]["status"],
                                 "merged")

    def test_fix_review_no_comments(self):
        # 3637-3638
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(
                tmp, {"t1": {"name": "t1", "thread_id": "th1"}})
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "fix-review", "--task", "t1"]
            out = self._run_main(argv, patches=[
                mock.patch.object(host, "Server",
                                  lambda **kw: _FakeServer()),
                mock.patch.object(host, "open_review_comments",
                                  return_value=[]),
            ])
            self.assertIn("没有未处理的评审评论", out)

    def test_interrupt_remove_failure(self):
        # 3929-3931
        with tempfile.TemporaryDirectory() as tmp:
            state_path, cfg_path = self._state_cfg(tmp, {
                "t1": {"name": "t1", "thread_id": "th1",
                       "turns": [{"result": {"turn_id": "turn9"}}]}})
            req_dir = os.path.join(tmp, "interrupt-requests")
            os.makedirs(req_dir)
            with open(os.path.join(req_dir, "t1.json"), "w",
                      encoding="utf-8") as f:
                f.write("{}")
            argv = ["host.py", "--state", state_path,
                    "--config", cfg_path, "interrupt", "--task", "t1"]
            out = self._run_main(argv, patches=[
                mock.patch.object(host, "Server",
                                  lambda **kw: _FakeServer()),
                mock.patch("os.remove",
                           side_effect=OSError("remove down")),
            ])
            self.assertIn("turn=turn9", out)
            with open(state_path, encoding="utf-8") as f:
                self.assertEqual(json.load(f)["tasks"]["t1"]["status"],
                                 "interrupted")


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