1324 lines
58 KiB
Python
1324 lines
58 KiB
Python
"""
|
||
tests/test_agent_core.py
|
||
========================
|
||
离线确定性测试：用脚本化 mock 流函数验证 core/agent 的 pi 1:1 语义。
|
||
运行：conda run -n haocode python -m pytest tests/test_agent_core.py -v
|
||
（无网络、无 Qt 依赖）
|
||
"""
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||
|
||
# P0-01：provider 用例经统一配置入口读取临时配置（测试 provider），不碰真实 data/config.json
|
||
from tests._test_env import isolate # noqa: E402
|
||
isolate("agentcore")
|
||
|
||
import pytest
|
||
|
||
from core.agent import (Agent, AgentConfig, AgentError, ModelConfig, RetryConfig)
|
||
from core.agent.compaction import (
|
||
compact_context, find_cut_point, find_turn_start, find_valid_cut_points,
|
||
prepare_compaction, serialize_conversation, format_file_operations,
|
||
FileOperations, extract_file_ops_from_message, compute_file_lists,
|
||
)
|
||
from core.agent.context import (clamp_max_tokens_to_context, should_compact,
|
||
estimate_context_tokens)
|
||
from core.agent.recovery import AgentRunner
|
||
from core.agent.stream_fn import to_openai_messages, from_openai_messages
|
||
from core.agent.tools import default_tools
|
||
from core.agent.types import (AgentMessage, ToolCall, AgentTool,
|
||
AgentToolResult)
|
||
|
||
|
||
# ======================================================================
|
||
# 脚本化 mock 流：每个 turn 一个 dict，按序消费
|
||
# ======================================================================
|
||
def make_scripted_stream(turns):
|
||
state = {"i": 0}
|
||
|
||
def stream_fn(context, model, signal, max_tokens, tools=None):
|
||
turn = turns[min(state["i"], len(turns) - 1)]
|
||
state["i"] += 1
|
||
content = turn.get("content", "")
|
||
reasoning = turn.get("reasoning", "")
|
||
tool_calls = turn.get("tool_calls", [])
|
||
chunks = max(1, turn.get("chunks", 3))
|
||
|
||
if turn.get("raise"):
|
||
raise turn["raise"]
|
||
|
||
if reasoning:
|
||
yield ("event", _thinking(reasoning))
|
||
for i in range(0, len(content), max(1, len(content) // chunks) if content else 1):
|
||
piece = content[i:i + max(1, len(content) // chunks) if content else 1]
|
||
if signal.aborted:
|
||
return # 中止：生成器提前结束（真实流行为）
|
||
yield ("event", _text(piece))
|
||
for tc in tool_calls:
|
||
yield ("event", _tc_delta(0, tc["id"], "id"))
|
||
yield ("event", _tc_delta(0, tc["name"], "name"))
|
||
yield ("event", _tc_delta(0, _json(tc["arguments"]), "arguments"))
|
||
|
||
usage = turn.get("usage", {"input": 10, "output": 10})
|
||
final = AgentMessage(
|
||
role="assistant", content=content, reasoning=reasoning,
|
||
tool_calls=[ToolCall(id=tc["id"], name=tc["name"],
|
||
arguments=tc["arguments"]) for tc in tool_calls],
|
||
stop_reason=turn.get("stop_reason", "stop"), usage=usage,
|
||
)
|
||
yield ("event", _done())
|
||
yield ("final", final)
|
||
|
||
return stream_fn
|
||
|
||
|
||
def _text(t):
|
||
from core.agent.types import AssistantMessageEvent
|
||
return AssistantMessageEvent.text_delta(t)
|
||
|
||
|
||
def _thinking(t):
|
||
from core.agent.types import AssistantMessageEvent
|
||
return AssistantMessageEvent.thinking_delta(t)
|
||
|
||
|
||
def _tc_delta(idx, d, field):
|
||
from core.agent.types import AssistantMessageEvent
|
||
return AssistantMessageEvent.toolcall_delta(idx, d, field=field)
|
||
|
||
|
||
def _done():
|
||
from core.agent.types import AssistantMessageEvent
|
||
return AssistantMessageEvent.done()
|
||
|
||
|
||
def _json(o):
|
||
import json
|
||
return json.dumps(o, ensure_ascii=False)
|
||
|
||
|
||
def make_agent(turns, **cfg_kw):
|
||
model = ModelConfig(provider="mock", name="mock-model",
|
||
context_window=cfg_kw.pop("context_window", 100000),
|
||
max_tokens=cfg_kw.pop("max_tokens", 32768),
|
||
api_key="x", base_url="http://mock")
|
||
cfg = AgentConfig(model=model, tools=cfg_kw.pop("tools", []), **cfg_kw)
|
||
agent = Agent(cfg)
|
||
agent.set_stream_fn(make_scripted_stream(turns))
|
||
return agent
|
||
|
||
|
||
def run_events(agent):
|
||
events = []
|
||
agent.subscribe(lambda e: events.append(e))
|
||
return events
|
||
|
||
|
||
# ======================================================================
|
||
# 1. 基础对话 + 事件顺序（对照 pi 事件流）
|
||
# ======================================================================
|
||
def test_basic_chat_event_order():
|
||
agent = make_agent([{"content": "你好呀！", "chunks": 4}])
|
||
events = run_events(agent)
|
||
result = agent.prompt("hi")
|
||
types = [e.type for e in events]
|
||
assert types[0] == "agent_start"
|
||
assert types[1] == "turn_start"
|
||
assert "message_start" in types and "message_end" in types
|
||
assert types.count("message_update") == 4 # 4 个 chunk → 4 次增量
|
||
assert types[-1] == "agent_end"
|
||
assert result.stop_reason == "stop"
|
||
assert agent.state.messages[0].role == "user"
|
||
assert agent.state.messages[1].role == "assistant"
|
||
assert agent.state.messages[1].content == "你好呀！"
|
||
assert not agent.state.is_streaming
|
||
|
||
|
||
# ======================================================================
|
||
# 2. 工具循环：bash 调用 → 执行 → 第二轮回答（对照 pi tool 管线）
|
||
# ======================================================================
|
||
def test_tool_loop_executes_and_continues():
|
||
agent = make_agent(
|
||
[{"tool_calls": [{"id": "call_1", "name": "bash",
|
||
"arguments": {"command": "echo hi-from-tool"}}],
|
||
"stop_reason": "stop"},
|
||
{"content": "命令执行完了。"}],
|
||
tools=default_tools(),
|
||
)
|
||
events = run_events(agent)
|
||
result = agent.prompt("执行一下")
|
||
types = [e.type for e in events]
|
||
assert "tool_execution_start" in types
|
||
assert "tool_execution_end" in types
|
||
# 工具结果消息进了上下文
|
||
trs = [m for m in agent.state.messages if m.role == "toolResult"]
|
||
assert len(trs) == 1
|
||
assert trs[0].tool_name == "bash"
|
||
tr_text = trs[0].content if isinstance(trs[0].content, str) else \
|
||
"".join(c.get("text", "") for c in trs[0].content)
|
||
assert "hi-from-tool" in tr_text
|
||
# 两轮：2 个 assistant
|
||
assistants = [m for m in agent.state.messages if m.role == "assistant"]
|
||
assert len(assistants) == 2
|
||
assert assistants[1].content == "命令执行完了。"
|
||
assert result.stop_reason == "stop"
|
||
# end 事件带错误标记=False
|
||
end_evt = [e for e in events if e.type == "tool_execution_end"][0]
|
||
assert end_evt.is_error is False
|
||
|
||
|
||
# ======================================================================
|
||
# 3. length 截断 + 工具调用 → 一律失败不执行（对照 failToolCalls）
|
||
# ======================================================================
|
||
def test_length_truncation_fails_tool_calls():
|
||
agent = make_agent(
|
||
[{"tool_calls": [{"id": "c1", "name": "bash",
|
||
"arguments": {"command": "echo should-not-run"}}],
|
||
"stop_reason": "length"},
|
||
{"content": "恢复回答。"}],
|
||
tools=default_tools(),
|
||
)
|
||
events = run_events(agent)
|
||
result = agent.prompt("go")
|
||
trs = [m for m in agent.state.messages if m.role == "toolResult"]
|
||
assert len(trs) == 1
|
||
assert trs[0].is_error is True
|
||
tr_text = trs[0].content if isinstance(trs[0].content, str) else \
|
||
"".join(c.get("text", "") for c in trs[0].content)
|
||
assert "未执行" in tr_text
|
||
# 循环继续（hasMoreToolCalls=True）→ 第二轮
|
||
assistants = [m for m in agent.state.messages if m.role == "assistant"]
|
||
assert len(assistants) == 2
|
||
assert result.stop_reason == "stop"
|
||
|
||
|
||
# ======================================================================
|
||
# 4. 可恢复 length：删除坏消息 + 压缩 + 重试一次（对照 isRecoverableLength）
|
||
# ======================================================================
|
||
def test_recoverable_length_compact_retry():
|
||
# 预填 30 条大历史（~60K token）使 1:1 切点（保留 20K）有可摘要内容
|
||
agent = make_agent(
|
||
[{"content": "", "stop_reason": "length",
|
||
"usage": {"input": 100, "output": 50}}, # 50 < max_tokens → 可恢复
|
||
{"content": "压缩后恢复。"}],
|
||
max_tokens=100,
|
||
)
|
||
for i in range(30):
|
||
agent.state.messages.append(
|
||
AgentMessage(role="user",
|
||
content=f"历史消息 {i} " + "长" * 2000))
|
||
summarize_calls = []
|
||
|
||
def summarize(prompt, system_prompt, max_tokens):
|
||
summarize_calls.append((prompt, system_prompt, max_tokens))
|
||
return "SUMMARY-OK"
|
||
|
||
runner = AgentRunner(agent, summarize_fn=summarize)
|
||
events = run_events(agent)
|
||
result = runner.run("新问题")
|
||
assert result.stop_reason == "stop"
|
||
assert runner.last_action == "length_compact"
|
||
assert len(summarize_calls) == 1
|
||
# pi 原版摘要提示词（逐字）进入了调用
|
||
prompt, system, mt = summarize_calls[0]
|
||
assert "<conversation>" in prompt and "## Goal" in prompt
|
||
assert "context summarization assistant" in system
|
||
assert mt == min(int(0.8 * 16384), 100) # min(0.8×reserve, model.maxTokens)
|
||
# 上下文里有压缩摘要（kind 标记，对照 pi compaction 条目）
|
||
sums = [m for m in agent.state.messages if m.kind == "compaction_summary"]
|
||
assert len(sums) == 1 and "SUMMARY-OK" in sums[0].content
|
||
# 保留尾巴非空且不含被摘要的历史
|
||
assert len(agent.state.messages) > 1
|
||
# 坏的 length 助手消息被移除（只剩恢复后的那个）
|
||
assistants = [m for m in agent.state.messages if m.role == "assistant"]
|
||
assert all(m.stop_reason != "length" for m in assistants)
|
||
|
||
|
||
# ======================================================================
|
||
# 5. 可重试错误：退避重试（对照 isRetryable + 1600ms×1.6^n）
|
||
# ======================================================================
|
||
def test_retryable_error_backoff_retry():
|
||
agent = make_agent(
|
||
[{"raise": AgentError(message="429 Too Many Requests",
|
||
kind="rate_limit", status_code=429,
|
||
recoverable=True)},
|
||
{"content": "重试成功。"}],
|
||
retry=RetryConfig(max_attempts=3, base_delay_ms=10, factor=2.0,
|
||
max_delay_ms=50),
|
||
)
|
||
runner = AgentRunner(agent)
|
||
result = runner.run("hi")
|
||
assert result.stop_reason == "stop"
|
||
assert runner._retry_attempt == 1
|
||
assert runner.last_action == "retry"
|
||
assert agent.state.messages[-1].content == "重试成功。"
|
||
|
||
|
||
# ======================================================================
|
||
# 6. 不可重试错误：立即结束（401 auth）
|
||
# ======================================================================
|
||
def test_auth_error_stops_immediately():
|
||
agent = make_agent(
|
||
[{"raise": AgentError(message="401 Unauthorized", kind="auth",
|
||
status_code=401)}])
|
||
runner = AgentRunner(agent,
|
||
summarize_fn=lambda *a: "S") # 压缩可用也不该触发
|
||
result = runner.run("hi")
|
||
assert result.stop_reason == "error"
|
||
assert result.error.kind == "auth"
|
||
assert runner.last_action in ("none",)
|
||
|
||
|
||
# ======================================================================
|
||
# 7. steering：turn 边界注入（对照 getSteeringMessages）
|
||
# ======================================================================
|
||
def test_steering_injected_at_turn_boundary():
|
||
agent = make_agent(
|
||
[{"content": "第一段回答。"},
|
||
{"content": "看到转向了。"}],
|
||
)
|
||
events = run_events(agent)
|
||
# 第一轮的 message_end 之后注入 steering（只注入一次）
|
||
steered = {"done": False}
|
||
|
||
def on_event(e):
|
||
if e.type == "message_end" and e.message is not None \
|
||
and e.message.role == "assistant" and not steered["done"]:
|
||
steered["done"] = True
|
||
agent.steer("插一句：改方向")
|
||
agent.subscribe(on_event)
|
||
result = agent.prompt("原始问题")
|
||
contents = [m.content for m in agent.state.messages if m.role == "user"]
|
||
assert "插一句：改方向" in contents
|
||
assistants = [m for m in agent.state.messages if m.role == "assistant"]
|
||
assert len(assistants) == 2
|
||
assert result.stop_reason == "stop"
|
||
|
||
|
||
# ======================================================================
|
||
# 8. followUp：run 结束前注入 → 自动续跑（对照外层循环）
|
||
# ======================================================================
|
||
def test_follow_up_continues_run():
|
||
agent = make_agent(
|
||
[{"content": "第一段。"},
|
||
{"content": "第二段（followUp 触发）。"}],
|
||
)
|
||
seen = {"n": 0}
|
||
|
||
def on_event(e):
|
||
if e.type == "message_end" and e.message is not None \
|
||
and e.message.role == "assistant":
|
||
seen["n"] += 1
|
||
if seen["n"] == 1:
|
||
agent.follow_up("追加一问")
|
||
agent.subscribe(on_event)
|
||
result = agent.prompt("第一问")
|
||
user_msgs = [m for m in agent.state.messages if m.role == "user"]
|
||
assert len(user_msgs) == 2
|
||
assistants = [m for m in agent.state.messages if m.role == "assistant"]
|
||
assert len(assistants) == 2
|
||
assert result.stop_reason == "stop"
|
||
|
||
|
||
# ======================================================================
|
||
# 9. abort：流中中止 → stop_reason=aborted → agent_end（对照 pi 中止语义）
|
||
# ======================================================================
|
||
def test_abort_mid_stream():
|
||
agent = make_agent([{"content": "ABCDEFGH", "chunks": 8},
|
||
{"content": "不应到达"}])
|
||
events = run_events(agent)
|
||
fired = {"n": 0}
|
||
|
||
def on_event(e):
|
||
if e.type == "message_update":
|
||
fired["n"] += 1
|
||
if fired["n"] == 2: # 收到第 2 个 chunk 后中止
|
||
agent.abort()
|
||
agent.subscribe(on_event)
|
||
result = agent.prompt("hi")
|
||
assert result.stop_reason == "aborted"
|
||
types = [e.type for e in events]
|
||
assert types[-1] == "agent_end"
|
||
# 中止消息保留已输出部分
|
||
last_a = [m for m in agent.state.messages if m.role == "assistant"][-1]
|
||
assert last_a.stop_reason == "aborted"
|
||
assert "AB" in last_a.content
|
||
# 第二轮未发生
|
||
assert len([m for m in agent.state.messages if m.role == "assistant"]) == 1
|
||
|
||
|
||
# ======================================================================
|
||
# 10. 输出预算钳制公式（对照 simple-options clampMaxTokensToContext）
|
||
# ======================================================================
|
||
def test_clamp_max_tokens_formula():
|
||
model = ModelConfig(context_window=100000, max_tokens=32768)
|
||
# 构造已知输入：50000 个 ASCII 字符 ≈ 12500 token（/4）+ 每消息 4
|
||
msg = AgentMessage(role="user", content="a" * 50000)
|
||
clamped, input_tokens = clamp_max_tokens_to_context(model, [msg])
|
||
# 12500 + 4 = 12504；max_output = 100000 - 12504 - 4096 = 833... 不，= 83400
|
||
assert clamped == min(32768, 100000 - input_tokens - 4096)
|
||
# 输入几乎占满窗口 → 输出预算被压到 1
|
||
big = AgentMessage(role="user", content="a" * 400000) # 100004 token
|
||
assert clamp_max_tokens_to_context(model, [big]) is None # 溢出
|
||
mid = AgentMessage(role="user", content="a" * 392000) # 98004 token
|
||
clamped2, _ = clamp_max_tokens_to_context(model, [mid])
|
||
assert clamped2 == 1 # 100000-98004-4096 < 0 → 下限 1
|
||
|
||
|
||
# ======================================================================
|
||
# 11. 压缩切分点（1:1 对照 pi findCutPoint）：token 预算回扫 + 有效切点 + 断轮
|
||
# ======================================================================
|
||
def test_compaction_split_point_alignment():
|
||
msgs = [AgentMessage(role="user", content=f"u{i}") for i in range(4)]
|
||
a = AgentMessage(role="assistant", content="", tool_calls=[
|
||
ToolCall(id="c1", name="bash", arguments={})])
|
||
msgs.append(a)
|
||
msgs.append(AgentMessage(role="toolResult", tool_call_id="c1",
|
||
tool_name="bash", content="r"))
|
||
msgs.extend([AgentMessage(role="assistant", content=f"a{i}") for i in range(4)])
|
||
# 有效切点：user/assistant 位置；toolResult(index 5) 绝不能做切点
|
||
cuts = find_valid_cut_points(msgs, 0, len(msgs))
|
||
assert 5 not in cuts
|
||
assert set(cuts) == {0, 1, 2, 3, 4, 6, 7, 8, 9}
|
||
# 小预算 → 尾部回扫立刻达标 → 切在 index 9（assistant）→ 断轮
|
||
cut = find_cut_point(msgs, 0, len(msgs), keep_recent_tokens=1)
|
||
assert cut.first_kept_index == 9
|
||
assert cut.is_split_turn is True
|
||
# 轮起点 = 向前最近的 user（index 3）
|
||
assert cut.turn_start_index == 3
|
||
assert find_turn_start(msgs, 9, 0) == 3
|
||
# 大预算（够不着）→ 切点落在第一个有效切点 index 0（user）→ 不断轮
|
||
cut2 = find_cut_point(msgs, 0, len(msgs), keep_recent_tokens=10 ** 6)
|
||
assert cut2.first_kept_index == 0 and cut2.is_split_turn is False
|
||
|
||
|
||
# ======================================================================
|
||
# 11b. should_compact 触发公式（1:1 对照 pi: tokens > window - reserve）
|
||
# ======================================================================
|
||
def test_should_compact_formula():
|
||
model = ModelConfig(context_window=100000, max_tokens=32768)
|
||
# 330000 ASCII 字符 ≈ 82500 + 4 = 82504 token < 100000-16384=83616 → 不触发
|
||
small = AgentMessage(role="user", content="a" * 330000)
|
||
ok, t1 = should_compact([small], model, reserve_tokens=16384)
|
||
assert not ok and t1 == 82504
|
||
# 335000 ASCII ≈ 83754 > 83616 → 触发
|
||
big = AgentMessage(role="user", content="a" * 335000)
|
||
ok2, t2 = should_compact([big], model, reserve_tokens=16384)
|
||
assert ok2 and t2 == 83754
|
||
|
||
|
||
# ======================================================================
|
||
# 11c. usage 锚定估算（1:1 对照 pi estimateContextTokens）
|
||
# ======================================================================
|
||
def test_usage_anchored_estimation():
|
||
a = AgentMessage(role="assistant", content="x" * 100,
|
||
stop_reason="stop", usage={"totalTokens": 5000})
|
||
tail = AgentMessage(role="user", content="b" * 4000) # 1000 + 4 = 1004
|
||
est = estimate_context_tokens([a, tail])
|
||
assert est.usage_tokens == 5000
|
||
assert est.trailing_tokens == 1004
|
||
assert est.tokens == 6004
|
||
assert est.last_usage_index == 0
|
||
# aborted/error 的 assistant usage 无效（对照 pi getAssistantUsage）
|
||
a2 = AgentMessage(role="assistant", content="y",
|
||
stop_reason="error", usage={"totalTokens": 9999})
|
||
est2 = estimate_context_tokens([a2, tail])
|
||
assert est2.last_usage_index is None
|
||
assert est2.tokens == 1004 + (1 + 4)
|
||
|
||
|
||
# ======================================================================
|
||
# 11d. serialize_conversation 格式（1:1 对照 pi utils.ts）
|
||
# ======================================================================
|
||
def test_serialize_conversation_format():
|
||
msgs = [
|
||
AgentMessage(role="user", content="做个工具"),
|
||
AgentMessage(role="assistant", reasoning="想一想…", content="好，",
|
||
tool_calls=[ToolCall(id="c1", name="bash",
|
||
arguments={"command": "ls"})]),
|
||
AgentMessage(role="toolResult", tool_call_id="c1", tool_name="bash",
|
||
content="R" * 3000),
|
||
]
|
||
s = serialize_conversation(msgs)
|
||
assert "[User]: 做个工具" in s
|
||
assert "[Assistant thinking]: 想一想…" in s
|
||
assert "[Assistant]: 好，" in s
|
||
assert "[Assistant tool calls]: bash(command=\"ls\")" in s
|
||
# 工具结果 2000 字符截断（pi TOOL_RESULT_MAX_CHARS）
|
||
assert "[Tool result]: " + "R" * 2000 in s
|
||
assert "[... 1000 more characters truncated]" in s
|
||
|
||
|
||
# ======================================================================
|
||
# 11e. 文件操作提取与附录（1:1 对照 pi extractFileOps*/formatFileOperations）
|
||
# ======================================================================
|
||
def test_file_ops_extraction_and_appendix():
|
||
ops = FileOperations()
|
||
extract_file_ops_from_message(AgentMessage(
|
||
role="assistant", tool_calls=[
|
||
ToolCall(id="1", name="read", arguments={"path": "/a/f.py"}),
|
||
ToolCall(id="2", name="write", arguments={"path": "/a/g.py"}),
|
||
ToolCall(id="3", name="edit", arguments={"path": "/a/f.py"}),
|
||
]), ops)
|
||
read_files, modified = compute_file_lists(ops)
|
||
assert read_files == [] # f.py 被改过 → 不算只读
|
||
assert modified == ["/a/f.py", "/a/g.py"]
|
||
appendix = format_file_operations(read_files, modified)
|
||
assert "<modified-files>\n/a/f.py\n/a/g.py\n</modified-files>" in appendix
|
||
# 压缩后摘要尾部带文件附录
|
||
calls = []
|
||
|
||
def summarize(prompt, system, mt):
|
||
calls.append((prompt, mt))
|
||
return "S"
|
||
msgs = [AgentMessage(role="user", content="u" + "x" * 40000),
|
||
AgentMessage(role="assistant",
|
||
tool_calls=[ToolCall(id="1", name="read",
|
||
arguments={"path": "/a/f.py"})]),
|
||
AgentMessage(role="user", content="u2" + "x" * 40000),
|
||
AgentMessage(role="user", content="u3" + "x" * 40000)]
|
||
out = compact_context(msgs, ModelConfig(max_tokens=8192), summarize)
|
||
assert out is not None
|
||
assert out[0].kind == "compaction_summary"
|
||
assert "<read-files>\n/a/f.py\n</read-files>" in out[0].content
|
||
|
||
|
||
# ======================================================================
|
||
# 11f. 迭代式 previousSummary 更新（1:1 对照 pi UPDATE_SUMMARIZATION_PROMPT）
|
||
# ======================================================================
|
||
def test_iterative_summary_update():
|
||
calls = []
|
||
|
||
def summarize(prompt, system, mt):
|
||
calls.append(prompt)
|
||
return "UPDATED"
|
||
prev = AgentMessage(role="user", content="旧摘要内容ABC",
|
||
kind="compaction_summary")
|
||
msgs = [prev,
|
||
AgentMessage(role="user", content="n" + "x" * 40000),
|
||
AgentMessage(role="user", content="n2" + "x" * 40000),
|
||
AgentMessage(role="user", content="n3" + "x" * 40000)]
|
||
out = compact_context(msgs, ModelConfig(max_tokens=8192), summarize)
|
||
assert out is not None and out[0].kind == "compaction_summary"
|
||
assert "UPDATED" in out[0].content
|
||
# 旧摘要进 <previous-summary>，且用 UPDATE 提示词
|
||
p = calls[0]
|
||
assert "<previous-summary>\n旧摘要内容ABC\n</previous-summary>" in p
|
||
assert "NEW conversation messages to incorporate" in p
|
||
|
||
|
||
# ======================================================================
|
||
# 11g. 断轮双摘要拼接（1:1 对照 pi compact() 的 split-turn 分支）
|
||
# ======================================================================
|
||
def test_split_turn_double_summary():
|
||
calls = []
|
||
|
||
def summarize(prompt, system, mt):
|
||
calls.append(prompt)
|
||
return f"SUMMARY-{len(calls)}"
|
||
# 轮1: u1/a1（40K 各一）；轮2: u2/a2/a3/a4（a 连续 → 切点必落 assistant → 断轮）
|
||
msgs = [AgentMessage(role="user", content="u1" + "x" * 40000),
|
||
AgentMessage(role="assistant", content="a1" + "x" * 40000),
|
||
AgentMessage(role="user", content="u2" + "x" * 40000),
|
||
AgentMessage(role="assistant", content="a2" + "x" * 40000),
|
||
AgentMessage(role="assistant", content="a3" + "x" * 40000),
|
||
AgentMessage(role="assistant", content="a4" + "x" * 40000)]
|
||
out = compact_context(msgs, ModelConfig(max_tokens=8192), summarize)
|
||
assert out is not None
|
||
assert len(calls) == 2 # 历史摘要 + 轮前缀摘要（pi split-turn 双调用）
|
||
assert "PREFIX of a turn" in calls[1]
|
||
c = out[0].content
|
||
assert "SUMMARY-1" in c and "SUMMARY-2" in c
|
||
assert "---" in c
|
||
assert "**Turn Context (split turn):**" in c
|
||
# 保留尾巴 = 切点之后的 a3/a4
|
||
assert [m.content[:2] for m in out[1:]] == ["a3", "a4"]
|
||
|
||
|
||
# ======================================================================
|
||
# 12. pi 消息 → OpenAI 消息转换（tool 回传格式）
|
||
# ======================================================================
|
||
def test_to_openai_messages_format():
|
||
a = AgentMessage(role="assistant", content="", tool_calls=[
|
||
ToolCall(id="c1", name="bash", arguments={"command": "ls"})])
|
||
t = AgentMessage(role="toolResult", tool_call_id="c1", tool_name="bash",
|
||
content="ok")
|
||
out = to_openai_messages([
|
||
AgentMessage(role="user", content="hi"), a, t,
|
||
])
|
||
assert out[0] == {"role": "user", "content": "hi"}
|
||
assert out[1]["role"] == "assistant"
|
||
assert out[1]["tool_calls"][0]["function"]["name"] == "bash"
|
||
assert out[2]["role"] == "tool"
|
||
assert out[2]["tool_call_id"] == "c1"
|
||
assert out[2]["content"] == "ok"
|
||
|
||
|
||
# ======================================================================
|
||
# 12b. reasoning 回传（thinking 模式，DeepSeek 400 修复）
|
||
# ======================================================================
|
||
def test_reasoning_passthrough_toggle():
|
||
"""thinking 模式要求 assistant 的 reasoning_content 原样回传 → 默认开启；
|
||
显式 False 关闭（特殊网关逃生阀）。只影响带 reasoning 的 assistant 条目。"""
|
||
from core.agent.types import ModelConfig
|
||
a = AgentMessage(role="assistant", content="答", reasoning="思考过程...",
|
||
tool_calls=[ToolCall(id="c1", name="bash",
|
||
arguments={"command": "ls"})])
|
||
msgs = [AgentMessage(role="user", content="hi"), a,
|
||
AgentMessage(role="toolResult", tool_call_id="c1",
|
||
tool_name="bash", content="ok")]
|
||
# 默认：reasoning_content 回传
|
||
out_def = to_openai_messages(msgs)
|
||
assert out_def[1]["reasoning_content"] == "思考过程..."
|
||
assert out_def[1]["content"] == "答"
|
||
assert out_def[1]["tool_calls"][0]["function"]["name"] == "bash"
|
||
assert out_def[2]["role"] == "tool"
|
||
# 显式关闭（逃生阀）：不注入
|
||
out_off = to_openai_messages(msgs, pass_reasoning=False)
|
||
assert "reasoning_content" not in out_off[1]
|
||
# 无 reasoning 的 assistant 条目不产生空字段
|
||
b = AgentMessage(role="assistant", content="无思考", tool_calls=[])
|
||
out_mix = to_openai_messages([b], pass_reasoning=True)
|
||
assert "reasoning_content" not in out_mix[0]
|
||
# ModelConfig 默认开
|
||
mc = ModelConfig(name="t")
|
||
assert mc.pass_reasoning is True
|
||
|
||
|
||
# ======================================================================
|
||
# 12c. bash 秒级读秒 + 真超时（进程树杀，修 Windows 孤儿进程假超时）
|
||
# ======================================================================
|
||
def test_bash_timer_ticks_and_real_timeout():
|
||
"""Popen 秒级滴答：on_timer 每秒回调（气泡读秒）；
|
||
到期杀进程树 → 真中断（修复前：Windows 孤儿子进程跑满全时长才报超时）。"""
|
||
import time as _t
|
||
import core.agent.tools as T
|
||
from core.agent.types import AbortSignal
|
||
ticks = []
|
||
ctx = {"cwd": os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"shell": True, "on_timer": lambda el, to: ticks.append((el, to))}
|
||
# 1) 3s 命令 timeout=5 → 正常结束 + 收到 1、2 滴
|
||
r = T.tool_bash("tb1", {"command": r'python -c "import time; time.sleep(3)"',
|
||
"timeout": 5}, AbortSignal(), None, ctx)
|
||
assert not r.is_error, r.as_text()
|
||
assert "[exit 0]" in r.as_text(), r.as_text()
|
||
els = [x[0] for x in ticks]
|
||
assert 1 in els and 2 in els, f"读秒缺失: {ticks}"
|
||
assert all(x[1] == 5 for x in ticks), ticks
|
||
# 2) 5s 命令 timeout=1 → 真杀：总耗时 < 4s（修复前孤儿进程要跑满 5s）
|
||
ticks.clear()
|
||
t0 = _t.time()
|
||
r2 = T.tool_bash("tb2", {"command": r'python -c "import time; time.sleep(5)"',
|
||
"timeout": 1}, AbortSignal(), None, ctx)
|
||
real = _t.time() - t0
|
||
assert r2.is_error and "超时" in r2.as_text(), r2.as_text()
|
||
assert real < 4.0, f"进程树杀不生效，实际阻塞 {real:.1f}s"
|
||
assert [x[0] for x in ticks] == [1], f"超前应读 1 秒: {ticks}"
|
||
|
||
|
||
# ======================================================================
|
||
# 13. 同一 Agent 多次 prompt（会话记忆累积，对照 pi Agent 复用）
|
||
# ======================================================================
|
||
def test_agent_reuse_across_prompts():
|
||
agent = make_agent(
|
||
[{"content": "第一次。"}, {"content": "第二次。"}])
|
||
r1 = agent.prompt("问题1")
|
||
r2 = agent.prompt("问题2")
|
||
users = [m for m in agent.state.messages if m.role == "user"]
|
||
assert len(users) == 2
|
||
# 第二轮上下文包含第一轮全部消息
|
||
assert r1.stop_reason == "stop" and r2.stop_reason == "stop"
|
||
|
||
|
||
# ======================================================================
|
||
# 14. AgentWorker 接线回归（P0：必须注入 stream_fn，否则"未配置 stream_fn"）
|
||
# ======================================================================
|
||
def test_agent_worker_stream_fn_wiring():
|
||
"""AgentWorker.run() 必须把 openai_stream 注入 Agent（对照 pi 的
|
||
agentLoopConfig.streamFn 注入点）。用假流函数验证整条接线。"""
|
||
import core.llm_engine as le
|
||
from PyQt6.QtCore import QCoreApplication
|
||
|
||
QCoreApplication.instance() or QCoreApplication([])
|
||
|
||
chunks, errors = [], []
|
||
real_stream = le.openai_stream
|
||
|
||
def fake_stream(context, model, signal, max_tokens, tools=None):
|
||
yield ("event", _text("你好，"))
|
||
yield ("event", _text("世界！"))
|
||
yield ("final", AgentMessage(role="assistant", content="你好，世界！",
|
||
stop_reason="stop",
|
||
usage={"input": 5, "output": 5}))
|
||
|
||
le.openai_stream = fake_stream
|
||
try:
|
||
cfg = le._load_config()
|
||
providers = list(cfg.get("providers", {}).keys())
|
||
assert providers, "config.json 里没有 provider"
|
||
w = le.AgentWorker(providers[0], "any-model",
|
||
[{"role": "user", "content": "hi"}])
|
||
w.chunk_received.connect(lambda t: chunks.append(t))
|
||
w.error_occurred.connect(lambda e: errors.append(e))
|
||
w.run() # 同步跑（不走线程），验证接线
|
||
finally:
|
||
le.openai_stream = real_stream
|
||
|
||
assert not errors, f"出现错误（stream_fn 未注入？）: {errors}"
|
||
assert "".join(chunks) == "你好，世界！"
|
||
|
||
|
||
# ======================================================================
|
||
# 15. 系统提示词：请求头部注入 + 不入历史 + SYSTEM_PROMPT.md 加载
|
||
# ======================================================================
|
||
def test_system_prompt_injected_at_request_head():
|
||
seen = {}
|
||
|
||
def spy_stream(context, model, signal, max_tokens, tools=None):
|
||
seen["context"] = list(context)
|
||
yield ("event", _text("ok"))
|
||
yield ("final", AgentMessage(role="assistant", content="ok",
|
||
stop_reason="stop",
|
||
usage={"input": 1, "output": 1}))
|
||
|
||
model = ModelConfig(provider="mock", name="m", context_window=100000,
|
||
max_tokens=8192, api_key="x", base_url="http://mock")
|
||
cfg = AgentConfig(model=model, tools=[], system_prompt="SYS-TEST-123")
|
||
agent = Agent(cfg)
|
||
agent.set_stream_fn(spy_stream)
|
||
agent.prompt("hi")
|
||
|
||
ctx = seen["context"]
|
||
assert ctx[0].role == "system" and ctx[0].content == "SYS-TEST-123"
|
||
# 系统提示词不能混进会话历史（不占压缩/持久化）
|
||
assert all(m.role != "system" for m in agent.state.messages)
|
||
# to_openai_messages：头部 system 放行，中途 system 跳过
|
||
from core.agent.stream_fn import to_openai_messages
|
||
msgs = [ctx[0],
|
||
AgentMessage(role="user", content="u"),
|
||
AgentMessage(role="system", content="stray")]
|
||
out = to_openai_messages(msgs)
|
||
assert out[0] == {"role": "system", "content": "SYS-TEST-123"}
|
||
assert sum(1 for m in out if m["role"] == "system") == 1
|
||
|
||
|
||
def test_system_prompt_md_file_loaded():
|
||
from core.llm_engine import load_system_prompt
|
||
prompt = load_system_prompt()
|
||
assert "haocode" in prompt and len(prompt) > 200
|
||
|
||
|
||
# ======================================================================
|
||
# 16. tools 必须进 API 请求（回归：模型因此"文字演戏"工具调用）
|
||
# ======================================================================
|
||
def test_openai_stream_sends_tools():
|
||
import types as _pytypes
|
||
import core.agent.stream_fn as sf
|
||
from core.agent.stream_fn import AbortSignal
|
||
from core.agent.tools import default_tools
|
||
|
||
captured = {}
|
||
|
||
class _FakeStream:
|
||
def __init__(self, chunks):
|
||
self._it = iter(chunks)
|
||
|
||
def __iter__(self):
|
||
return self
|
||
|
||
def __next__(self):
|
||
return next(self._it)
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
def _chunk(choices, usage=None):
|
||
return _pytypes.SimpleNamespace(usage=usage, choices=choices)
|
||
|
||
def _delta(content=None, tool_calls=None):
|
||
return _pytypes.SimpleNamespace(
|
||
content=content, tool_calls=tool_calls,
|
||
**{"reasoning_content": None})
|
||
|
||
_tc = _pytypes.SimpleNamespace(
|
||
index=0, id="call_1",
|
||
function=_pytypes.SimpleNamespace(name="bash",
|
||
arguments='{"command":"echo hi"}'))
|
||
chunks = [
|
||
_chunk([_pytypes.SimpleNamespace(delta=_delta(tool_calls=[_tc]),
|
||
finish_reason="tool_calls")]),
|
||
_chunk([], usage=_pytypes.SimpleNamespace(prompt_tokens=3,
|
||
completion_tokens=2)),
|
||
]
|
||
|
||
class _FakeCompletions:
|
||
def create(self, **params):
|
||
captured.update(params)
|
||
return _FakeStream(chunks)
|
||
|
||
real_openai = sf.OpenAI
|
||
sf.OpenAI = lambda **kw: _pytypes.SimpleNamespace(
|
||
chat=_pytypes.SimpleNamespace(completions=_FakeCompletions()))
|
||
try:
|
||
model = ModelConfig(provider="t", name="m", context_window=100000,
|
||
max_tokens=100, api_key="k", base_url="http://x")
|
||
sig = AbortSignal()
|
||
out = list(sf.openai_stream([AgentMessage(role="user", content="hi")],
|
||
model, sig, 100, tools=default_tools()))
|
||
finally:
|
||
sf.OpenAI = real_openai
|
||
|
||
tools = captured.get("tools")
|
||
assert tools, "tools 没进请求体！"
|
||
names = {t["function"]["name"] for t in tools}
|
||
assert names == {"read", "bash", "write", "edit"}
|
||
assert tools[0]["type"] == "function"
|
||
|
||
final = [p for k, p in out if k == "final"][0]
|
||
assert final.tool_calls and final.tool_calls[0].name == "bash"
|
||
assert final.tool_calls[0].arguments == {"command": "echo hi"}
|
||
|
||
|
||
# ======================================================================
|
||
# 17. 循环层兜底：文字"演"的工具调用被解析执行
|
||
# ======================================================================
|
||
def test_text_tool_calls_loop_fallback():
|
||
turns = [
|
||
{"content": "我来看看。<bash>echo haocode-test</bash> 稍等",
|
||
"tool_calls": []},
|
||
{"content": "命令输出正常。"},
|
||
]
|
||
agent = make_agent(turns, tools=default_tools())
|
||
result = agent.prompt("看一下")
|
||
|
||
msgs = agent.state.messages
|
||
asst1 = next(m for m in msgs
|
||
if m.role == "assistant" and m.tool_calls)
|
||
assert asst1.tool_calls[0].name == "bash"
|
||
assert asst1.tool_calls[0].arguments["command"] == "echo haocode-test"
|
||
tr = next(m for m in msgs if m.role == "toolResult")
|
||
assert "haocode-test" in (tr.content if isinstance(tr.content, str)
|
||
else str(tr.content))
|
||
assert result.stop_reason == "stop"
|
||
|
||
|
||
# ======================================================================
|
||
# 18. from_openai_messages 全保真（worker 模式上下文丢失 bug 的回归守护）
|
||
# ======================================================================
|
||
def test_from_openai_messages_preserves_tool_history():
|
||
"""build_api_context 产物（含 tool 消息 + assistant tool_calls）经
|
||
from_openai_messages → to_openai_messages 往返后必须保真：
|
||
工具历史不能丢，残缺参数 JSON 必须兑底成合法 "{}"。"""
|
||
import json as _json
|
||
src = [
|
||
{"role": "user", "content": "帮我运行 main.py"},
|
||
{"role": "assistant", "content": "我来读一下。"},
|
||
{"role": "assistant", "content": None, "tool_calls": [
|
||
{"id": "c1", "type": "function",
|
||
"function": {"name": "read", "arguments": '{"path": "main.py"}'}}]},
|
||
{"role": "tool", "tool_call_id": "c1", "content": "print('hello')"},
|
||
{"role": "assistant", "content": None, "tool_calls": [
|
||
{"id": "c2", "type": "function",
|
||
"function": {"name": "bash",
|
||
"arguments": '{"command": "python main.py"}'}}]},
|
||
{"role": "tool", "tool_call_id": "c2", "content": "hello"},
|
||
{"role": "assistant", "content": "运行成功。"},
|
||
{"role": "user", "content": "刚才的代码是什么？"},
|
||
]
|
||
msgs = from_openai_messages(src)
|
||
# 角色序列保真（8 条，无丢失、无空 assistant）
|
||
assert [m.role for m in msgs] == [
|
||
"user", "assistant", "assistant", "toolResult",
|
||
"assistant", "toolResult", "assistant", "user"]
|
||
# tool_call 保真
|
||
assert msgs[2].tool_calls[0].id == "c1"
|
||
assert msgs[2].tool_calls[0].name == "read"
|
||
assert msgs[2].tool_calls[0].arguments == {"path": "main.py"}
|
||
assert msgs[4].tool_calls[0].name == "bash"
|
||
# toolResult 保真
|
||
assert msgs[3].tool_call_id == "c1"
|
||
assert msgs[3].content == "print('hello')"
|
||
# 纯工具轮不再是空 assistant
|
||
assert msgs[2].content == "" and len(msgs[2].tool_calls) == 1
|
||
|
||
# 往返：to_openai_messages(from_openai_messages(x)) 角色序一致 + JSON 合法
|
||
rt = to_openai_messages(msgs)
|
||
assert [m["role"] for m in rt] == [m["role"] for m in src]
|
||
for m in rt:
|
||
if m["role"] == "assistant" and m.get("tool_calls"):
|
||
for tc in m["tool_calls"]:
|
||
_json.loads(tc["function"]["arguments"]) # 必须是合法 JSON
|
||
|
||
# 残缺参数 JSON 兑底（中断导致的历史截断场景）
|
||
bad = [{"role": "user", "content": "hi"},
|
||
{"role": "assistant", "content": None, "tool_calls": [
|
||
{"id": "c9", "type": "function",
|
||
"function": {"name": "bash", "arguments": '{"command": "ls -'}}]},
|
||
{"role": "tool", "tool_call_id": "c9", "content": "out"}]
|
||
bmsgs = from_openai_messages(bad)
|
||
assert bmsgs[1].tool_calls[0].arguments == {}
|
||
assert bmsgs[1].tool_calls[0].raw_arguments == "{}"
|
||
brt = to_openai_messages(bmsgs)
|
||
_json.loads(brt[1]["tool_calls"][0]["function"]["arguments"])
|
||
|
||
# token 估算器能消化 toolResult/tool_calls（不崩 + 计入工具历史）
|
||
est_full = estimate_context_tokens(msgs).tokens
|
||
est_notools = estimate_context_tokens(
|
||
[m for m in msgs if m.role != "toolResult"]).tokens
|
||
assert est_full > est_notools
|
||
|
||
|
||
def test_agent_worker_second_turn_sees_tool_history():
|
||
"""集成：AgentWorker 第 2 轮真实发出的 context 必须包含第 1 轮的全部
|
||
工具历史（toolResult + assistant tool_calls）——任务完成后追问不丢上下文。"""
|
||
import core.llm_engine as le
|
||
from PyQt6.QtCore import QCoreApplication
|
||
|
||
QCoreApplication.instance() or QCoreApplication([])
|
||
seen = {}
|
||
real_stream = le.openai_stream
|
||
|
||
def fake_stream(context, model, signal, max_tokens, tools=None):
|
||
seen["context"] = list(context)
|
||
yield ("event", _text("ok"))
|
||
yield ("final", AgentMessage(role="assistant", content="ok",
|
||
stop_reason="stop",
|
||
usage={"input": 1, "output": 1}))
|
||
|
||
le.openai_stream = fake_stream
|
||
try:
|
||
cfg = le._load_config()
|
||
providers = list(cfg.get("providers", {}).keys())
|
||
assert providers, "config.json 里没有 provider"
|
||
w = le.AgentWorker(providers[0], "any-model", [
|
||
{"role": "user", "content": "帮我运行 main.py"},
|
||
{"role": "assistant", "content": "我来读一下。"},
|
||
{"role": "assistant", "content": None, "tool_calls": [
|
||
{"id": "c1", "type": "function",
|
||
"function": {"name": "read",
|
||
"arguments": '{"path": "main.py"}'}}]},
|
||
{"role": "tool", "tool_call_id": "c1",
|
||
"content": "print('hello')"},
|
||
{"role": "assistant", "content": "运行成功。"},
|
||
{"role": "user", "content": "刚才的代码是什么？"},
|
||
])
|
||
errors = []
|
||
w.error_occurred.connect(lambda e: errors.append(e))
|
||
w.run() # 同步跑
|
||
finally:
|
||
le.openai_stream = real_stream
|
||
|
||
assert not errors, f"AgentWorker 报错: {errors}"
|
||
ctx = seen["context"]
|
||
# 工具历史必须完整在场
|
||
trs = [m for m in ctx if m.role == "toolResult"]
|
||
assert len(trs) == 1 and trs[0].tool_call_id == "c1"
|
||
assert "print('hello')" in trs[0].content
|
||
asst_tc = [m for m in ctx if m.role == "assistant" and m.tool_calls]
|
||
assert len(asst_tc) == 1 and asst_tc[0].tool_calls[0].name == "read"
|
||
# 末尾是本次提问
|
||
assert ctx[-1].role == "user" and ctx[-1].content == "刚才的代码是什么？"
|
||
|
||
|
||
# ======================================================================
|
||
# 🆕 P0: usage 锚点时效校验（对照 pi getLastAssistantUsageInfo）
|
||
# ======================================================================
|
||
def test_usage_anchor_stale_after_compaction_summary():
|
||
"""压缩后：保留窗里 assistant 的旧 usage（来自压缩前）必须失效，
|
||
否则下一次 should_compact 会锚定到过期 60K → 误触发二次压缩。"""
|
||
stale = AgentMessage(role="assistant", content="旧回答",
|
||
timestamp=2000, stop_reason="stop",
|
||
usage={"input": 55000, "output": 5000})
|
||
# 对照组：无摘要 → 应锚定到 stale usage
|
||
est_ctrl = estimate_context_tokens([
|
||
AgentMessage(role="user", content="旧问题", timestamp=1000),
|
||
stale,
|
||
])
|
||
assert est_ctrl.last_usage_index == 1
|
||
assert est_ctrl.usage_tokens == 60000
|
||
|
||
# 实验组：新摘要消息（timestamp 更晚）插到前面 → 旧 usage 失效
|
||
summary = AgentMessage(role="user", content="（压缩摘要）",
|
||
timestamp=9000, kind="compaction_summary")
|
||
est = estimate_context_tokens([summary,
|
||
AgentMessage(role="user", content="旧问题", timestamp=1000),
|
||
stale])
|
||
assert est.last_usage_index is None, f"过期 usage 未被失效: {est}"
|
||
assert est.usage_tokens == 0
|
||
|
||
|
||
def test_usage_anchor_recovers_after_new_assistant():
|
||
"""压缩后第一个新 assistant 回复到达 → 新 usage 自动成为锚点"""
|
||
msgs = [
|
||
AgentMessage(role="user", content="（压缩摘要）", timestamp=9000,
|
||
kind="compaction_summary"),
|
||
AgentMessage(role="assistant", content="旧回答", timestamp=2000,
|
||
stop_reason="stop", usage={"input": 55000, "output": 5000}),
|
||
AgentMessage(role="user", content="新问题", timestamp=10000),
|
||
AgentMessage(role="assistant", content="新回答", timestamp=11000,
|
||
stop_reason="stop", usage={"input": 20000, "output": 300}),
|
||
]
|
||
est = estimate_context_tokens(msgs)
|
||
assert est.last_usage_index == 3, f"应锚定到新回复: {est}"
|
||
assert est.usage_tokens == 20300
|
||
|
||
|
||
def test_estimate_no_anchor_without_timestamps_backward_compatible():
|
||
"""无 timestamp 信息（全 0，旧 DB 回放）→ 行为等同旧版「取最后一条有效」"""
|
||
msgs = [
|
||
AgentMessage(role="user", content="q1"),
|
||
AgentMessage(role="assistant", content="a1", stop_reason="stop",
|
||
usage={"input": 100, "output": 10}),
|
||
AgentMessage(role="user", content="q2"),
|
||
AgentMessage(role="assistant", content="a2", stop_reason="stop",
|
||
usage={"input": 200, "output": 20}),
|
||
]
|
||
est = estimate_context_tokens(msgs)
|
||
assert est.last_usage_index == 3
|
||
assert est.usage_tokens == 220
|
||
|
||
|
||
# ======================================================================
|
||
# 🆕 P2: 无锚点时 system 提示词 + 工具 schema 计入估算/钳制
|
||
# ======================================================================
|
||
class _FakeTool:
|
||
name = "read"
|
||
description = "读取文件内容"
|
||
parameters = {"type": "object", "properties": {"path": {"type": "string"}}}
|
||
|
||
|
||
def test_estimate_includes_system_and_tools_without_anchor():
|
||
sysp = "你是一个编程助手。" * 100 # 900 字（。U+3002 不在 CJK 类 → 估算 ≈825）
|
||
base_msgs = [AgentMessage(role="user", content="hi")]
|
||
plain = estimate_context_tokens(list(base_msgs))
|
||
full = estimate_context_tokens(list(base_msgs),
|
||
system_prompt=sysp, tools=[_FakeTool()])
|
||
# system(≈825) + 工具 schema(>30) 必须被计入
|
||
assert full.tokens >= plain.tokens + 800, f"{plain} -> {full}"
|
||
|
||
# 有锚点时：usage 已精确覆盖 system+tools，不重复计
|
||
anchor_msgs = [AgentMessage(role="assistant", content="x",
|
||
stop_reason="stop",
|
||
usage={"input": 1000, "output": 100})]
|
||
a1 = estimate_context_tokens(list(anchor_msgs))
|
||
a2 = estimate_context_tokens(list(anchor_msgs),
|
||
system_prompt=sysp, tools=[_FakeTool()])
|
||
assert a1.tokens == a2.tokens == 1100, f"{a1} vs {a2}"
|
||
|
||
|
||
def test_clamp_counts_system_tools():
|
||
# 小窗口（8K）：钳制结果受「窗口-输入-4096」约束，system+tools 计入后必须缩小
|
||
model = ModelConfig(provider="p", name="m",
|
||
context_window=8000, max_tokens=8192)
|
||
msgs = [AgentMessage(role="user", content="hi")]
|
||
c1 = clamp_max_tokens_to_context(model, list(msgs))
|
||
c2 = clamp_max_tokens_to_context(model, list(msgs),
|
||
system_prompt="x" * 4000,
|
||
tools=[_FakeTool()])
|
||
# 输入估算上升 → 输出预算如实缩小（对照 pi 传完整 Context）
|
||
assert c2[1] > c1[1] + 900, f"input: {c1[1]} -> {c2[1]}"
|
||
assert c2[0] < c1[0], f"max_tokens: {c1[0]} -> {c2[0]}"
|
||
|
||
|
||
# ======================================================================
|
||
# 🆕 M1/M2: 流式字段提取（reasoning 三字段 / choice.usage 兜底）
|
||
# ======================================================================
|
||
def test_pick_reasoning_field_priority():
|
||
from types import SimpleNamespace as NS
|
||
from core.agent.stream_fn import _pick_reasoning
|
||
# reasoning_content 优先（vLLM/Qwen）
|
||
assert _pick_reasoning(NS(reasoning="a", reasoning_content="b",
|
||
reasoning_text="c")) == "b"
|
||
# 其次 reasoning（llama.cpp）
|
||
assert _pick_reasoning(NS(reasoning="a")) == "a"
|
||
# 再次 reasoning_text
|
||
assert _pick_reasoning(NS(reasoning_text="c")) == "c"
|
||
# 全空
|
||
assert _pick_reasoning(NS()) == ""
|
||
assert _pick_reasoning(None) == ""
|
||
# pydantic 对象的 model_dump 兜底
|
||
class _D:
|
||
def model_dump(self):
|
||
return {"reasoning_text": "z"}
|
||
assert _pick_reasoning(_D()) == "z"
|
||
|
||
|
||
def test_pick_usage_choice_fallback():
|
||
from types import SimpleNamespace as NS
|
||
from core.agent.stream_fn import _pick_usage
|
||
u1 = object()
|
||
assert _pick_usage(NS(usage=u1)) is u1
|
||
# chunk.usage 为空 → choice.usage 兜底（Moonshot 系）
|
||
u2 = object()
|
||
assert _pick_usage(NS(usage=None, choices=[NS(usage=u2)])) is u2
|
||
assert _pick_usage(NS(usage=None, choices=[])) is None
|
||
assert _pick_usage(NS(usage=None)) is None
|
||
|
||
|
||
# ======================================================================
|
||
# 🆕 M3: AgentRunner 重试回调（onRetryScheduled / onRetryFinished）
|
||
# ======================================================================
|
||
def test_agent_runner_retry_callbacks():
|
||
from core.agent.recovery import AgentRunner
|
||
from core.agent.types import AgentConfig, ModelConfig, RetryConfig, RunResult
|
||
|
||
events = []
|
||
|
||
class _State:
|
||
def __init__(self):
|
||
self.messages = [
|
||
AgentMessage(role="user", content="q"),
|
||
AgentMessage(role="assistant", stop_reason="error",
|
||
error_message="connection error"),
|
||
]
|
||
|
||
class _Agent:
|
||
def __init__(self):
|
||
self.state = _State()
|
||
self.config = AgentConfig(
|
||
model=ModelConfig(provider="p", name="m",
|
||
context_window=128000, max_tokens=8192),
|
||
retry=RetryConfig(max_attempts=3, base_delay_ms=1))
|
||
self.continued = False
|
||
|
||
def has_queued(self):
|
||
return False
|
||
|
||
def continue_(self):
|
||
self.continued = True
|
||
self.state.messages[-1] = AgentMessage(
|
||
role="assistant", stop_reason="stop", content="ok")
|
||
return RunResult(stop_reason="stop")
|
||
|
||
agent = _Agent()
|
||
runner = AgentRunner(
|
||
agent, summarize_fn=None,
|
||
on_retry_scheduled=lambda a, m, d, r: events.append(("sched", a, m, r)),
|
||
on_retry_finished=lambda ok: events.append(("fin", ok)))
|
||
|
||
runner._post_loop(RunResult(stop_reason="stop"))
|
||
|
||
assert agent.continued, "可重试错误未触发 continue"
|
||
kinds = [e[0] for e in events]
|
||
assert "sched" in kinds and "fin" in kinds, f"events={events}"
|
||
sched = next(e for e in events if e[0] == "sched")
|
||
assert sched[1] == 1 and sched[2] == 3
|
||
assert sched[3] == "connection error", f"reason 未透传: {sched}"
|
||
assert events[-1] == ("fin", True)
|
||
|
||
|
||
# ======================================================================
|
||
# 🆕 P1: usage/timestamp 入库回放链路
|
||
# ======================================================================
|
||
def test_from_openai_maps_usage_and_timestamp():
|
||
msgs = from_openai_messages([
|
||
{"role": "user", "content": "q", "timestamp": 111},
|
||
{"role": "assistant", "content": "a", "timestamp": 222,
|
||
"usage": {"input": 100, "output": 50}},
|
||
{"role": "tool", "tool_call_id": "c1", "content": "r",
|
||
"timestamp": 333},
|
||
])
|
||
assert msgs[0].timestamp == 111
|
||
assert msgs[1].timestamp == 222
|
||
assert msgs[1].usage == {"input": 100, "output": 50}
|
||
assert msgs[2].timestamp == 333
|
||
est = estimate_context_tokens(msgs)
|
||
assert est.last_usage_index == 1 and est.usage_tokens == 150
|
||
|
||
|
||
def test_db_usage_roundtrip():
|
||
"""add_message(usage=...) → get_message_chain 行带 usage → 锚定生效"""
|
||
import json as _json
|
||
import tempfile
|
||
import time as _time
|
||
from core.db_manager import DBManager
|
||
|
||
tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
||
tmp.close()
|
||
try:
|
||
db = DBManager(tmp.name)
|
||
sid = db.create_session("P1 usage 测试")["id"]
|
||
uid = db.add_message(sid, "user", "问题一", None)
|
||
_time.sleep(1.1) # 保证 created_at（秒）递增，模拟真实时序
|
||
db.add_message(sid, "assistant", "回答一（很短）", uid["id"],
|
||
usage=_json.dumps({"input": 30000, "output": 2000}))
|
||
chain = db.get_message_chain(sid)
|
||
assert [m["role"] for m in chain] == ["user", "assistant"], \
|
||
f"chain={[(m['role'], m['parent_id']) for m in chain]}"
|
||
assert _json.loads(chain[1]["usage"]) == {"input": 30000, "output": 2000}
|
||
|
||
# 模拟 build_api_context 的回放（timestamp/usage 挂到 OpenAI dict）
|
||
oai = [{"role": m["role"], "content": m["content"],
|
||
"timestamp": int(m["created_at"]) * 1000,
|
||
**({"usage": _json.loads(m["usage"])}
|
||
if m.get("usage") else {})}
|
||
for m in chain]
|
||
agent_msgs = from_openai_messages(oai)
|
||
est = estimate_context_tokens(agent_msgs)
|
||
assert est.last_usage_index == 1, f"usage 未锚定: {est}"
|
||
assert est.usage_tokens == 32000
|
||
finally:
|
||
try:
|
||
os.remove(tmp.name)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
# ======================================================================
|
||
# 轮中主动压缩检查（🆕 haocode 增强，偏离 pi 1:1）
|
||
# 内层循环每次 LLM 请求前跑 should_compact 同公式，超阈值先压缩再发。
|
||
# ======================================================================
|
||
|
||
def _mid_tool(name, text):
|
||
"""确定性工具：返回固定文本（避免真实 bash）"""
|
||
def execute(tool_call_id, args, signal, on_update, context):
|
||
return AgentToolResult.text(text)
|
||
return AgentTool(
|
||
name=name, description=name,
|
||
parameters={"type": "object", "properties": {}},
|
||
execute=execute, label=name)
|
||
|
||
|
||
def _prefill_history(agent, n=8, chars=2000):
|
||
for i in range(n):
|
||
agent.state.messages.append(
|
||
AgentMessage(role="user", content=f"hist{i} " + "长" * chars))
|
||
|
||
|
||
def _capture_requests(agent, captured):
|
||
"""包一层 stream_fn，记录每次请求的 api 上下文"""
|
||
orig = agent._stream_fn
|
||
|
||
def wrapper(context, model, signal, max_tokens, tools=None):
|
||
captured.append(list(context))
|
||
yield from orig(context, model, signal, max_tokens, tools)
|
||
|
||
agent._stream_fn = wrapper
|
||
|
||
|
||
def test_mid_turn_compaction_fires_on_growth():
|
||
# 窗口 40000 → 阈值 23616；预填 8×2000 字 ≈ 16k（轮首检查不触发）。
|
||
# ⚠️ 脚本流的 usage 必须真实（P0 锚点机制：估算锚定到最后一条
|
||
# assistant 的真实 usage.input + 其后新增消息）。turn1 请求时真实
|
||
# 输入 ≈ 16053 → usage.input=16000。
|
||
agent = make_agent(
|
||
[{"tool_calls": [{"id": "c1", "name": "midtool",
|
||
"arguments": {}}], "stop_reason": "stop",
|
||
"usage": {"input": 16000, "output": 10}},
|
||
{"content": "done.",
|
||
"usage": {"input": 20000, "output": 10}}],
|
||
tools=[_mid_tool("midtool", "长" * 8000)], # 8k → 锚定后 ≈ 24k > 23616
|
||
context_window=40000,
|
||
)
|
||
_prefill_history(agent)
|
||
summarize_calls = []
|
||
|
||
def summarize(prompt, system_prompt, max_tokens):
|
||
summarize_calls.append(1)
|
||
return "SUMMARY-MID"
|
||
|
||
runner = AgentRunner(agent, summarize_fn=summarize)
|
||
agent.config.compact_fn = runner.compact_if_needed # 模拟 llm_engine 接线
|
||
captured = []
|
||
_capture_requests(agent, captured)
|
||
|
||
result = runner.run("go")
|
||
assert result.stop_reason == "stop"
|
||
# 轮中压缩恰好触发一次（轮首 16k 不触发；压缩后 ≈20k 不再触发）
|
||
assert len(summarize_calls) == 1
|
||
assert runner.compactions_performed == 1
|
||
assert any(ev.get("path") == "mid_turn" for ev in runner.compaction_events)
|
||
# 状态首条是摘要，最早两条历史被摘要掉
|
||
assert agent.state.messages[0].kind == "compaction_summary"
|
||
assert "SUMMARY-MID" in agent.state.messages[0].content
|
||
all_text = "".join(str(getattr(m, "content", "")) for m in agent.state.messages)
|
||
assert "hist0" not in all_text and "hist1" not in all_text
|
||
assert "hist7" in all_text
|
||
# 第二次请求（压缩后的那个）上下文首条就是摘要
|
||
assert len(captured) == 2
|
||
assert captured[1][0].kind == "compaction_summary"
|
||
|
||
|
||
def test_mid_turn_compaction_failure_capped():
|
||
# 单条巨型工具结果（30k ≥ keep_recent 20k）在尾部 → 无有效切点
|
||
# （cut_pinned_at_zero）→ 压缩失败；连败 2 次后本轮不再尝试（防刷屏护栏）。
|
||
# 窗口 100000 → 阈值 83616；预填 34×2000 ≈ 68k。usage 必须真实
|
||
# （锚点估算 = 最后 asst 的 usage.input + 其后新增）。
|
||
import core.agent.recovery as rec
|
||
calls = []
|
||
orig = rec.compact_context
|
||
|
||
def counting(*a, **kw):
|
||
calls.append(1)
|
||
return orig(*a, **kw)
|
||
|
||
rec.compact_context = counting
|
||
try:
|
||
agent = make_agent(
|
||
[{"tool_calls": [{"id": "c1", "name": "midtool",
|
||
"arguments": {}}], "stop_reason": "stop",
|
||
"usage": {"input": 60000, "output": 10}},
|
||
{"tool_calls": [{"id": "c2", "name": "midtool",
|
||
"arguments": {}}], "stop_reason": "stop",
|
||
"usage": {"input": 65000, "output": 10}},
|
||
{"content": "done.",
|
||
"usage": {"input": 50000, "output": 10}}],
|
||
tools=[_mid_tool("midtool", "长" * 30000)],
|
||
context_window=100000,
|
||
)
|
||
for i in range(34):
|
||
agent.state.messages.append(
|
||
AgentMessage(role="user", content=f"hist{i} " + "长" * 2000))
|
||
|
||
def summarize(prompt, system_prompt, max_tokens):
|
||
raise AssertionError("prepare 返回 None → 不应调用 LLM 摘要")
|
||
|
||
runner = AgentRunner(agent, summarize_fn=summarize)
|
||
agent.config.compact_fn = runner.compact_if_needed
|
||
result = runner.run("go")
|
||
assert result.stop_reason == "stop"
|
||
# 3 次请求机会：第 2、3 次前尝试压缩（均失败），第… 耗尽后跳过
|
||
# turn2 前: 60000+30004=90004 > 83616 → 尝试#1 失败
|
||
# turn3 前: 65000+30004=95004 > 83616 → 尝试#2 失败 → 耗尽
|
||
assert len(calls) == 2, f"compact_context 应恰好尝试 2 次，实际 {len(calls)}"
|
||
assert runner._mid_turn_exhausted is True
|
||
assert runner.compactions_performed == 0
|
||
# 状态未被修改（无摘要），原样继续
|
||
assert not any(getattr(m, "kind", None) == "compaction_summary"
|
||
for m in agent.state.messages)
|
||
finally:
|
||
rec.compact_context = orig
|
||
|
||
|
||
def test_mid_turn_no_compaction_below_threshold():
|
||
# 工具结果小（2k）→ 锚定后 16000+2004 ≈ 18k < 23616 → 不压缩，正常完成
|
||
agent = make_agent(
|
||
[{"tool_calls": [{"id": "c1", "name": "midtool",
|
||
"arguments": {}}], "stop_reason": "stop",
|
||
"usage": {"input": 16000, "output": 10}},
|
||
{"content": "done.",
|
||
"usage": {"input": 18000, "output": 10}}],
|
||
tools=[_mid_tool("midtool", "长" * 2000)],
|
||
context_window=40000,
|
||
)
|
||
_prefill_history(agent)
|
||
summarize_calls = []
|
||
|
||
def summarize(prompt, system_prompt, max_tokens):
|
||
summarize_calls.append(1)
|
||
return "SHOULD-NOT-RUN"
|
||
|
||
runner = AgentRunner(agent, summarize_fn=summarize)
|
||
agent.config.compact_fn = runner.compact_if_needed
|
||
result = runner.run("go")
|
||
assert result.stop_reason == "stop"
|
||
assert summarize_calls == []
|
||
assert runner.compactions_performed == 0
|
||
assert not any(getattr(m, "kind", None) == "compaction_summary"
|
||
for m in agent.state.messages)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(pytest.main([__file__, "-v"]))
|