codex-app-server/cloud/tests/test_task_turn_ids.py

"""任务列表显示 turn 单测：执行器解析、worker 写回、接口字段、迁移、控制台列。

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

只用假 runner 与内存库，不跑真实模型回合、不联网；turn id
用固定假值，绝不是真实密钥或真实回合编号。
"""

import json
import os
import sqlite3
import sys
import tempfile
import time
import unittest
from pathlib import Path
from types import SimpleNamespace

try:
    from cloud.sandbox import LocalExecutor, _parse_host_result
    from cloud.server import _task_to_dict, init_db, migrate_tasks_table
    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.sandbox import LocalExecutor, _parse_host_result
    from cloud.server import _task_to_dict, init_db, migrate_tasks_table
    from cloud.worker import Worker

TURN = "11111111-2222-3333-4444-555555555555"
ROOT = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"


def _host_json(name):
    return json.dumps({"task": name, "status": "completed",
                       "text": "done", "turn_id": TURN,
                       "root_turn_id": ROOT})


class ParseTestCase(unittest.TestCase):
    def test_local_result_carries_turn_ids(self):
        ex = LocalExecutor("/fake/host.py")
        out = ex._result_from_runner(
            "t1", SimpleNamespace(returncode=0,
                                  stdout=_host_json("t1"), stderr=""))
        self.assertEqual(out["turn_id"], TURN)
        self.assertEqual(out["root_turn_id"], ROOT)

    def test_local_result_missing_ids_is_none(self):
        ex = LocalExecutor("/fake/host.py")
        out = ex._result_from_runner(
            "t1", SimpleNamespace(
                returncode=0,
                stdout=json.dumps({"task": "t1", "status": "completed",
                                   "text": "done"}),
                stderr=""))
        self.assertIsNone(out["turn_id"])
        self.assertIsNone(out["root_turn_id"])

    def test_docker_parse_carries_turn_ids(self):
        out = _parse_host_result(
            "t1", SimpleNamespace(returncode=0,
                                  stdout=_host_json("t1"), stderr=""))
        self.assertEqual(out["turn_id"], TURN)
        self.assertEqual(out["root_turn_id"], ROOT)


class WorkerWritebackTestCase(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)

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

    def _insert_task(self, task_id):
        now = time.time()
        conn = sqlite3.connect(self.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(?, 'p1', 'do work', 'queued',"
                " NULL, '[]', 0, 'tester', ?, ?, NULL)",
                (task_id, now, now),
            )
            conn.commit()
        finally:
            conn.close()

    def test_run_all_persists_turn_ids(self):
        def runner(cmd):
            if "dispatch" in cmd:
                name = cmd[cmd.index("--name") + 1]
            else:
                name = cmd[cmd.index("--task") + 1]
            return SimpleNamespace(returncode=0,
                                   stdout=_host_json(name), stderr="")
        self._insert_task("t1")
        worker = Worker(self.db_path, "/fake/host.py", runner=runner,
                        env={"XCM_API_KEY": "test-placeholder"})
        worker.run_all()
        conn = sqlite3.connect(self.db_path)
        try:
            row = conn.execute(
                "SELECT turn_id, root_turn_id FROM tasks WHERE id = 't1'"
            ).fetchone()
        finally:
            conn.close()
        self.assertEqual(row, (TURN, ROOT))

    def test_writeback_without_ids_keeps_first(self):
        # 直接验 COALESCE 语义：首轮写了 turn，次轮没给 turn
        # 时写回不得把首轮的值洗成空（对应续跑 JSON 里没有
        # turn 字段的旧 host 打印）。
        self._insert_task("t1")
        worker = Worker(self.db_path, "/fake/host.py",
                        env={"XCM_API_KEY": "test-placeholder"})
        worker._write_execution_result(
            "t1", result_text="first", turn_id=TURN, root_turn_id=ROOT,
            started_at=time.time(), finished_at=time.time())
        worker._write_execution_result(
            "t1", result_text="second",
            started_at=time.time(), finished_at=time.time())
        conn = sqlite3.connect(self.db_path)
        try:
            row = conn.execute(
                "SELECT turn_id, root_turn_id FROM tasks WHERE id = 't1'"
            ).fetchone()
        finally:
            conn.close()
        self.assertEqual(row, (TURN, ROOT))


class ServerFieldsTestCase(unittest.TestCase):
    def test_task_to_dict_includes_turn_ids(self):
        row = {"id": "t1", "project_id": "p1", "prompt": "p",
               "status": "completed", "idempotency_key": None,
               "depends_on": "[]", "priority": 0, "created_by": "u",
               "created_at": 1.0, "updated_at": 2.0,
               "last_heartbeat": None, "turn_id": TURN,
               "root_turn_id": ROOT}
        d = _task_to_dict(row, [])
        self.assertEqual(d["turn_id"], TURN)
        self.assertEqual(d["root_turn_id"], ROOT)

    def test_task_to_dict_missing_columns_is_none(self):
        row = {"id": "t1", "project_id": "p1", "prompt": "p",
               "status": "completed", "idempotency_key": None,
               "depends_on": "[]", "priority": 0, "created_by": "u",
               "created_at": 1.0, "updated_at": 2.0,
               "last_heartbeat": None}
        d = _task_to_dict(row, [])
        self.assertIsNone(d["turn_id"])
        self.assertIsNone(d["root_turn_id"])

    def test_migrate_adds_turn_columns(self):
        conn = sqlite3.connect(":memory:")
        try:
            conn.execute(
                "CREATE TABLE tasks(id TEXT PRIMARY KEY,"
                " project_id TEXT, prompt TEXT, status TEXT)")
            migrate_tasks_table(conn)
            cols = {r[1] for r in
                    conn.execute("PRAGMA table_info(tasks)").fetchall()}
        finally:
            conn.close()
        self.assertIn("turn_id", cols)
        self.assertIn("root_turn_id", cols)


class ConsoleColumnsTestCase(unittest.TestCase):
    def test_console_task_list_has_turn_columns(self):
        html = (Path(__file__).resolve().parents[2]
                / "cloud" / "console.html").read_text()
        # React 版（2026-10-29 起）：左栏是紧凑任务列表，不再是
        # 9 列大表，turn / root turn 信息改在右侧任务详情里展示
        # （三栏布局的用户定案）；断言意图不变——turn 信息没有丢，
        # 只是从表头移到了详情面板，且没有 7 列合并的旧表。
        self.assertIn("root turn", html)
        self.assertIn("turn", html)
        self.assertIn("turn_id", html)
        self.assertIn("root_turn_id", html)
        self.assertNotIn("colspan='7'", html)
        self.assertNotIn('colspan="7"', html)


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