admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tasks): 完整结果另存 result_text_full，详情不再只剩 2000 字

Browse Source 
旧链路三层截断：sandbox stdout 截 2000 字、worker 写回截
4000 字，长答案尾部在详情里永远查不到（只把上限调大是
把截断点后移，不是修掉）。修法：截断版保持不变（列表/
摘要用），完整版另存：
- sandbox _result_from_runner/_parse_host_result（local 与
  docker dispatch 路）回包新增 stdout_full（上限 200000 字，
  只防无界增长）；goal 路结论在尾部、tail 已保结论，不动。
- worker _write_execution_result 新增 result_text_full 列
  （_ensure_result_columns + server migrate_tasks_table 幂等
  补列），终局与取消分支均透传 stdout_full。
- server：任务 dict 加 result_full_available（不直接塞全文
  进列表负载），新增 GET /tasks/{id}/result-full（read 权限
  同款校验，无完整版时 404，不回摘要冒充完整）。
单测：test_queue_sandbox 45 项（含新增 stdout_full 1 项）、
test_marker_tools 30 项。
This commit is contained in:
host committed 2026-10-30 02:45:51 +08:00
1 parent 276d433da6
commit 92b7f8e8ff
4 files changed
+71 -6
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cloud/sandbox.py
+12
View File
@@ -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", "")),
cloud/server.py
+27
View File
@@ -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:
cloud/tests/test_queue_sandbox.py
+14
View File
@@ -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。
cloud/worker.py
+18 -6
View File
@@ -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)