host
ce85043a24
2026-10-29 实弹：468 字 final_answer 在链路里只剩前 300 字、 断在 fu 后，控制台照原样当助手回复展示，用户误以为模型 没写完。截断本身是设计（防链路文件写爆），缺的是标记。 - host.py _summarize_trace_item：agentMessage 超 300 字时 条目带 truncated=True，短消息不带，完整回复仍在任务 result_text 里，不在链路条目里。 - 控制台：TraceEntry 加 truncated 字段，mapTrace 与 traceLineText 在被截断的消息后标注“链路摘要已截断至 300 字，完整回复见右侧任务结果”，console.html 为构建产物同步。 - 单测：tests/test_host_trace.py 新增截断标记用例（468 字 截 300 字且带标记、短消息不带标记）。 验证：host 10 项、cloud test_task_trace 14 项全过； React tsc+vite 构建通过，产物 322151 字节、含截断标注、 含 #150 的 total_tokens/首字/总耗时，无回退。
196 lines
8.7 KiB
Python
196 lines
8.7 KiB
Python
"""实时链路（host 侧）单测：通知摘要器 + 链路文件追加。
|
||
|
||
不启动 app-server：Server 实例用 object.__new__ 绕开 __init__
|
||
（__init__ 会拉起 codex 子进程），只测与进程无关的方法。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
import host
|
||
from host import Server, TRACE_MAX_LINES, _summarize_trace_item
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
import host
|
||
from host import Server, TRACE_MAX_LINES, _summarize_trace_item
|
||
|
||
|
||
def _msg(item, ts=1700000000000):
|
||
return {"method": "item/completed",
|
||
"params": {"item": item, "completedAtMs": ts}}
|
||
|
||
|
||
class SummarizeTraceItemTestCase(unittest.TestCase):
|
||
def test_exec_truncates_command(self):
|
||
entries = _summarize_trace_item(
|
||
"item/completed",
|
||
{"item": {"type": "commandExecution", "id": "c1",
|
||
"command": "x" * 600, "cwd": "/w",
|
||
"exitCode": 0, "durationMs": 12},
|
||
"completedAtMs": 1700000000000})
|
||
self.assertEqual(len(entries), 1)
|
||
self.assertEqual(entries[0]["kind"], "exec")
|
||
self.assertEqual(len(entries[0]["command"]), 500)
|
||
self.assertEqual(entries[0]["exitCode"], 0)
|
||
self.assertEqual(entries[0]["ts"], 1700000000000)
|
||
|
||
def test_file_change_split_per_change(self):
|
||
entries = _summarize_trace_item(
|
||
"item/completed",
|
||
{"item": {"type": "fileChange", "id": "f1", "status": "completed",
|
||
"changes": [{"path": "a.py", "kind": "update"},
|
||
{"path": "b.py", "kind": "add"}]},
|
||
"startedAtMs": 1700000000001})
|
||
self.assertEqual([e["kind"] for e in entries], ["file", "file"])
|
||
self.assertEqual(entries[0]["id"], "f1:a.py")
|
||
self.assertEqual(entries[1]["path"], "b.py")
|
||
|
||
def test_message_reasoning_tool_call_plan(self):
|
||
cases = [
|
||
({"type": "agentMessage", "id": "m1", "phase": "final",
|
||
"text": "done"}, "message"),
|
||
({"type": "reasoning", "id": "r1",
|
||
"summary": ["a", "b"]}, "reasoning"),
|
||
({"type": "mcpToolCall", "id": "t1", "server": "s",
|
||
"tool": "read", "status": "completed"}, "tool"),
|
||
({"type": "functionCallOutput", "id": "o1",
|
||
"name": "read_file"}, "call"),
|
||
({"type": "plan", "id": "p1", "text": "step"}, "plan"),
|
||
]
|
||
for item, kind in cases:
|
||
entries = _summarize_trace_item("item/completed",
|
||
{"item": item})
|
||
self.assertEqual(len(entries), 1)
|
||
self.assertEqual(entries[0]["kind"], kind)
|
||
|
||
def test_message_truncation_is_flagged(self):
|
||
# 2026-10-29 实弹：468 字 final_answer 在链路里只剩前
|
||
# 300 字、断在“fu”后，控制台照原样当助手回复展示，
|
||
# 用户误以为模型没写完。截断本身是设计（防链路文件
|
||
# 写爆），但必须带标记，短消息不带标记。
|
||
long_entries = _summarize_trace_item(
|
||
"item/completed",
|
||
{"item": {"type": "agentMessage", "id": "m-long",
|
||
"phase": "final_answer", "text": "x" * 468}})
|
||
self.assertEqual(len(long_entries[0]["text"]), 300)
|
||
self.assertTrue(long_entries[0].get("truncated"))
|
||
short_entries = _summarize_trace_item(
|
||
"item/completed",
|
||
{"item": {"type": "agentMessage", "id": "m-short",
|
||
"phase": "final_answer", "text": "done"}})
|
||
self.assertNotIn("truncated", short_entries[0])
|
||
|
||
def test_skipped_notifications(self):
|
||
self.assertEqual(_summarize_trace_item(
|
||
"item/completed", {"item": {"type": "userMessage", "id": "u1",
|
||
"text": "hi"}}), [])
|
||
self.assertEqual(_summarize_trace_item(
|
||
"item/completed", {"item": {"type": "agentMessage", "id": "m2",
|
||
"text": ""}}), [])
|
||
self.assertEqual(_summarize_trace_item("item/completed", {}), [])
|
||
self.assertEqual(_summarize_trace_item(
|
||
"item/completed", {"item": {"type": "unknownThing",
|
||
"id": "z"}}), [])
|
||
|
||
|
||
class TraceFileTestCase(unittest.TestCase):
|
||
def _bare_server(self):
|
||
srv = object.__new__(Server)
|
||
srv.trace_path = None
|
||
srv._trace_lines = 0
|
||
return srv
|
||
|
||
def test_set_trace_file_and_append(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
state_path = os.path.join(d, "host-state.json")
|
||
srv = self._bare_server()
|
||
srv.set_trace_file(state_path, "abc/def?id=1")
|
||
self.assertEqual(
|
||
srv.trace_path,
|
||
os.path.join(d, "host-traces", "abc-def-id-1.jsonl"))
|
||
srv._append_trace(_msg({"type": "commandExecution", "id": "c1",
|
||
"command": "ls", "cwd": "/w",
|
||
"exitCode": 0, "durationMs": 3}))
|
||
srv._append_trace(_msg({"type": "userMessage", "id": "u1",
|
||
"text": "hi"}))
|
||
with open(srv.trace_path, encoding="utf-8") as f:
|
||
lines = f.read().strip().split("\n")
|
||
self.assertEqual(len(lines), 1)
|
||
self.assertEqual(json.loads(lines[0])["kind"], "exec")
|
||
|
||
def test_set_trace_file_failure_keeps_none(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
# host-traces 已被普通文件占住，建目录失败时不抛、不影响回合。
|
||
blocker = os.path.join(d, "host-traces")
|
||
with open(blocker, "w", encoding="utf-8") as f:
|
||
f.write("x")
|
||
srv = self._bare_server()
|
||
srv.set_trace_file(os.path.join(d, "host-state.json"), "t1")
|
||
self.assertIsNone(srv.trace_path)
|
||
|
||
def test_append_stops_at_max_lines(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
srv = self._bare_server()
|
||
srv.trace_path = os.path.join(d, "t.jsonl")
|
||
srv._trace_lines = TRACE_MAX_LINES - 1
|
||
srv._append_trace(_msg({"type": "commandExecution", "id": "c1",
|
||
"command": "a"}))
|
||
srv._append_trace(_msg({"type": "commandExecution", "id": "c2",
|
||
"command": "b"}))
|
||
with open(srv.trace_path, encoding="utf-8") as f:
|
||
lines = f.read().strip().split("\n")
|
||
self.assertEqual(len(lines), 1)
|
||
self.assertEqual(json.loads(lines[0])["id"], "c1")
|
||
|
||
def test_summarize_carries_thread(self):
|
||
# 模型内子代理与主任务同文件时，服务端凭 thread 归属条目。
|
||
entries = _summarize_trace_item(
|
||
"item/completed",
|
||
{"item": {"type": "commandExecution", "id": "c1",
|
||
"command": "ls"},
|
||
"threadId": "th-sub", "completedAtMs": 1700000000000})
|
||
self.assertEqual(entries[0]["thread"], "th-sub")
|
||
entries = _summarize_trace_item(
|
||
"item/completed",
|
||
{"item": {"type": "commandExecution", "id": "c2",
|
||
"command": "ls"},
|
||
"completedAtMs": 1700000000000})
|
||
self.assertIsNone(entries[0]["thread"])
|
||
|
||
def test_append_trace_marker_to_parent_file(self):
|
||
# delegate 向父链路报到：路径清洗与 set_trace_file 一致，
|
||
# 写失败不抛（尽力而为，不影响回合）。
|
||
from host import _append_trace_marker, _trace_file_path
|
||
with tempfile.TemporaryDirectory() as d:
|
||
state_path = os.path.join(d, "host-state.json")
|
||
self.assertEqual(
|
||
_trace_file_path(state_path, "abc/def?id=1"),
|
||
os.path.join(d, "host-traces", "abc-def-id-1.jsonl"))
|
||
_append_trace_marker(
|
||
state_path, "parent task!",
|
||
{"kind": "agent-start", "id": "agent:rev",
|
||
"name": "rev", "prompt": "review"})
|
||
path = os.path.join(d, "host-traces", "parent-task-.jsonl")
|
||
with open(path, encoding="utf-8") as f:
|
||
entry = json.loads(f.read().strip())
|
||
self.assertEqual(entry["kind"], "agent-start")
|
||
self.assertEqual(entry["name"], "rev")
|
||
# 目录被文件占住时也不抛。
|
||
blocker = os.path.join(d, "host-traces")
|
||
shutil.rmtree(blocker)
|
||
with open(blocker, "w", encoding="utf-8") as f:
|
||
f.write("x")
|
||
_append_trace_marker(state_path, "parent task!",
|
||
{"kind": "agent-done", "id": "agent:rev:done",
|
||
"name": "rev", "status": "completed"})
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|