tzt
184c11199a
- AgentService._history_from_session：从全部轮次的 write_file/edit_file 工具调用提取已读写文件清单（去重/单路径裁 120 字符/封顶 20 个），以摘要轮 注入对话上下文——轮次滑出窗口后事实清单仍在（丢上下文不丢事实） - Workspace.rollup：done 折叠行追加（产出: <artifact 锚点>），上下文折叠后 产物事实在 a:// 锚点体系内仍可寻址；既有子串断言全部兼容 - 新增 tests/test_branch_manifest.py 4 项
74 lines
3.2 KiB
Python
74 lines
3.2 KiB
Python
"""折叠摘要携带文件清单（T-M2，采纳 pi branch-summarization 设计）。"""
|
||
import gateway.agent as ag
|
||
from gateway.agent import AgentSession, AgentService
|
||
from router_system.workspace import Workspace
|
||
|
||
|
||
def _mk_turn(rid, task, tool_calls=None, state="done", response="好"):
|
||
return {"request_id": rid, "task": task, "response": response,
|
||
"state": state, "tool_calls": tool_calls or [], "tokens": 10,
|
||
"error": None, "ts": 0.0}
|
||
|
||
|
||
def _sess(turns):
|
||
return AgentSession({"id": "as-x", "title": "s", "turns": turns})
|
||
|
||
|
||
def test_history_carries_file_manifest_from_out_of_window_turns():
|
||
"""文件清单来自全部轮次：写文件轮滑出窗口后清单仍在摘要里。"""
|
||
turns = [
|
||
_mk_turn("t1", "建笔记", tool_calls=[
|
||
{"id": "c1", "name": "write_file", "arguments": {"path": "notes.md"}},
|
||
]),
|
||
_mk_turn("t2", "改笔记", tool_calls=[
|
||
{"id": "c2", "name": "edit_file", "arguments": {"path": "notes.md"}},
|
||
]),
|
||
_mk_turn("t3", "闲聊"),
|
||
]
|
||
hist = AgentService._history_from_session(_sess(turns), max_turns=1)
|
||
flat = "".join(str(m.get("content")) for m in hist)
|
||
# 窗口只剩 t3，但文件清单（t1/t2 产生）随摘要进入上下文
|
||
assert "notes.md" in flat
|
||
assert "上下文摘要" in flat
|
||
# 重复路径去重
|
||
assert flat.count("notes.md") >= 1
|
||
|
||
|
||
def test_history_without_file_tools_has_no_manifest():
|
||
"""无写文件工具调用 -> 不注入摘要轮（零噪音）。"""
|
||
turns = [_mk_turn("t1", "问个问题", tool_calls=[
|
||
{"id": "c1", "name": "web_fetch", "arguments": {"url": "https://x"}},
|
||
])]
|
||
hist = AgentService._history_from_session(_sess(turns), max_turns=6)
|
||
assert all("上下文摘要" not in str(m.get("content")) for m in hist)
|
||
|
||
|
||
def test_manifest_caps_path_length_and_count():
|
||
"""路径裁剪 120 字符、清单封顶 20 个，防摘要膨胀。"""
|
||
turns = [_mk_turn(f"t{i}", "写", tool_calls=[
|
||
{"id": f"c{i}", "name": "write_file", "arguments": {"path": f"很长的路径{i}" * 30}},
|
||
]) for i in range(25)]
|
||
hist = AgentService._history_from_session(_sess(turns), max_turns=0)
|
||
manifest = next(m for m in hist if "上下文摘要" in str(m.get("content")))
|
||
content = str(manifest["content"])
|
||
assert "等共 25 个" in content
|
||
body = content.split("：", 1)[1] # 剥掉固定前缀后逐段校验单路径裁剪
|
||
for seg in body.split("（等共")[0].split("、"):
|
||
assert len(seg) <= 120
|
||
|
||
|
||
def test_rollup_line_carries_artifact_anchor():
|
||
"""workspace rollup 折叠行携带产物锚点（a:// 体系内可寻址）。"""
|
||
ws = Workspace.new(request_id="abc123def456", query="写个快排",
|
||
api_token_cap=8000, rounds_cap=6)
|
||
ws.apply_brief({
|
||
"goal": "快排", "constraints": [], "tags": ["code"],
|
||
"acceptance": [{"id": "a1", "check": "可运行", "machine_checkable": True}],
|
||
"plan": [{"id": "s1", "task": "实现", "deps": [], "done_criteria": "过"}],
|
||
})
|
||
ws.add_progress("s1", "done", "完成", "a://s1_main.py")
|
||
ws.rollup()
|
||
archive = "".join(ws["archive"])
|
||
assert "s1" in archive
|
||
assert "（产出: a://s1_main.py）" in archive
|