Merge pull request #162 from openqj/feat-task-result-full
feat(tasks): 完整结果另存 result_text_full,详情不再只剩 2000 字
This commit is contained in:
4 files changed
+107
-22
No files matched your search
@@ -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", "")),
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -546,6 +546,21 @@ class TestUsageParsing(unittest.TestCase):
|
||||
self.assertEqual(result["usage"], usage)
|
||||
self.assertEqual(result["status"], "completed")
|
||||
|
||||
def test_result_from_runner_keeps_stdout_full(self):
|
||||
# #165 后 stdout 是解析出的 text 截到 20000 字(不再是
|
||||
# 2000 字截 JSON),stdout_full 仍必须带完整文本,
|
||||
# 否则超 20000 字的长答案尾部在库里永远查不到。
|
||||
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"]), 21000)
|
||||
self.assertIn("y" * 5000, result["stdout_full"])
|
||||
|
||||
def test_result_from_runner_usage_survives_long_text(self):
|
||||
# usage 在 JSON 末尾:旧的截头解析在 text 很长时会丢 usage、
|
||||
# 还会把 failed 误判成 completed;解析必须用完整 stdout。
|
||||
|
||||
+53
-22
@@ -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,33 +1466,57 @@ 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)
|
||||
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(
|
||||
"UPDATE tasks SET result_text = ?, error_text = ?,"
|
||||
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,
|
||||
(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(
|
||||
"UPDATE tasks SET result_text = ?, error_text = ?,"
|
||||
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,
|
||||
(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:
|
||||
@@ -5072,7 +5101,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 +5190,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)
|
||||
|
||||
Reference in new issue
Block a user