"""旧任务继续会话单测：终端任务带新消息 + 全部选项重排队，
worker 在同一会话线程上再跑一轮（不是新任务、不是分支）。

不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
SQLite；worker/sandbox 侧注入假 runner 录命令。
"""

import http.client
import json
import os
import sqlite3
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
from types import SimpleNamespace

try:
    from cloud.auth import AuthStore
    from cloud.sandbox import DockerExecutor, LocalExecutor
    from cloud.server import create_server, init_db
    from cloud.worker import Worker
except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.auth import AuthStore
    from cloud.sandbox import DockerExecutor, LocalExecutor
    from cloud.server import create_server, init_db
    from cloud.worker import Worker


class ServerContinueTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        self.auth = AuthStore(self.db_path)
        self.owner_id, self.owner_key = self.auth.create_user("owner")
        self.other_id, self.other_key = self.auth.create_user("other")
        self.server = create_server(self.db_path, "127.0.0.1", 0)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True)
        self.thread.start()
        status, self.project = self._req(
            "POST", "/projects", {"name": "demo"}, self.owner_key)
        self.assertEqual(status, 201)

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)
        try:
            self.auth.conn.close()
        except Exception:
            pass
        self.tmp.cleanup()

    def _req(self, method, path, body=None, key=None):
        conn = http.client.HTTPConnection(
            "127.0.0.1", self.port, timeout=10)
        hdrs = {}
        payload = None
        if key:
            hdrs["X-API-Key"] = key
        if body is not None:
            payload = json.dumps(body)
            hdrs["Content-Type"] = "application/json"
        conn.request(method, path, body=payload, headers=hdrs)
        resp = conn.getresponse()
        raw = resp.read()
        conn.close()
        try:
            return resp.status, json.loads(raw.decode("utf-8"))
        except (ValueError, UnicodeDecodeError):
            return resp.status, {}

    def _task_in_status(self, status, **extra):
        _s, task = self._req(
            "POST", "/tasks",
            {"project_id": self.project["id"], "prompt": "original"},
            self.owner_key)
        self.assertEqual(_s, 201, task)
        conn = sqlite3.connect(self.db_path)
        sets = ["status = ?"]
        vals = [status]
        for k, v in extra.items():
            sets.append("%s = ?" % k)
            vals.append(v)
        vals.append(task["id"])
        conn.execute("UPDATE tasks SET %s WHERE id = ?"
                     % ", ".join(sets), vals)
        conn.commit()
        conn.close()
        return task["id"]

    def _continue(self, task_id, body, key=None):
        return self._req(
            "POST", "/tasks/%s/continue" % task_id, body,
            key or self.owner_key)

    def test_continue_completed_with_options(self):
        tid = self._task_in_status("completed")
        status, task = self._continue(tid, {
            "message": "follow up please", "model": "gpt-6-sol",
            "effort": "high", "approval": "auto", "plan": 1})
        self.assertEqual(status, 200, task)
        self.assertEqual(task["status"], "queued")
        self.assertEqual(task["resume_message"], "follow up please")
        self.assertEqual(task["model"], "gpt-6-sol")
        self.assertEqual(task["effort"], "high")
        self.assertEqual(task["approval"], "auto")
        self.assertEqual(task["plan"], 1)
        self.assertEqual(task["prompt"], "original")

    def test_continue_keeps_options_when_absent(self):
        tid = self._task_in_status("failed", model="gpt-5.6-terra")
        status, task = self._continue(tid, {"message": "again"})
        self.assertEqual(status, 200, task)
        self.assertEqual(task["model"], "gpt-5.6-terra")
        self.assertEqual(task["resume_message"], "again")

    def test_continue_goal_mode_converts_to_new_semantics(self):
        tid = self._task_in_status(
            "completed", goal="old goal", verify_command="true",
            goal_max_rounds=3)
        status, task = self._continue(
            tid, {"message": "new goal", "goal_mode": 1})
        self.assertEqual(status, 200, task)
        self.assertEqual(task["goal_mode"], 1)
        self.assertEqual(task["goal"], "new goal")
        self.assertIsNone(task["verify_command"])
        self.assertEqual(task["goal_max_rounds"], 0)

    def test_continue_rejects_non_terminal_and_merged(self):
        for status in ("queued", "running"):
            tid = self._task_in_status(status)
            code, _body = self._continue(tid, {"message": "hi"})
            self.assertEqual(code, 409, status)
        tid = self._task_in_status("merged")
        code, _body = self._continue(tid, {"message": "hi"})
        self.assertEqual(code, 409)
        code, _body = self._continue("no-such-task", {"message": "hi"})
        self.assertEqual(code, 404)

    def test_continue_rejects_bad_input_and_conflicts(self):
        tid = self._task_in_status("completed")
        for body, code in [
                ({"message": ""}, 400),
                ({"message": "x" * 20001}, 400),
                ({"message": "hi", "effort": "ultra"}, 400),
                ({"message": "hi", "approval": "never"}, 400),
                ({"message": "hi", "plan": 1, "approval": "full"}, 400),
                ({"message": "hi", "plan": 1, "goal_mode": 1}, 400)]:
            got, _body = self._continue(tid, body)
            self.assertEqual(got, code, body)

    def test_continue_forbidden_for_non_member(self):
        tid = self._task_in_status("completed")
        code, _body = self._continue(tid, {"message": "hi"},
                                     key=self.other_key)
        self.assertEqual(code, 403)


class WorkerContinueTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        init_db(self.db_path)
        self.env = {"XCM_API_KEY": "test-placeholder"}

    def tearDown(self):
        self.tmp.cleanup()

    def _insert(self, task_id, **extra):
        now = time.time()
        cols = ["id", "project_id", "prompt", "status",
                "idempotency_key", "depends_on", "priority",
                "created_by", "created_at", "updated_at",
                "last_heartbeat"]
        vals = [task_id, "p1", "original", "queued", None, "[]", 0,
                "tester", now, now, None]
        for k, v in extra.items():
            cols.append(k)
            vals.append(v)
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "INSERT INTO tasks(%s) VALUES(%s)"
                % (", ".join(cols), ", ".join("?" * len(cols))), vals)
            conn.commit()
        finally:
            conn.close()

    def _row(self, task_id):
        conn = sqlite3.connect(self.db_path)
        try:
            row = conn.execute(
                "SELECT status, resume_message FROM tasks WHERE id = ?",
                (task_id,)).fetchone()
        finally:
            conn.close()
        return row

    def test_resume_normal_runs_new_message_on_same_thread(self):
        self._insert("t1", resume_message="follow up please")
        seen = []

        def fake(cmd):
            seen.append(list(cmd))
            if "dispatch" in cmd:
                # host 任务已存在：dispatch 失败，worker 回落到 run。
                return SimpleNamespace(returncode=1, stdout="任务已存在",
                                       stderr="")
            if "run" in cmd:
                return SimpleNamespace(returncode=0, stdout="ok",
                                       stderr="")
            raise AssertionError("不应有第三种命令: %s" % cmd)

        worker = Worker(self.db_path, "/fake/host.py", runner=fake,
                        env=self.env)
        result = worker.run_one()
        self.assertEqual(result["status"], "completed")
        run_cmds = [c for c in seen if "run" in c]
        self.assertTrue(run_cmds)
        self.assertEqual(run_cmds[0][run_cmds[0].index("--prompt") + 1],
                         "follow up please")
        status, resume = self._row("t1")
        self.assertEqual(status, "completed")
        self.assertIsNone(resume)

    def test_resume_goal_routes_to_goal_resume(self):
        self._insert("t2", goal_mode=1, goal="old goal",
                     goal_max_rounds=0,
                     resume_message="new goal")
        seen = []

        def fake(cmd):
            seen.append(list(cmd))
            if "goal-resume" in cmd:
                return SimpleNamespace(
                    returncode=0,
                    stdout="round 1: turn=completed verify_rc=- "
                           "verify_out=\ngoal met after rounds\n",
                    stderr="")
            raise AssertionError("goal 续跑不应走 goal/dispatch: %s" % cmd)

        worker = Worker(self.db_path, "/fake/host.py", runner=fake,
                        env=self.env)
        result = worker.run_one()
        self.assertEqual(result["status"], "completed")
        self.assertTrue(any("goal-resume" in c for c in seen))
        cmd = [c for c in seen if "goal-resume" in c][0]
        self.assertEqual(cmd[cmd.index("--goal-text") + 1], "new goal")
        self.assertEqual(cmd[cmd.index("--max-rounds") + 1], "0")
        status, resume = self._row("t2")
        self.assertEqual(status, "completed")
        self.assertIsNone(resume)


class SandboxContinueTestCase(unittest.TestCase):
    def _recording(self, calls):
        def fake(cmd):
            calls.append(list(cmd))
            return SimpleNamespace(returncode=0, stdout="ok", stderr="")
        return fake

    def test_local_goal_resume_command(self):
        calls = []
        ex = LocalExecutor("/fake/host.py",
                           runner=self._recording(calls))
        ex.goal_resume_task({"id": "t1", "prompt": "new goal"},
                            "new goal")
        self.assertIn("goal-resume", calls[0])
        self.assertEqual(calls[0][calls[0].index("--goal-text") + 1],
                         "new goal")
        self.assertEqual(calls[0][calls[0].index("--max-rounds") + 1],
                         "0")

    def test_docker_goal_resume_command(self):
        from unittest import mock
        calls = []
        ex = DockerExecutor(
            "/fake/host.py", runner=self._recording(calls),
            env={"XCM_API_KEY": "placeholder-not-a-real-key"})
        with mock.patch("cloud.sandbox.shutil.which",
                        return_value="/usr/bin/docker"):
            ex.goal_resume_task({"id": "t1", "prompt": "new goal"},
                                "new goal")
        self.assertIn("goal-resume", calls[0])
        self.assertIn("new goal", calls[0])

    def test_popen_goal_resume_command(self):
        import cloud.sandbox as sandbox_mod
        calls = []

        class FakePopen(object):
            def __init__(self, cmd, **kwargs):
                calls.append(list(cmd))

        orig = sandbox_mod.subprocess.Popen
        sandbox_mod.subprocess.Popen = FakePopen
        try:
            ex = LocalExecutor("/fake/host.py")
            ex.popen_task({"id": "t1", "prompt": "new goal"},
                          kind="goal-resume")
        finally:
            sandbox_mod.subprocess.Popen = orig
        self.assertIn("goal-resume", calls[0])
        self.assertEqual(calls[0][calls[0].index("--goal-text") + 1],
                         "new goal")


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