fix(worker): 无 result_text_full 列的老库/手工表写回回落，不拖垮终局

CI 失败根因：P0 取消守卫测试 mock 掉 _ensure_result_columns
并手工建表（无新列），带新列的 UPDATE 直接
OperationalError。修法：带新列写回失败且错误含
result_text_full 时回滚并用旧 SQL 重写（摘要照写、完整版
丢弃），其它错误原样重抛。

cloud/tests/test_queue_sandbox.py

@@ -547,8 +547,9 @@ class TestUsageParsing(unittest.TestCase):
         self.assertEqual(result["status"], "completed")
 
     def test_result_from_runner_keeps_stdout_full(self):
-        # 截断版 stdout 仍是 2000 字，但 stdout_full 必须带完整
-        # 文本，否则长答案尾部在库里永远查不到。
+        # #165 后 stdout 是解析出的 text 截到 20000 字（不再是
+        # 2000 字截 JSON），stdout_full 仍必须带完整文本，
+        # 否则超 20000 字的长答案尾部在库里永远查不到。
         fake = SimpleNamespace(
             returncode=0,
             stdout=json.dumps({"status": "completed",
@@ -557,7 +558,7 @@ class TestUsageParsing(unittest.TestCase):
         result = LocalExecutor("/fake/host.py",
                                runner=lambda cmd: fake
                                )._result_from_runner("t1", fake)
-        self.assertLessEqual(len(result["stdout"]), 2000)
+        self.assertLessEqual(len(result["stdout"]), 21000)
         self.assertIn("y" * 5000, result["stdout_full"])
 
     def test_result_from_runner_usage_survives_long_text(self):

cloud/worker.py

@@ -1467,37 +1467,56 @@ class Worker:
                 status_clause = " AND status = ?"
                 extra = (only_if_status,)
             full = (result_full[:200000] if result_full else None)
+            short = ((result_text or "")[:21000] if result_text else None)
+            err_short = ((error_text or "")[:4000]
+                         if error_text else None)
+            now = time.time()
             if child_pid is None:
-                cur = conn.execute(
+                sql_full = (
                     "UPDATE tasks SET result_text = ?,"
                     " result_text_full = ?, error_text = ?,"
                     " actual_model = ?, started_at = ?, finished_at = ?,"
                     " turn_id = COALESCE(?, turn_id),"
                     " root_turn_id = COALESCE(?, root_turn_id),"
-                    " updated_at = ? WHERE id = ?" + status_clause,
-                    ((result_text or "")[:21000] if result_text else None,
-                     full,
-                     (error_text or "")[:4000] if error_text else None,
-                     actual_model, started_at, finished_at,
-                     turn_id, root_turn_id,
-                     time.time(), task_id) + extra,
-                )
+                    " updated_at = ? WHERE id = ?" + status_clause)
+                params_full = (short, full, err_short, actual_model,
+                               started_at, finished_at, turn_id,
+                               root_turn_id, now, task_id) + extra
+                sql_old = sql_full.replace(
+                    " result_text_full = ?,", "")
+                params_old = (short, err_short, actual_model,
+                              started_at, finished_at, turn_id,
+                              root_turn_id, now, task_id) + extra
             else:
-                cur = conn.execute(
+                sql_full = (
                     "UPDATE tasks SET result_text = ?,"
                     " result_text_full = ?, error_text = ?,"
                     " actual_model = ?, started_at = ?, finished_at = ?,"
                     " child_pid = ?,"
                     " turn_id = COALESCE(?, turn_id),"
                     " root_turn_id = COALESCE(?, root_turn_id),"
-                    " updated_at = ? WHERE id = ?" + status_clause,
-                    ((result_text or "")[:21000] if result_text else None,
-                     full,
-                     (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) + extra,
-                )
+                    " updated_at = ? WHERE id = ?" + status_clause)
+                params_full = (short, full, err_short, actual_model,
+                               started_at, finished_at, child_pid,
+                               turn_id, root_turn_id, now, task_id) + extra
+                sql_old = sql_full.replace(
+                    " result_text_full = ?,", "")
+                params_old = (short, err_short, actual_model,
+                              started_at, finished_at, child_pid,
+                              turn_id, root_turn_id, now, task_id) + extra
+            try:
+                cur = conn.execute(sql_full, params_full)
+            except Exception as exc:
+                # 老库/被 mock 掉迁移的库没有 result_text_full 列：
+                # 完整版丢弃、摘要照写，绝不让新列拖垮终局写回
+                # （P0 取消守卫测试就是 mock 掉迁移的手工表）。
+                if "result_text_full" not in str(exc):
+                    raise
+                try:
+                    conn.rollback()
+                except Exception:
+                    pass
+                cur = conn.execute(sql_old, params_old)
             conn.commit()
             return cur.rowcount == 1
         finally: