"""派发队列与沙箱执行器的单测。

运行（在 ~/workspace/codex-app-server 下）：
    python3 -m unittest discover -s cloud/tests -p "test_queue_sandbox.py" -v

说明：cloud/ 没有 __init__.py（协调员统一创建），因此这里把 cloud/
目录插到 sys.path 前面再导入 queue / sandbox，与 discover 的
top-level 目录无关。
"""

import json
import os
import sys
import unittest
from types import SimpleNamespace
from unittest import mock

sys.path.insert(0, os.path.dirname(os.path.dirname(
    os.path.abspath(__file__))))

from queue import TaskQueue, budget_exceeded, order_tasks, ready_tasks
from sandbox import (CloudExecutor, DockerExecutor, Executor,
                     LocalExecutor, _parse_goal_output)


def _task(task_id, depends_on=None, priority=0, status="queued"):
    task = {"id": task_id, "priority": priority, "status": status}
    if depends_on is not None:
        task["depends_on"] = depends_on
    return task


class TestOrderTasks(unittest.TestCase):
    def test_dependency_order(self):
        # A -> B 时 B 在 A 后
        ordered = order_tasks([_task("B", depends_on=["A"]),
                               _task("A")])
        self.assertEqual([t["id"] for t in ordered], ["A", "B"])

    def test_dependency_chain(self):
        ordered = order_tasks([_task("C", depends_on=["B"]),
                               _task("B", depends_on=["A"]),
                               _task("A")])
        self.assertEqual([t["id"] for t in ordered], ["A", "B", "C"])

    def test_priority_within_wave(self):
        # 同一波（均无依赖）按 priority 大的先
        ordered = order_tasks([_task("low", priority=1),
                               _task("high", priority=10),
                               _task("mid", priority=5)])
        self.assertEqual([t["id"] for t in ordered],
                         ["high", "mid", "low"])

    def test_priority_does_not_jump_dependency(self):
        # 高优先级也不能越过依赖：B 依赖 A，即使 B 优先级更高也排 A 后
        ordered = order_tasks([_task("B", depends_on=["A"], priority=99),
                               _task("A", priority=0)])
        self.assertEqual([t["id"] for t in ordered], ["A", "B"])

    def test_missing_dependency_raises(self):
        with self.assertRaises(ValueError):
            order_tasks([_task("B", depends_on=["NOPE"])])

    def test_cycle_raises(self):
        with self.assertRaises(ValueError):
            order_tasks([_task("A", depends_on=["B"]),
                         _task("B", depends_on=["A"])])

    def test_self_dependency_raises(self):
        with self.assertRaises(ValueError):
            order_tasks([_task("A", depends_on=["A"])])

    def test_defaults(self):
        # depends_on / priority 缺省时按 0 处理、不报错
        ordered = order_tasks([{"id": "A"}, {"id": "B"}])
        self.assertEqual([t["id"] for t in ordered], ["A", "B"])


class TestReadyTasks(unittest.TestCase):
    def test_only_dependency_satisfied_queued(self):
        tasks = [_task("A"),
                 _task("B", depends_on=["A"]),
                 _task("C", depends_on=["A"], status="running"),
                 _task("D", depends_on=["MISSING"])]
        ready = ready_tasks(tasks, completed_ids=[])
        self.assertEqual([t["id"] for t in ready], ["A"])

        # A 已完成（状态随之改为 completed）后，B 依赖满足且 queued；
        # C 是 running 不返回；D 依赖未满足不返回
        done_tasks = [_task("A", status="completed"),
                      _task("B", depends_on=["A"]),
                      _task("C", depends_on=["A"], status="running"),
                      _task("D", depends_on=["MISSING"])]
        ready = ready_tasks(done_tasks, completed_ids=["A"])
        self.assertEqual([t["id"] for t in ready], ["B"])

    def test_ready_ordered_by_priority(self):
        tasks = [_task("low", priority=1), _task("high", priority=9)]
        ready = ready_tasks(tasks, completed_ids=[])
        self.assertEqual([t["id"] for t in ready], ["high", "low"])


class TestBudgetExceeded(unittest.TestCase):
    def test_below_limits(self):
        self.assertFalse(budget_exceeded(5, 10, 0.5, 1.0))

    def test_at_limit_counts_as_exceeded(self):
        self.assertTrue(budget_exceeded(10, 10, 0.5, 1.0))
        self.assertTrue(budget_exceeded(5, 10, 1.0, 1.0))

    def test_none_limit_means_unlimited(self):
        self.assertFalse(budget_exceeded(10 ** 9, None, 10 ** 9, None))
        self.assertTrue(budget_exceeded(10 ** 9, None, 10 ** 9, 1.0))


class TestTaskQueue(unittest.TestCase):
    def test_next_ready_flow(self):
        queue = TaskQueue()
        queue.add(_task("A"))
        queue.add(_task("B", depends_on=["A"]))
        self.assertEqual(queue.next_ready([])["id"], "A")
        queue.mark_running("A")
        # A 在 running，不再是可派任务；B 依赖未完成，也不可派
        self.assertIsNone(queue.next_ready([]))
        queue.mark_done("A", "completed")
        self.assertEqual(queue.next_ready(["A"])["id"], "B")

    def test_mark_done_validates_status(self):
        queue = TaskQueue([_task("A")])
        with self.assertRaises(ValueError):
            queue.mark_done("A", "bogus-status")
        with self.assertRaises(ValueError):
            queue.add(_task("A"))  # 重复 id

    def test_all_host_statuses_accepted(self):
        queue = TaskQueue()
        for status in ("running", "completed", "failed",
                       "interrupted", "merged", "merge-failed"):
            queue.add(_task("t-" + status, status=status))
        # 非 queued 任务一律不可派
        self.assertIsNone(queue.next_ready([]))


class TestExecutors(unittest.TestCase):
    def test_base_executor_must_implement(self):
        with self.assertRaises(NotImplementedError):
            Executor().execute({"id": "x"})

    def test_local_executor_success(self):
        seen = []

        def fake_runner(cmd):
            seen.append(cmd)
            return SimpleNamespace(returncode=0, stdout="ok-status")

        executor = LocalExecutor("/fake/host.py", config_path="/fake/cfg.json",
                                 runner=fake_runner)
        result = executor.execute({"id": "t1", "project_id": "p",
                                   "repo": "/r", "prompt": "hi"})
        self.assertEqual(result["executor"], "local")
        self.assertEqual(result["task_id"], "t1")
        self.assertEqual(result["status"], "completed")
        self.assertEqual(result["stdout"], "ok-status")
        # 只发只读 status 命令，不发 dispatch/run（不直接跑模型）
        self.assertIn("status", seen[0])
        self.assertNotIn("dispatch", seen[0])
        self.assertNotIn("run", seen[0])
        self.assertIn("--config", seen[0])

    def test_local_executor_failure(self):
        def fake_runner(cmd):
            return SimpleNamespace(returncode=1, stdout="boom")

        result = LocalExecutor("/fake/host.py",
                               runner=fake_runner).execute({"id": "t2"})
        self.assertEqual(result["status"], "failed")
        self.assertEqual(result["stdout"], "boom")

    def test_local_executor_truncates_stdout(self):
        def fake_runner(cmd):
            return SimpleNamespace(returncode=0, stdout="x" * 5000)

        result = LocalExecutor("/fake/host.py",
                               runner=fake_runner).execute({"id": "t3"})
        self.assertEqual(len(result["stdout"]), 2000)

    def test_docker_executor_no_docker_raises_not_local(self):
        with mock.patch("sandbox.shutil.which", return_value=None):
            with self.assertRaises(RuntimeError) as ctx:
                DockerExecutor().execute({"id": "t1"})
            self.assertIn("docker not available", str(ctx.exception))
            # 报错而不是返回 local 结果：绝不静默降级为本地执行

    def test_docker_executor_with_docker_runs_container(self):
        seen = []

        def fake_runner(cmd, timeout=None):
            seen.append(cmd)
            return SimpleNamespace(returncode=0, stdout="ok-status")

        with mock.patch("sandbox.shutil.which",
                        return_value="/usr/bin/docker"):
            executor = DockerExecutor(
                "/fake/host.py", runner=fake_runner,
                env={"XCM_API_KEY": "placeholder-not-a-real-key"})
            result = executor.execute({"id": "t1", "worktree": "/tmp/wt1"})
        self.assertEqual(result["executor"], "docker")
        self.assertEqual(result["status"], "completed")
        self.assertEqual(seen[0][0:2], ["docker", "run"])
        self.assertIn("--rm", seen[0])
        self.assertIn("/tmp/wt1:/work", seen[0])
        self.assertIn("/fake:/app:ro", seen[0])
        # -e 只传名，值绝不出现在命令列表里。
        self.assertIn("XCM_API_KEY", seen[0])
        self.assertNotIn("placeholder-not-a-real-key", seen[0])
        self.assertNotIn("placeholder-not-a-real-key", " ".join(seen[0]))

    def test_cloud_executor_raises_with_cloud_account(self):
        with self.assertRaises(RuntimeError) as ctx:
            CloudExecutor().execute({"id": "t1"})
        self.assertIn("cloud account", str(ctx.exception))


class TestGoalExecutors(unittest.TestCase):
    def test_parse_goal_output(self):
        met, err = _parse_goal_output(
            "round 1: turn=completed verify_rc=0 verify_out=ok\n"
            "goal met after rounds\n")
        self.assertTrue(met)
        self.assertIsNone(err)
        met, err = _parse_goal_output(
            "round 1: turn=completed verify_rc=1 verify_out=file missing\n"
            "round 2: turn=completed verify_rc=1 verify_out=file missing\n"
            "goal unmet after rounds\n")
        self.assertFalse(met)
        self.assertIn("file missing", err)
        self.assertIn("2 rounds", err)
        met, err = _parse_goal_output("some other output\n")
        self.assertIsNone(met)
        met, err = _parse_goal_output("")
        self.assertIsNone(met)

    def test_local_goal_task_met(self):
        seen = []

        def fake(cmd):
            seen.append(cmd)
            return SimpleNamespace(
                returncode=0,
                stdout="round 1: turn=completed verify_rc=0 verify_out=ok\n"
                       "goal met after rounds\n",
                stderr="")

        executor = LocalExecutor("/fake/host.py", runner=fake)
        result = executor.goal_task(
            {"id": "t1", "prompt": "do it", "verify_command": "test -f x",
             "goal_max_rounds": 2, "repo": "/r"})
        self.assertEqual(result["status"], "completed")
        self.assertTrue(result["goal_met"])
        self.assertIsNone(result["goal_error"])
        self.assertEqual(result["host_status"], "goal-met")
        self.assertIn("goal", seen[0])
        self.assertIn("--verify", seen[0])
        self.assertIn("test -f x", seen[0])
        self.assertIn("/r", seen[0])
        self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "2")

    def test_local_goal_task_unmet(self):
        def fake(cmd):
            return SimpleNamespace(
                returncode=0,
                stdout="round 1: turn=completed verify_rc=1 "
                       "verify_out=file missing\n"
                       "goal unmet after rounds\n",
                stderr="")

        result = LocalExecutor("/fake/host.py",
                               runner=fake).goal_task(
            {"id": "t1", "prompt": "do it",
             "verify_command": "test -f x"})
        self.assertEqual(result["status"], "failed")
        self.assertFalse(result["goal_met"])
        self.assertIn("file missing", result["goal_error"])
        self.assertEqual(result["host_status"], "goal-unmet")

    def test_local_goal_task_defaults_prompt_and_rounds(self):
        seen = []

        def fake(cmd):
            seen.append(list(cmd))
            return SimpleNamespace(
                returncode=0, stdout="goal met after rounds\n", stderr="")

        LocalExecutor("/fake/host.py", runner=fake).goal_task(
            {"id": "t1", "prompt": "do it", "verify_command": "true"})
        self.assertIn("do it", seen[0])
        self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "3")

    def test_local_goal_task_no_conclusion_is_failed_not_completed(self):
        def fake(cmd):
            return SimpleNamespace(returncode=0, stdout="nothing\n",
                                   stderr="")

        result = LocalExecutor("/fake/host.py",
                               runner=fake).goal_task(
            {"id": "t1", "prompt": "do it",
             "verify_command": "true"})
        self.assertEqual(result["status"], "failed")
        self.assertIsNone(result["goal_met"])
        self.assertIn("无结论", result["goal_error"])

    def test_local_goal_task_nonzero_without_conclusion_is_interrupted(self):
        # 2026-10-28 长 goal 实弹：无结论的异常退出（进程被杀、
        # 连接断后恢复时抛错）必须记 interrupted 保住断点续跑，
        # 绝不能一律 failed（旧语义直接断了 60 分钟工作的活路）。
        def fake(cmd):
            return SimpleNamespace(returncode=1, stdout="", stderr="boom")

        result = LocalExecutor("/fake/host.py",
                               runner=fake).goal_task(
            {"id": "t1", "prompt": "do it",
             "verify_command": "true"})
        self.assertEqual(result["status"], "interrupted")
        self.assertIsNone(result["goal_met"])
        self.assertIn("boom", result["goal_error"])

    def test_local_goal_task_nonzero_with_unmet_stays_failed(self):
        def fake(cmd):
            return SimpleNamespace(
                returncode=1,
                stdout="round 1: turn=completed verify_rc=1 "
                       "verify_out=file missing\n"
                       "goal unmet after rounds\n",
                stderr="")

        result = LocalExecutor("/fake/host.py",
                               runner=fake).goal_task(
            {"id": "t1", "prompt": "do it",
             "verify_command": "true"})
        self.assertEqual(result["status"], "failed")
        self.assertIs(result["goal_met"], False)

    def test_local_goal_task_interrupted_marker(self):
        def fake(cmd):
            return SimpleNamespace(
                returncode=0,
                stdout="round 1: turn=interrupted verify_rc=- "
                       "verify_out=\n"
                       "goal interrupted after rounds: relay boom\n",
                stderr="")

        result = LocalExecutor("/fake/host.py",
                               runner=fake).goal_task(
            {"id": "t1", "prompt": "do it",
             "verify_command": "true"})
        self.assertEqual(result["status"], "interrupted")
        self.assertIn("relay boom", result["goal_error"])

    def test_docker_goal_task_nonzero_without_conclusion_is_interrupted(self):
        def fake_runner(cmd, timeout=None):
            return SimpleNamespace(returncode=1, stdout="",
                                   stderr="boom")

        with mock.patch("sandbox.shutil.which",
                        return_value="/usr/bin/docker"):
            executor = DockerExecutor(
                "/fake/host.py", runner=fake_runner,
                env={"XCM_API_KEY": "placeholder-not-a-real-key"})
            result = executor.goal_task(
                {"id": "t1", "prompt": "do it",
                 "verify_command": "test -f x"})
        self.assertEqual(result["status"], "interrupted")
        self.assertIn("boom", result["goal_error"])

    def test_docker_goal_resume_nonzero_without_conclusion_is_interrupted(self):
        # 覆盖 docker 侧两处 resume 解析（goal_resume_task 与
        # goal_resume_breakpoint_task 共用 shape B 链路，两个都测）。
        def fake_runner(cmd, timeout=None):
            return SimpleNamespace(returncode=1, stdout="",
                                   stderr="boom")

        with mock.patch("sandbox.shutil.which",
                        return_value="/usr/bin/docker"):
            executor = DockerExecutor(
                "/fake/host.py", runner=fake_runner,
                env={"XCM_API_KEY": "placeholder-not-a-real-key"})
            for method in ("goal_resume_task",
                           "goal_resume_breakpoint_task"):
                result = getattr(executor, method)(
                    {"id": "t1", "prompt": "do it"})
                self.assertEqual(result["status"], "interrupted")
                self.assertIn("boom", result["goal_error"])

    def test_goal_resume_refusal_is_failed_not_interrupted(self):
        # 断点丢失拒绝（2026-10-29 实弹）：host-state 里没有任务时
        # goal-resume 只打印拒绝行、正常退出；三处解析（本地、
        # Docker goal_resume_task、Docker 断点）都必须判 failed
        # 终态，绝不能判 interrupted——否则自动恢复 12 秒一轮
        # 空转到 50 次上限。
        refusal = ("goal-resume 拒绝：host-state 里没有该任务 t1，"
                   "断点已丢失，不能续跑\n")
        fake = SimpleNamespace(returncode=0, stdout=refusal, stderr="")
        result = LocalExecutor("/fake/host.py",
                               runner=lambda cmd: fake
                               )._goal_result_from_runner("t1", fake)
        self.assertEqual(result["status"], "failed")
        self.assertIn("goal-resume 拒绝", result["goal_error"])

        def fake_runner(cmd, timeout=None, env=None):
            return SimpleNamespace(returncode=0, stdout=refusal,
                                   stderr="")

        with mock.patch("sandbox.shutil.which",
                        return_value="/usr/bin/docker"):
            executor = DockerExecutor(
                "/fake/host.py", runner=fake_runner,
                env={"XCM_API_KEY": "placeholder-not-a-real-key"})
            for method in ("goal_resume_task",
                           "goal_resume_breakpoint_task"):
                result = getattr(executor, method)(
                    {"id": "t1", "prompt": "do it"})
                self.assertEqual(result["status"], "failed")
                self.assertIn("goal-resume 拒绝", result["goal_error"])

    def test_local_goal_task_already_exists_names_retry(self):
        def fake(cmd):
            return SimpleNamespace(
                returncode=1, stdout="",
                stderr="Traceback\nRuntimeError: 任务已存在")

        result = LocalExecutor("/fake/host.py",
                               runner=fake).goal_task(
            {"id": "t1", "prompt": "do it",
             "verify_command": "true"})
        self.assertEqual(result["status"], "failed")
        # P1 新语义：goal 失败请用继续而非重试（旧“请新建任务”已废）。
        self.assertIn("请用继续而非重试", result["goal_error"])

    def test_local_goal_task_without_verify_is_llm_decided(self):
        # 新语义：没有验收命令不再抛错——走模型自决分支：
        # 命令里没有 --verify，轮次缺省仍为 3（任务行没写时）。
        seen = []

        def fake(cmd):
            seen.append(list(cmd))
            return SimpleNamespace(
                returncode=0,
                stdout="goal met after rounds\n",
                stderr="")

        result = LocalExecutor("/fake/host.py",
                               runner=fake).goal_task(
            {"id": "t1", "prompt": "do it"})
        self.assertEqual(result["status"], "completed")
        self.assertNotIn("--verify", seen[0])
        self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "3")

    def test_base_goal_task_not_implemented(self):
        with self.assertRaises(NotImplementedError):
            Executor().goal_task({"id": "t1"})

    def test_popen_goal_command(self):
        seen = []

        class _FakePopen:
            def __init__(self, cmd, **kwargs):
                seen.append(list(cmd))

        with mock.patch("sandbox.subprocess.Popen", _FakePopen):
            LocalExecutor("/fake/host.py").popen_task(
                {"id": "t1", "prompt": "do it",
                 "verify_command": "test -f x", "goal_max_rounds": 2},
                kind="goal")
        self.assertIn("goal", seen[0])
        self.assertIn("test -f x", seen[0])
        self.assertEqual(seen[0][seen[0].index("--max-rounds") + 1], "2")

    def test_docker_goal_task_unmet(self):
        seen = []

        def fake_runner(cmd, timeout=None):
            seen.append(cmd)
            return SimpleNamespace(
                returncode=0,
                stdout="round 1: turn=completed verify_rc=1 "
                       "verify_out=file missing\n"
                       "goal unmet after rounds\n",
                stderr="")

        with mock.patch("sandbox.shutil.which",
                        return_value="/usr/bin/docker"):
            executor = DockerExecutor(
                "/fake/host.py", runner=fake_runner,
                env={"XCM_API_KEY": "placeholder-not-a-real-key"})
            result = executor.goal_task(
                {"id": "t1", "prompt": "do it",
                 "verify_command": "test -f x"})
        self.assertEqual(result["executor"], "docker")
        self.assertEqual(result["status"], "failed")
        self.assertFalse(result["goal_met"])
        self.assertIn("goal", seen[0])
        self.assertIn("test -f x", seen[0])


class TestUsageParsing(unittest.TestCase):
    def test_result_from_runner_usage(self):
        usage = {"input_tokens": 1000, "cached_input_tokens": 400,
                 "cache_write_input_tokens": 0, "output_tokens": 100,
                 "reasoning_output_tokens": 20, "total_tokens": 1100,
                 "completions": 2}
        fake = SimpleNamespace(
            returncode=0,
            stdout=json.dumps({"status": "completed", "text": "done",
                               "usage": usage}),
            stderr="")
        result = LocalExecutor("/fake/host.py",
                               runner=lambda cmd: fake
                               )._result_from_runner("t1", fake)
        self.assertEqual(result["usage"], usage)
        self.assertEqual(result["status"], "completed")

    def test_result_from_runner_usage_survives_long_text(self):
        # usage 在 JSON 末尾：旧的截头解析在 text 很长时会丢 usage、
        # 还会把 failed 误判成 completed；解析必须用完整 stdout。
        usage = {"input_tokens": 100, "cached_input_tokens": 0,
                 "cache_write_input_tokens": 0, "output_tokens": 10,
                 "reasoning_output_tokens": 0, "total_tokens": 110,
                 "completions": 1}
        fake = SimpleNamespace(
            returncode=0,
            stdout=json.dumps({"status": "failed", "text": "x" * 5000,
                               "usage": usage}),
            stderr="")
        result = LocalExecutor("/fake/host.py",
                               runner=lambda cmd: fake
                               )._result_from_runner("t1", fake)
        self.assertEqual(result["status"], "failed")
        self.assertEqual(result["usage"], usage)

    def test_result_from_runner_dirty_usage_dropped(self):
        for bad in ({"input_tokens": "abc"},
                    {"input_tokens": 0, "output_tokens": 0,
                     "total_tokens": 0},
                    "not-a-dict"):
            fake = SimpleNamespace(
                returncode=0,
                stdout=json.dumps({"status": "completed", "usage": bad}),
                stderr="")
            result = LocalExecutor("/fake/host.py",
                                   runner=lambda cmd: fake
                                   )._result_from_runner("t1", fake)
            self.assertIsNone(result["usage"])

    def test_goal_usage_line(self):
        fake = SimpleNamespace(
            returncode=0,
            stdout="round 1: turn=completed verify_rc=0 verify_out=ok\n"
                   "goal met after rounds\n"
                   "usage input=100 cached=40 cache_write=0 output=50 "
                   "reasoning=10 total=150 completions=2\n",
            stderr="")
        result = LocalExecutor("/fake/host.py",
                               runner=lambda cmd: fake
                               )._goal_result_from_runner("t1", fake)
        self.assertEqual(result["usage"], {
            "input_tokens": 100, "cached_input_tokens": 40,
            "cache_write_input_tokens": 0, "output_tokens": 50,
            "reasoning_output_tokens": 10, "total_tokens": 150,
            "completions": 2})
        fake2 = SimpleNamespace(
            returncode=0, stdout="goal met after rounds\n", stderr="")
        result2 = LocalExecutor("/fake/host.py",
                                runner=lambda cmd: fake2
                                )._goal_result_from_runner("t1", fake2)
        self.assertIsNone(result2["usage"])

    def test_docker_goal_usage_line(self):
        def fake_runner(cmd, timeout=None):
            return SimpleNamespace(
                returncode=0,
                stdout="round 1: turn=completed verify_rc=0 verify_out=ok\n"
                       "goal met after rounds\n"
                       "usage input=100 cached=40 cache_write=0 output=50 "
                       "reasoning=10 total=150 completions=2\n",
                stderr="")

        with mock.patch("sandbox.shutil.which",
                        return_value="/usr/bin/docker"):
            executor = DockerExecutor(
                "/fake/host.py", runner=fake_runner,
                env={"XCM_API_KEY": "placeholder-not-a-real-key"})
            result = executor.goal_task(
                {"id": "t1", "prompt": "do it",
                 "verify_command": "test -f x"})
        self.assertEqual(result["usage"]["total_tokens"], 150)
        self.assertEqual(result["usage"]["cached_input_tokens"], 40)


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