"""P0 三件单测：断点进库、错误分类+熔断、Run/Step 时间线。

只测本路改动，不跑真实模型、不联网。
- 分类：KeyError/配置缺失等永久错误判 failed 死信；网络抖动
  等瞬时错误保持 interrupted；未知错误不分类（走原有逻辑）。
- 熔断：退避期内不重排队；库断点轮次连续 3 次无进展即停，
  标 auto_resume_stalled；有进展时 stall 清零。
- 断点：worker 镜像把磁盘 host-state 断点写进 task_checkpoints；
  条目丢失后可按库断点重建最小条目（不造假：库里无断点时
  返回 False）。
- 时间线：timeline 接口按任务返回 steps 与 checkpoints。
"""

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 (LocalExecutor, _classify_goal_error)
    from cloud.server import create_server, init_db
    from cloud.worker import Worker
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.auth import AuthStore
    from cloud.sandbox import (LocalExecutor, _classify_goal_error)
    from cloud.server import create_server, init_db
    from cloud.worker import Worker


def _insert_task(db_path, task_id, status="interrupted", **cols):
    now = time.time()
    conn = sqlite3.connect(db_path)
    try:
        conn.execute(
            "INSERT INTO tasks(id, project_id, prompt, status,"
            " idempotency_key, depends_on, priority, created_by,"
            " created_at, updated_at, last_heartbeat)"
            " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)",
            (task_id, "p1", "do work", status, "tester", now, now),
        )
        for key, value in cols.items():
            conn.execute(
                "UPDATE tasks SET %s = ? WHERE id = ?" % key,
                (value, task_id))
        conn.commit()
    finally:
        conn.close()


def _task_row(db_path, task_id):
    conn = sqlite3.connect(db_path)
    try:
        return conn.execute(
            "SELECT status, goal_resume_pending, auto_resume_count,"
            " auto_resume_stall_count, checkpoint_round_at_last_resume,"
            " resume_after, interrupted_reason FROM tasks WHERE id = ?",
            (task_id,)).fetchone()
    finally:
        conn.close()


class ClassifyTestCase(unittest.TestCase):
    def test_permanent_keyerror(self):
        kind, snippet = _classify_goal_error(
            "Traceback (most recent call last):\n"
            "KeyError: '3829f5f3efb049dfafaecbd15232064c'\n", "")
        self.assertEqual(kind, "permanent")
        self.assertIn("KeyError", snippet)

    def test_permanent_missing_env(self):
        kind, _ = _classify_goal_error(
            "", "Missing environment variable: SUB2API_API_KEY")
        self.assertEqual(kind, "permanent")

    def test_transient_gateway(self):
        kind, _ = _classify_goal_error("", "HTTP 503 Service Unavailable")
        self.assertEqual(kind, "transient")

    def test_unknown_not_classified(self):
        kind, snippet = _classify_goal_error("some odd output", "")
        self.assertIsNone(kind)
        self.assertIsNone(snippet)

    def test_keyerror_no_conclusion_is_failed(self):
        fake = SimpleNamespace(
            returncode=1,
            stdout="Traceback (most recent call last):\n"
                   "KeyError: 't1'\n",
            stderr="")
        result = LocalExecutor("/fake/host.py")._goal_result_from_runner(
            "t1", fake)
        self.assertEqual(result["status"], "failed")
        self.assertIn("KeyError", result["goal_error"])

    def test_transient_no_conclusion_stays_interrupted(self):
        fake = SimpleNamespace(
            returncode=1,
            stdout="",
            stderr="RemoteDisconnected: connection closed")
        result = LocalExecutor("/fake/host.py")._goal_result_from_runner(
            "t1", fake)
        self.assertEqual(result["status"], "interrupted")


class StallGuardTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "test.db")
        init_db(self.db_path)
        self.worker = Worker(
            self.db_path, "/fake/host.py", env={"XCM_API_KEY": "x"},
            task_timeout=1800)

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

    def _checkpoint(self, task_id, rounds):
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "INSERT INTO task_checkpoints(task_id, round_no,"
                " thread_id, goal_status, state_sha, verify_out,"
                " created_at) VALUES(?, ?, 'th', 'goal-running',"
                " 'sha', '', ?)", (task_id, rounds, time.time()))
            conn.commit()
        finally:
            conn.close()

    def test_backoff_blocks_early_resume(self):
        _insert_task(self.db_path, "T1", goal_mode=1, goal="g",
                     interrupted_reason="sweep", paused_by_user=0,
                     resume_after=time.time() + 600)
        self.assertEqual(self.worker.auto_resume_interrupted(), [])
        self.assertEqual(_task_row(self.db_path, "T1")[0], "interrupted")

    def test_three_no_progress_stalls(self):
        _insert_task(self.db_path, "T1", goal_mode=1, goal="g",
                     interrupted_reason="sweep", paused_by_user=0,
                     checkpoint_round_at_last_resume=5,
                     auto_resume_stall_count=2)
        self._checkpoint("T1", 5)
        self.assertEqual(self.worker.auto_resume_interrupted(), [])
        row = _task_row(self.db_path, "T1")
        self.assertEqual(row[0], "interrupted")
        self.assertEqual(row[6], "auto_resume_stalled")

    def test_progress_resets_stall(self):
        _insert_task(self.db_path, "T1", goal_mode=1, goal="g",
                     interrupted_reason="sweep", paused_by_user=0,
                     checkpoint_round_at_last_resume=5,
                     auto_resume_stall_count=2)
        self._checkpoint("T1", 6)
        self.assertEqual(self.worker.auto_resume_interrupted(), ["T1"])
        row = _task_row(self.db_path, "T1")
        self.assertEqual(row[0], "queued")
        self.assertEqual(row[3], 0)
        self.assertEqual(row[4], 6)
        self.assertGreater(row[5], time.time())


class CheckpointMirrorTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "test.db")
        init_db(self.db_path)
        self.state_dir = os.path.join(self.tmp.name, "work")
        os.makedirs(self.state_dir)
        cfg_path = os.path.join(self.tmp.name, "host-config.json")
        with open(cfg_path, "w", encoding="utf-8") as fh:
            json.dump({"state": "work/host-state.json"}, fh)
        self.state_path = os.path.join(self.state_dir, "host-state.json")
        with open(self.state_path, "w", encoding="utf-8") as fh:
            json.dump({"version": 1, "tasks": {"T1": {
                "goal_rounds_done": 3, "thread_id": "th1",
                "goal_status": "goal-running",
                "goal_last_verify": "ok"}}}, fh)
        self.worker = Worker(
            self.db_path, "/fake/host.py", env={"XCM_API_KEY": "x"},
            task_timeout=1800)
        self.worker.executor = LocalExecutor(
            "/fake/host.py", config_path=cfg_path)
        _insert_task(self.db_path, "T1", status="running",
                     goal_mode=1, goal="g", verify_command=None,
                     goal_max_rounds=0)
        self.task = {"id": "T1", "project_id": "p1"}

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

    def test_mirror_then_rebuild(self):
        self.assertEqual(self.worker._mirror_checkpoint_once(self.task), 3)
        conn = sqlite3.connect(self.db_path)
        try:
            row = conn.execute(
                "SELECT round_no, thread_id FROM task_checkpoints"
                " WHERE task_id = 'T1'").fetchone()
        finally:
            conn.close()
        self.assertEqual(row, (3, "th1"))
        # 模拟任务 2 的断点丢失：磁盘条目被删，库里有断点。
        with open(self.state_path, "w", encoding="utf-8") as fh:
            json.dump({"version": 1, "tasks": {}}, fh)
        self.assertTrue(
            self.worker._rebuild_host_state_from_checkpoint(self.task))
        with open(self.state_path, encoding="utf-8") as fh:
            entry = json.load(fh)["tasks"]["T1"]
        self.assertEqual(entry["goal_rounds_done"], 3)
        self.assertEqual(entry["thread_id"], "th1")
        self.assertTrue(entry["rebuilt_from_checkpoint"])

    def test_mirror_imports_host_steps(self):
        steps_dir = os.path.join(self.state_dir, "host-steps")
        os.makedirs(steps_dir)
        with open(os.path.join(steps_dir, "T1.jsonl"),
                  "w", encoding="utf-8") as fh:
            fh.write(json.dumps(
                {"round": 1, "turn_status": "completed",
                 "verify_out": "", "ts": time.time()}) + "\n")
            fh.write("not json\n")
            fh.write(json.dumps(
                {"round": 2, "turn_status": "completed",
                 "verify_out": "", "ts": time.time()}) + "\n")
        self.worker._mirror_checkpoint_once(self.task)
        self.worker._mirror_checkpoint_once(self.task)  # 去重：不重复插
        conn = sqlite3.connect(self.db_path)
        try:
            rows = conn.execute(
                "SELECT step_no, kind FROM task_steps"
                " WHERE task_id = 'T1' ORDER BY step_no").fetchall()
        finally:
            conn.close()
        self.assertEqual(rows, [(1, "round"), (2, "round")])

    def test_mirror_falls_back_to_status_label(self):
        # host-state 条目写的是 status 不是 goal_status：实弹里
        # 镜像到的 goal_status 全是 None，修法是取 status 回落。
        with open(self.state_path, "w", encoding="utf-8") as fh:
            json.dump({"version": 1, "tasks": {"T1": {
                "goal_rounds_done": 2, "thread_id": "th2",
                "status": "goal-met",
                "goal_last_verify": ""}}}, fh)
        self.assertEqual(self.worker._mirror_checkpoint_once(self.task), 2)
        conn = sqlite3.connect(self.db_path)
        try:
            row = conn.execute(
                "SELECT round_no, thread_id, goal_status"
                " FROM task_checkpoints WHERE task_id = 'T1'").fetchone()
        finally:
            conn.close()
        self.assertEqual(row, (2, "th2", "goal-met"))

    def test_rebuild_without_checkpoint_is_false(self):
        with open(self.state_path, "w", encoding="utf-8") as fh:
            json.dump({"version": 1, "tasks": {}}, fh)
        self.assertFalse(
            self.worker._rebuild_host_state_from_checkpoint(self.task))


class TimelineApiTestCase(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.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)
        _insert_task(self.db_path, "T9", status="completed",
                     goal_mode=1, goal="g")
        conn = sqlite3.connect(self.db_path)
        try:
            conn.execute(
                "UPDATE tasks SET project_id = ? WHERE id = 'T9'",
                (self.project["id"],))
            conn.execute(
                "INSERT INTO task_steps(id, task_id, step_no, kind,"
                " summary, state_sha, created_at)"
                " VALUES('s1', 'T9', 1, 'claim', 'worker 认领', NULL, ?)",
                (time.time(),))
            conn.execute(
                "INSERT INTO task_checkpoints(task_id, round_no,"
                " thread_id, goal_status, state_sha, verify_out,"
                " created_at) VALUES('T9', 2, 'th', 'goal-running',"
                " 'sha', '', ?)", (time.time(),))
            conn.commit()
        finally:
            conn.close()

    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)
        headers = {"Content-Type": "application/json"}
        if key:
            headers["X-API-Key"] = key
        conn.request(method, path,
                     json.dumps(body) if body is not None else None,
                     headers)
        resp = conn.getresponse()
        data = resp.read()
        conn.close()
        try:
            return resp.status, json.loads(data)
        except Exception:
            return resp.status, data

    def test_timeline_returns_steps_and_checkpoints(self):
        status, body = self._req(
            "GET", "/tasks/T9/timeline", key=self.owner_key)
        self.assertEqual(status, 200)
        self.assertEqual(len(body["steps"]), 1)
        self.assertEqual(body["steps"][0]["kind"], "claim")
        self.assertEqual(len(body["checkpoints"]), 1)
        self.assertEqual(body["checkpoints"][0]["round_no"], 2)

    def test_timeline_missing_task_404(self):
        status, _ = self._req(
            "GET", "/tasks/nope/timeline", key=self.owner_key)
        self.assertEqual(status, 404)


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