diff --git a/cloud/console.html b/cloud/console.html index c84942d..f2bdd7c 100644 --- a/cloud/console.html +++ b/cloud/console.html @@ -201,8 +201,8 @@
- - + +
任务prompt状态executor主机PR更新
先选项目
任务turnroot turnprompt状态executor主机PR更新
先选项目
@@ -473,16 +473,18 @@ async function loadEnvironments() { async function loadTasks() { var pid = $("#projectSelect").value; - if (!pid) { $("#taskRows").innerHTML = "先选项目"; return; } + if (!pid) { $("#taskRows").innerHTML = "先选项目"; 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 = "该项目暂无任务"; return; } + if (!tasks.length) { $("#taskRows").innerHTML = "该项目暂无任务"; return; } $("#taskRows").innerHTML = tasks.slice().reverse().map(function (t) { var pr = t.pr_url ? "#" + esc(t.pr_number) + "" : ""; return "" + "" + esc(shortId(t.id)) + "" + + "" + esc(shortId(t.turn_id)) + "" + + "" + esc(shortId(t.root_turn_id)) + "" + "" + (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)) + "" + "" + esc(t.status) + "" + "" + esc(t.executor || "") + "" + @@ -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 = "加载失败：" + esc(e.message) + ""; } + } catch (e) { $("#taskRows").innerHTML = "加载失败：" + esc(e.message) + ""; } } function stopTrace() { diff --git a/cloud/db.py b/cloud/db.py index b66d198..177a73e 100644 --- a/cloud/db.py +++ b/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 同款， diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 238dd79..dd856fd 100644 --- a/cloud/sandbox.py +++ b/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, } diff --git a/cloud/server.py b/cloud/server.py index 6ddfada..7c59495 100644 --- a/cloud/server.py +++ b/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"), } diff --git a/cloud/tests/test_task_turn_ids.py b/cloud/tests/test_task_turn_ids.py new file mode 100644 index 0000000..d41a1db --- /dev/null +++ b/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("turn", html) + self.assertIn("root turn", 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() diff --git a/cloud/worker.py b/cloud/worker.py index 9e5ce27..7b624c9 100644 --- a/cloud/worker.py +++ b/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) diff --git a/host.py b/host.py index b405a43..a8d8924 100644 --- a/host.py +++ b/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":