feat: 任务列表显示 turn_id 与 root_turn_id

host.py dispatch 打印补 turn_id/root_turn_id；执行器解析（local/Docker）透传；worker 写回 tasks 新列（新值非空才覆盖）；server 迁移与接口字段；控制台任务列表新增 turn/root turn 两列（短显+悬停全文）；新单测 9 个。

cloud/console.html

@@ -201,8 +201,8 @@
 </div>
 <div class="table-wrap">
 <table>
-  <thead><tr><th>任务</th><th>prompt</th><th>状态</th><th>executor</th><th>主机</th><th>PR</th><th>更新</th></tr></thead>
-  <tbody id="taskRows"><tr><td colspan="7" class="muted">先选项目</td></tr></tbody>
+  <thead><tr><th>任务</th><th>turn</th><th>root turn</th><th>prompt</th><th>状态</th><th>executor</th><th>主机</th><th>PR</th><th>更新</th></tr></thead>
+  <tbody id="taskRows"><tr><td colspan="9" class="muted">先选项目</td></tr></tbody>
 </table>
 </div>
 </section>
@@ -473,16 +473,18 @@ async function loadEnvironments() {
 
 async function loadTasks() {
   var pid = $("#projectSelect").value;
-  if (!pid) { $("#taskRows").innerHTML = "<tr><td colspan='7' class='muted'>先选项目</td></tr>"; return; }
+  if (!pid) { $("#taskRows").innerHTML = "<tr><td colspan='9' class='muted'>先选项目</td></tr>"; return; }
   try {
     var d = await api("/tasks?project_id=" + encodeURIComponent(pid), "GET");
     var tasks = d.tasks || [];
     $("#taskCount").textContent = tasks.length + " 个任务";
-    if (!tasks.length) { $("#taskRows").innerHTML = "<tr><td colspan='7' class='muted'>该项目暂无任务</td></tr>"; return; }
+    if (!tasks.length) { $("#taskRows").innerHTML = "<tr><td colspan='9' class='muted'>该项目暂无任务</td></tr>"; return; }
     $("#taskRows").innerHTML = tasks.slice().reverse().map(function (t) {
       var pr = t.pr_url ? "<a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>#" + esc(t.pr_number) + "</a>" : "";
       return "<tr data-id='" + esc(t.id) + "' class='" + (currentTask && currentTask.id === t.id ? "selected" : "") + "'>" +
         "<td>" + esc(shortId(t.id)) + "</td>" +
+        "<td title='" + esc(t.turn_id || "") + "'>" + esc(shortId(t.turn_id)) + "</td>" +
+        "<td title='" + esc(t.root_turn_id || "") + "'>" + esc(shortId(t.root_turn_id)) + "</td>" +
         "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + ((t.goal_mode || t.verify_command) ? "[目标] " : "") + (t.environment_id ? "[环境 " + esc(shortId(t.environment_id)) + "] " : "") + ((t.attachments && t.attachments.length) ? "[附件×" + t.attachments.length + "] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
         "<td class='st-" + esc(t.status) + "'>" + esc(t.status) + "</td>" +
         "<td>" + esc(t.executor || "") + "</td>" +
@@ -493,7 +495,7 @@ async function loadTasks() {
     Array.prototype.forEach.call(document.querySelectorAll("#taskRows tr[data-id]"), function (tr) {
       tr.onclick = function () { showDetail(tr.getAttribute("data-id")); };
     });
-  } catch (e) { $("#taskRows").innerHTML = "<tr><td colspan='7' class='err'>加载失败：" + esc(e.message) + "</td></tr>"; }
+  } catch (e) { $("#taskRows").innerHTML = "<tr><td colspan='9' class='err'>加载失败：" + esc(e.message) + "</td></tr>"; }
 }
 
 function stopTrace() {

cloud/db.py

@@ -839,6 +839,8 @@ POSTGRES_SCHEMA = (
     "forked_from TEXT, "
     "environment_id TEXT, "
     "resume_message TEXT, "
+    "turn_id TEXT, "
+    "root_turn_id TEXT, "
     "UNIQUE(project_id, idempotency_key))",
     "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
     # 附件包与任务附件：与 server._ensure_uploads_tables 同款，

cloud/sandbox.py

@@ -227,6 +227,8 @@ class LocalExecutor(Executor):
         status = "failed"
         usage = None
         task_messages = []
+        turn_id = None
+        root_turn_id = None
         if returncode == 0:
             status = "completed"
             # JSON 解析必须用完整 stdout：截头后的前 2000 字在
@@ -245,6 +247,13 @@ class LocalExecutor(Executor):
                 # 原样透传给 worker 记 actual；脏数据由
                 # _clean_usage 挡掉，不进 actual。
                 usage = _clean_usage(parsed.get("usage"))
+                # 任务列表显示 turn 用：本轮回合的 turn_id 与
+                # root_turn_id（dispatch 打印里带，run 打印整
+                # 个 result 所以也有）；取不到时保持 None，
+                # worker 写回时不覆盖旧值（见 worker 的
+                # _write_execution_result）。
+                turn_id = parsed.get("turn_id")
+                root_turn_id = parsed.get("root_turn_id")
                 # 互发消息标记块：从完整解析出的 text 里提取（截头
                 # 版 text 尾部的块会丢），worker 在 completed 轮次
                 # 后投递；goal 路径不走这里，goal 任务只收不发。
@@ -266,6 +275,8 @@ class LocalExecutor(Executor):
             "stderr": _truncate_stdout(getattr(result, "stderr", "")),
             "usage": usage,
             "task_messages": task_messages,
+            "turn_id": turn_id,
+            "root_turn_id": root_turn_id,
         }
 
     def _host_name(self, task):
@@ -1240,6 +1251,8 @@ def _parse_host_result(task_id, fake):
     host_status = None
     usage = None
     task_messages = []
+    turn_id = None
+    root_turn_id = None
     if returncode == 0:
         status = "completed"
         # 与 _result_from_runner 同理：JSON 解析用完整 stdout，
@@ -1256,6 +1269,10 @@ def _parse_host_result(task_id, fake):
                       else "failed")
             usage = _clean_usage(parsed.get("usage"))
             task_messages = _extract_peer_messages(parsed.get("text"))
+            # 任务列表显示 turn 用：与 Local 版同款语义，取不到
+            # 时保持 None，worker 写回时不覆盖旧值。
+            turn_id = parsed.get("turn_id")
+            root_turn_id = parsed.get("root_turn_id")
     return {
         "executor": "docker",
         "task_id": task_id,
@@ -1266,6 +1283,8 @@ def _parse_host_result(task_id, fake):
         "stderr": _truncate_stdout(getattr(fake, "stderr", "")),
         "usage": usage,
         "task_messages": task_messages,
+        "turn_id": turn_id,
+        "root_turn_id": root_turn_id,
     }
 
 

cloud/server.py

@@ -757,6 +757,10 @@ def migrate_tasks_table(conn):
         # 续跑消息）。worker 认领后把 prompt 换成它跑一轮，跑完
         # 清掉；任务详情里照常显示原 prompt，续跑消息只作本轮输入。
         "resume_message": "TEXT",
+        # 任务列表显示 turn 用：最近一轮的 turn_id 与
+        # root_turn_id（worker 写回时只在新值非空时覆盖）。
+        "turn_id": "TEXT",
+        "root_turn_id": "TEXT",
     }
     for name, coltype in new_cols.items():
         if name not in cols:
@@ -1080,6 +1084,8 @@ def _task_to_dict(row, attachments=None):
         "plan": _row_get(row, "plan") or 0,
         "forked_from": _row_get(row, "forked_from"),
         "environment_id": _row_get(row, "environment_id"),
+        "turn_id": _row_get(row, "turn_id"),
+        "root_turn_id": _row_get(row, "root_turn_id"),
     }
 
 

cloud/tests/test_task_turn_ids.py

@@ -0,0 +1,187 @@
+"""任务列表显示 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()
+        self.assertIn("<th>turn</th>", html)
+        self.assertIn("<th>root turn</th>", html)
+        self.assertIn("shortId(t.turn_id)", html)
+        self.assertIn("shortId(t.root_turn_id)", html)
+        self.assertNotIn("colspan='7'", html)
+        self.assertNotIn('colspan="7"', html)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -607,7 +607,12 @@ class Worker:
                 "plan": "INTEGER",
                 "forked_from": "TEXT",
                 "environment_id": "TEXT",
-                "resume_message": "TEXT"}.items():
+                "resume_message": "TEXT",
+                # 任务列表显示 turn 用：本任务最近一轮产出过
+                # turn_id 的回合编号（local/Docker 两路都经
+                # _result_from_runner/_parse_host_result 提取）。
+                "turn_id": "TEXT",
+                "root_turn_id": "TEXT"}.items():
             if name not in cols:
                 try:
                     conn.execute(
@@ -640,8 +645,15 @@ class Worker:
     def _write_execution_result(self, task_id, result_text=None,
                                 error_text=None, actual_model=None,
                                 started_at=None, finished_at=None,
-                                child_pid=None):
-        """把执行结果写进 tasks 新列；老库缺列时先迁移再写。"""
+                                child_pid=None, turn_id=None,
+                                root_turn_id=None):
+        """把执行结果写进 tasks 新列；老库缺列时先迁移再写。
+
+        turn_id/root_turn_id 只在新值非空时覆盖：某轮没产出
+        turn（如提前失败门、解析不到 JSON）时保留上一轮的值，
+        列表里始终显示最近一个真实回合的编号，而不是被空值
+        洗掉。
+        """
         conn = _db_connect(self.db_path)
         try:
             self._ensure_result_columns(conn)
@@ -649,20 +661,27 @@ class Worker:
                 conn.execute(
                     "UPDATE tasks SET result_text = ?, error_text = ?,"
                     " actual_model = ?, started_at = ?, finished_at = ?,"
+                    " turn_id = COALESCE(?, turn_id),"
+                    " root_turn_id = COALESCE(?, root_turn_id),"
                     " updated_at = ? WHERE id = ?",
                     ((result_text or "")[:4000] if result_text else None,
                      (error_text or "")[:4000] if error_text else None,
                      actual_model, started_at, finished_at,
+                     turn_id, root_turn_id,
                      time.time(), task_id),
                 )
             else:
                 conn.execute(
                     "UPDATE tasks SET result_text = ?, error_text = ?,"
                     " actual_model = ?, started_at = ?, finished_at = ?,"
-                    " child_pid = ?, updated_at = ? WHERE id = ?",
+                    " child_pid = ?,"
+                    " turn_id = COALESCE(?, turn_id),"
+                    " root_turn_id = COALESCE(?, root_turn_id),"
+                    " updated_at = ? WHERE id = ?",
                     ((result_text or "")[:4000] if result_text else None,
                      (error_text or "")[:4000] if error_text else None,
                      actual_model, started_at, finished_at, child_pid,
+                     turn_id, root_turn_id,
                      time.time(), task_id),
                 )
             conn.commit()
@@ -2330,7 +2349,9 @@ class Worker:
             self._write_execution_result(
                 task_id, result_text=(result or {}).get("stdout", ""),
                 error_text="cancelled", actual_model=actual_model,
-                started_at=started_at, finished_at=time.time())
+                started_at=started_at, finished_at=time.time(),
+                turn_id=(result or {}).get("turn_id"),
+                root_turn_id=(result or {}).get("root_turn_id"))
             log_event(self.logger, "worker_cancelled", task_id=task_id,
                       project_id=task["project_id"], status="interrupted")
             return {"dispatched": True, "task_id": task_id,
@@ -2397,7 +2418,9 @@ class Worker:
         self._write_execution_result(
             task_id, result_text=stdout, error_text=error,
             actual_model=actual_model, started_at=started_at,
-            finished_at=finished_at)
+            finished_at=finished_at,
+            turn_id=(result or {}).get("turn_id"),
+            root_turn_id=(result or {}).get("root_turn_id"))
         # 附件暂存清理：任务终局后删 bundle 目录（重试走已存在
         # 续跑，文件已在 worktree 里，不再需要暂存）。
         self._cleanup_task_seed(task)

host.py

@@ -2373,6 +2373,8 @@ def main():
                               "cwd": task["cwd"], "status": task["status"],
                               "text": result.get("text", ""),
                               "error": result.get("error"),
+                              "turn_id": result.get("turn_id"),
+                              "root_turn_id": result.get("root_turn_id"),
                               "usage": result.get("usage")},
                              indent=2, ensure_ascii=False))
         elif args.cmd == "list":