Merge pull request #111 from openqj/feat-coord-row-vanished

协调员行消失按中断处理：本体行消失即放槽，子任务行消失按 interrupted 合流

cloud/tests/test_coordinator_row_vanished.py

@@ -0,0 +1,204 @@
+"""协调员行消失单测：本体行消失与子任务行消失都不等到超时。
+
+不跑真实模型回合、不联网：worker 侧用临时 SQLite + 假 runner，
+等待中消失用“线程里跑协调员、主线程删行、join 带超时断言”
+的办法——不用“删行后直接 run_one”（run_one 先认领，行已没了
+根本认领不上），也不靠 sleep 时长赌竞态：删行后下一轮轮询
+（poll_interval=0.05 秒）必然看到消失，join 超时即为失败信号。
+"""
+
+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.server import 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.server import init_db
+    from cloud.worker import Worker
+
+
+def _completed_runner(cmd):
+    return SimpleNamespace(returncode=0, stdout="{}", stderr="")
+
+
+def _insert_task(db_path, task_id, prompt="do work", status="queued",
+                 project_id="p1", coordinator_parts=None,
+                 parent_task_id=None, coordinator_part=None,
+                 result_text=None, error_text=None):
+    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, coordinator_parts,"
+            " parent_task_id, coordinator_part, result_text, error_text)"
+            " VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', ?, ?, NULL,"
+            " ?, ?, ?, ?, ?)",
+            (task_id, project_id, prompt, status, now, now,
+             json.dumps(coordinator_parts) if coordinator_parts else None,
+             parent_task_id, coordinator_part, result_text, error_text),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT id, status, result_text, error_text FROM tasks"
+            " WHERE id = ?", (task_id,)).fetchone()
+    finally:
+        conn.close()
+    if row is None:
+        return None
+    return {"id": row[0], "status": row[1], "result_text": row[2],
+            "error_text": row[3]}
+
+
+def _delete_row(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _set_running(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "UPDATE tasks SET status = 'running' WHERE id = ?",
+            (task_id,))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+class RowVanishedTestCase(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 _worker(self):
+        return Worker(self.db_path, "/fake/host.py",
+                      runner=_completed_runner, env=self.env)
+
+    def _parts(self):
+        return [{"name": "后端", "prompt": "做后端"},
+                {"name": "前端", "prompt": "做前端"}]
+
+    def _run_in_thread(self, worker, task, timeout):
+        """线程里跑 _run_coordinator，返回 (结果, 耗时)；join 超时即失败。"""
+        box = {}
+
+        def _target():
+            started = time.time()
+            box["result"] = worker._run_coordinator(
+                task, time.time(), poll_interval=0.05, timeout=timeout)
+            box["elapsed"] = time.time() - started
+
+        thread = threading.Thread(target=_target)
+        thread.start()
+        return thread, box
+
+    def test_parent_row_vanished_exits_as_interrupted(self):
+        """(a) 协调员本体行在等待中消失 → 远小于 timeout 退出，
+        返回 interrupted，不是 coordinator timeout 的 failed。"""
+        _insert_task(self.db_path, "coord",
+                     coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c1", status="completed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="后端好了")
+        _insert_task(self.db_path, "c2", status="queued",
+                     parent_task_id="coord", coordinator_part="前端")
+        _set_running(self.db_path, "coord")
+        worker = self._worker()
+        task = worker.get_task("coord")
+        thread, box = self._run_in_thread(worker, task, timeout=60)
+        time.sleep(0.3)
+        _delete_row(self.db_path, "coord")
+        thread.join(timeout=10)
+        self.assertFalse(thread.is_alive(),
+                         "本体行消失后协调员循环应在 10 秒内退出")
+        self.assertLess(box["elapsed"], 10)
+        self.assertEqual(box["result"]["status"], "interrupted")
+        self.assertNotIn("timeout", box["result"]["error"] or "")
+        # 行已没了，终局写回写不回去，也不抛错、不当完成。
+        self.assertIsNone(_row(self.db_path, "coord"))
+
+    def test_child_row_vanished_counts_as_interrupted(self):
+        """(b) 子任务行在等待中消失 → 合流报告里该子任务为
+        interrupted 且注明行已消失，整体不为 completed，其它
+        子任务终态不受影响，且不拖到 timeout。"""
+        _insert_task(self.db_path, "coord",
+                     coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c1", status="completed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="后端好了")
+        _insert_task(self.db_path, "c2", status="queued",
+                     parent_task_id="coord", coordinator_part="前端")
+        _set_running(self.db_path, "coord")
+        worker = self._worker()
+        task = worker.get_task("coord")
+        thread, box = self._run_in_thread(worker, task, timeout=60)
+        time.sleep(0.3)
+        _delete_row(self.db_path, "c2")
+        thread.join(timeout=10)
+        self.assertFalse(thread.is_alive(),
+                         "子任务行消失后协调员应在 10 秒内合流退出")
+        self.assertLess(box["elapsed"], 10)
+        result = box["result"]
+        self.assertNotEqual(result["status"], "completed")
+        self.assertNotIn("timeout", result["error"] or "")
+        self.assertIn("前端", result["error"] or "")
+        self.assertIn("(interrupted)", result["stdout"])
+        self.assertIn("行已消失", result["stdout"])
+        self.assertIn("后端好了", result["stdout"])
+        # 其它子任务不受影响：c1 仍是 completed，结果 intact。
+        c1 = _row(self.db_path, "c1")
+        self.assertEqual(c1["status"], "completed")
+        self.assertEqual(c1["result_text"], "后端好了")
+        # 协调员本体行仍在，且因有子任务未完成而整体 failed。
+        self.assertEqual(_row(self.db_path, "coord")["status"], "failed")
+
+    def test_all_completed_still_passes(self):
+        """(c) 正常等齐仍通过：行消失防线不改变正常合流语义。"""
+        _insert_task(self.db_path, "coord",
+                     coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c1", status="completed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="后端好了")
+        _insert_task(self.db_path, "c2", status="completed",
+                     parent_task_id="coord", coordinator_part="前端",
+                     result_text="前端好了")
+        _set_running(self.db_path, "coord")
+        worker = self._worker()
+        result = worker._run_coordinator(
+            worker.get_task("coord"), time.time(),
+            poll_interval=0.01, timeout=60)
+        self.assertEqual(result["status"], "completed")
+        self.assertIn("2/2", result["stdout"])
+        self.assertEqual(_row(self.db_path, "coord")["status"], "completed")
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -2895,9 +2895,11 @@ class Worker:
     def _coord_child_rows(self, child_ids):
         """子任务当前行：{id: {status, result_text, error_text}}。
 
-        已删或查不到的子任务不在结果里，调用方按缺失处理（等齐
-        判断里缺失不算终态，会一直等到超时——子任务行不应被删，
-        真删了就是超时失败，不静默当完成）。
+        已删或查不到的子任务按 interrupted 终态合成一行（error_text
+        注明行已消失）：等齐判断里缺失不再等到超时，合流报告里
+        该子任务标 interrupted 且整体不为 completed；孤儿标记不
+        会对不存在的行再 UPDATE（rowcount 本来就是 0）。行消失
+        不是完成，绝不能静默当 completed 计。
         """
         rows = {}
         conn = _db_connect(self.db_path)
@@ -2907,6 +2909,10 @@ class Worker:
                     "SELECT id, status, result_text, error_text FROM tasks"
                     " WHERE id = ?", (cid,)).fetchone()
                 if row is None:
+                    rows[cid] = {
+                        "status": "interrupted",
+                        "result_text": None,
+                        "error_text": "子任务行已消失（已删除）"}
                     continue
                 try:
                     rows[row["id"]] = {
@@ -3049,9 +3055,19 @@ class Worker:
                 self.heartbeat(task_id)
             except Exception:
                 pass
+            # 行消失防线：协调员本体行被删时 _was_cancelled 查不到
+            # 行会返回 False（get_task 为 None），不能因此等到
+            # deadline——行没了即按取消处理，立即跳出放槽；终局
+            # CAS 写不回去，最终状态按 interrupted 计（见循环后
+            # final_written 分支），不写回、不抛错、不当完成。
+            if self.get_task(task_id) is None:
+                cancelled = True
+                break
             if self._was_cancelled(task_id):
                 cancelled = True
                 break
+            # 等齐判断：_coord_child_rows 已把行消失的子任务合成
+            # interrupted 终态，缺失不再拖住循环等到超时。
             rows = self._coord_child_rows(child_ids)
             if all((rows.get(cid) or {}).get("status") in TERMINAL_STATUSES
                    for cid in child_ids):