"""
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__), ".."))

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"]))