diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 59e8ec8..55ac0f4 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -39,6 +39,16 @@ MAX_STDOUT_CHARS = 2000 # 状态 / usage / 标记块仍从完整 stdout 解析，不受此截断影响。 MAX_RESULT_CHARS = 20000 _RESULT_TRUNCATED_MARKER = "\n\n[结果已截断：原文超 20000 字，仅显示前 20000 字]" +# 完整结果存库上限：20000 字摘要之外，完整版另存 +# result_text_full，上限 200000 字只防无界增长，不作正常截断。 +MAX_FULL_STDOUT_CHARS = 200000 + + +def _full_stdout(text): + if text is None: + return "" + text = str(text) + return text[:MAX_FULL_STDOUT_CHARS] if len(text) > MAX_FULL_STDOUT_CHARS else text # 启动前分段日志（2026-10-29 任务 A）：docker run 是阻塞调用， # 容器输出要等返回才回传，中途卡死时 worker 侧没有任何标记； @@ -435,6 +445,7 @@ class LocalExecutor(Executor): "task_id": task_id, "status": status, "stdout": stdout, + "stdout_full": _full_stdout(stdout_full), "returncode": returncode, "host_status": host_status, # host.py 的 RuntimeError（如“任务已存在”）走 traceback 到 @@ -1835,6 +1846,7 @@ def _parse_host_result(task_id, fake): "task_id": task_id, "status": status, "stdout": stdout, + "stdout_full": _full_stdout(stdout_full), "returncode": returncode, "host_status": host_status, "stderr": _truncate_stdout(getattr(fake, "stderr", "")), diff --git a/cloud/server.py b/cloud/server.py index 2390c4d..517dc2c 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -1104,6 +1104,7 @@ def migrate_tasks_table(conn): "PRAGMA table_info(tasks)").fetchall()} new_cols = { "result_text": "TEXT", + "result_text_full": "TEXT", "error_text": "TEXT", "actual_model": "TEXT", "started_at": "REAL", @@ -1748,6 +1749,7 @@ def _task_to_dict(row, attachments=None, total_tokens=None): "updated_at": row["updated_at"], "last_heartbeat": row["last_heartbeat"], "result_text": _row_get(row, "result_text"), + "result_full_available": bool(_row_get(row, "result_text_full")), "error_text": _row_get(row, "error_text"), "actual_model": _row_get(row, "actual_model"), "started_at": _row_get(row, "started_at"), @@ -2277,6 +2279,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): user_id, rest[: -len("/trace/stream")]) elif rest.endswith("/trace") and method == "GET": self._task_trace(user_id, rest[: -len("/trace")]) + elif rest.endswith("/result-full") and method == "GET": + self._task_result_full(user_id, + rest[: -len("/result-full")]) elif rest.endswith("/agents") and method == "GET": self._task_agents(user_id, rest[: -len("/agents")]) elif rest.endswith("/cancel") and method == "POST": @@ -4626,6 +4631,28 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): for d in _trace_dirs_for_task(self.server, task)] return _read_trace_entries_multi(paths), "" + def _task_result_full(self, user_id, task_id): + """完整结果端点：result_text 仍是 4000 字摘要，完整版 + 在 result_text_full 里；无完整版时 404（老任务、goal + 任务、或本轮没拿到 stdout_full 的任务），不回摘要 + 冒充完整。权限与 _get_task 同款 read 校验。""" + row = self._fetch_task(task_id) + if row is None: + self._send_json(404, {"error": "task not found"}) + return + task = _task_to_dict(row) + with _auth_store(self.server.db_path) as store: + if not store.check_permission(user_id, task["project_id"], "read"): + self._send_json(403, {"error": "forbidden"}) + return + full = _row_get(row, "result_text_full") + if not full: + self._send_json(404, {"error": "result full not available"}) + return + self._send_json(200, {"task_id": task_id, + "status": task["status"], + "result_text_full": full}) + def _task_trace(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: diff --git a/cloud/tests/test_queue_sandbox.py b/cloud/tests/test_queue_sandbox.py index 5872122..1fa2e80 100644 --- a/cloud/tests/test_queue_sandbox.py +++ b/cloud/tests/test_queue_sandbox.py @@ -546,6 +546,20 @@ class TestUsageParsing(unittest.TestCase): self.assertEqual(result["usage"], usage) self.assertEqual(result["status"], "completed") + def test_result_from_runner_keeps_stdout_full(self): + # 截断版 stdout 仍是 2000 字，但 stdout_full 必须带完整 + # 文本，否则长答案尾部在库里永远查不到。 + fake = SimpleNamespace( + returncode=0, + stdout=json.dumps({"status": "completed", + "text": "y" * 5000}), + stderr="") + result = LocalExecutor("/fake/host.py", + runner=lambda cmd: fake + )._result_from_runner("t1", fake) + self.assertLessEqual(len(result["stdout"]), 2000) + self.assertIn("y" * 5000, result["stdout_full"]) + def test_result_from_runner_usage_survives_long_text(self): # usage 在 JSON 末尾：旧的截头解析在 text 很长时会丢 usage、 # 还会把 failed 误判成 completed；解析必须用完整 stdout。 diff --git a/cloud/worker.py b/cloud/worker.py index 0012990..db66dd8 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -1345,7 +1345,8 @@ class Worker: except Exception: return for name, coltype in { - "result_text": "TEXT", "error_text": "TEXT", + "result_text": "TEXT", "result_text_full": "TEXT", + "error_text": "TEXT", "actual_model": "TEXT", "started_at": "REAL", "finished_at": "REAL", "child_pid": "INTEGER", "executor": "TEXT", "pr_url": "TEXT", @@ -1439,7 +1440,11 @@ class Worker: error_text=None, actual_model=None, started_at=None, finished_at=None, child_pid=None, turn_id=None, - root_turn_id=None, only_if_status=None): + root_turn_id=None, only_if_status=None, + result_full=None): + # result_full：完整 stdout（dispatch/docker 路由 sandbox + # 的 stdout_full）；截断版 result_text 保持 4000 不变作 + # 摘要，完整版另存 result_text_full 供详情/端点取。 """把执行结果写进 tasks 新列；老库缺列时先迁移再写。 turn_id/root_turn_id 只在新值非空时覆盖：某轮没产出 @@ -1461,14 +1466,17 @@ class Worker: if only_if_status is not None: status_clause = " AND status = ?" extra = (only_if_status,) + full = (result_full[:200000] if result_full else None) if child_pid is None: cur = conn.execute( - "UPDATE tasks SET result_text = ?, error_text = ?," + "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, @@ -1476,13 +1484,15 @@ class Worker: ) else: cur = conn.execute( - "UPDATE tasks SET result_text = ?, error_text = ?," + "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, @@ -5072,7 +5082,8 @@ class Worker: started_at=started_at, finished_at=time.time(), turn_id=(result or {}).get("turn_id"), root_turn_id=(result or {}).get("root_turn_id"), - only_if_status="interrupted") + only_if_status="interrupted", + result_full=(result or {}).get("stdout_full")) # P1：取消分支先记一条成本再返回（与配额中断收尾 # 同款记账点）。旧版这里直接返回 cost_record_id=None， # 被取消前已花的模型费用就从账上丢了；取消不等于没 @@ -5160,7 +5171,8 @@ class Worker: finished_at=finished_at, turn_id=(result or {}).get("turn_id"), root_turn_id=(result or {}).get("root_turn_id"), - only_if_status=final) + only_if_status=final, + result_full=(result or {}).get("stdout_full")) # 附件暂存清理：任务终局后删 bundle 目录（重试走已存在 # 续跑，文件已在 worktree 里，不再需要暂存）。 self._cleanup_task_seed(task)