quercus/tests/core/test_loop.py

"""Agent loop：两轮对话、确认门、只读直通、视觉闭环回喂、会话日志。"""
from quercus_core.agent.loop import AgentLoop
from quercus_core.providers.base import AssistantTurn, ToolCall
from quercus_core.session.log import SessionLog, replay
from quercus_core.types import PngBytes, ToolResult

from mock_llm import MockProvider

PNG = PngBytes(b"\x89PNG\r\n\x1a\n" + bytes(range(64)))


def _sha256(data: bytes) -> str:
    import hashlib

    return hashlib.sha256(data).hexdigest()


class StubExecutor:
    """内存 stub 工具执行器（P1 无真实宿主实现）。"""

    def __init__(self):
        self.calls = []

    def execute(self, action):
        self.calls.append(action)
        if action.tool == "get_frame":
            return ToolResult(tool="get_frame", ok=True, summary="帧：00:00:02", images=(PNG,))
        if action.tool == "place_clip":
            return ToolResult(tool="place_clip", ok=True, summary="已放置片段 c1 到轨道 1")
        if action.tool == "ripple_delete":
            return ToolResult(tool="ripple_delete", ok=True, summary="已波纹删除 1 个片段")
        return ToolResult(tool=action.tool, ok=True, summary=f"{action.tool} ok")


def _frame_call(i: int) -> ToolCall:
    return ToolCall(
        id=f"call_frame_{i}",
        name="get_frame",
        arguments={"time": {"num": i, "den": 1}, "max_size": {"width": 32, "height": 18}},
    )


def test_two_round_loop_confirm_auto_approve():
    """第一轮含变更工具(触发确认门) + get_frame(PNG 回喂)；第二轮纯文本结束。"""
    mock = MockProvider(
        script=[
            AssistantTurn(
                text="",
                tool_calls=(
                    ToolCall(
                        id="call_place",
                        name="place_clip",
                        arguments={"clip_id": "c1", "track_index": 1, "time": {"num": 0, "den": 1}},
                    ),
                    _frame_call(2),
                ),
            ),
            AssistantTurn(text="剪辑完成", tool_calls=()),
        ]
    )
    stub = StubExecutor()
    confirmed = []

    def confirm(batch):
        confirmed.append(batch)
        return True

    loop = AgentLoop(provider=mock, executor=stub, confirm=confirm)
    result = loop.run("铺一个片段并看一眼画面")

    assert result.final_text == "剪辑完成"
    assert result.turns == 2
    assert result.tool_calls_executed == 2
    assert [a.tool for a in stub.calls] == ["place_clip", "get_frame"]
    assert len(confirmed) == 1
    assert len(confirmed[0].actions) == 2

    # 视觉闭环：get_frame 的 PNG 作为 tool 消息回喂 provider
    last_messages = mock.observed[-1]
    tool_msgs = [m for m in last_messages if m.role == "tool"]
    assert len(tool_msgs) == 2
    frame_msg = next(m for m in tool_msgs if m.name == "get_frame")
    assert PNG in frame_msg.images


def test_rejected_batch_mutation_not_executed():
    mock = MockProvider(
        script=[
            AssistantTurn(
                text="",
                tool_calls=(
                    ToolCall(id="c1", name="ripple_delete", arguments={"clip_ids": ["x1"]}),
                ),
            ),
            AssistantTurn(text="已取消删除", tool_calls=()),
        ]
    )
    stub = StubExecutor()
    loop = AgentLoop(provider=mock, executor=stub, confirm=lambda batch: False)
    result = loop.run("删掉 x1")

    assert stub.calls == []  # 变更工具未执行
    assert result.final_text == "已取消删除"
    assert loop.batch_decisions[0][1] is False
    # 拒绝结果回喂 LLM
    last_messages = mock.observed[-1]
    tool_msgs = [m for m in last_messages if m.role == "tool"]
    assert len(tool_msgs) == 1
    assert "未执行" in tool_msgs[0].text


def test_readonly_batch_bypasses_confirm():
    mock = MockProvider(
        script=[
            AssistantTurn(text="", tool_calls=(_frame_call(1),)),
            AssistantTurn(text="帧已取到", tool_calls=()),
        ]
    )
    stub = StubExecutor()
    confirm_calls = []

    def confirm(batch):
        confirm_calls.append(batch)
        return True

    loop = AgentLoop(provider=mock, executor=stub, confirm=confirm)
    result = loop.run("取一帧")

    assert confirm_calls == []  # 纯只读批次不触发确认门
    assert [a.tool for a in stub.calls] == ["get_frame"]
    assert result.final_text == "帧已取到"


def test_default_confirm_is_auto_approve():
    """未注入 confirm 回调时默认全开（计划文档 P1.3）。"""
    mock = MockProvider(
        script=[
            AssistantTurn(
                text="",
                tool_calls=(ToolCall(id="c1", name="add_marker", arguments={"time": {"num": 1, "den": 1}}),),
            ),
            AssistantTurn(text="已打点", tool_calls=()),
        ]
    )
    stub = StubExecutor()
    loop = AgentLoop(provider=mock, executor=stub)
    result = loop.run("打点")
    assert [a.tool for a in stub.calls] == ["add_marker"]
    assert result.final_text == "已打点"


def test_max_turns_bound():
    mock = MockProvider(
        script=[AssistantTurn(text="", tool_calls=(_frame_call(i),)) for i in range(3)]
    )
    stub = StubExecutor()
    loop = AgentLoop(provider=mock, executor=stub, max_turns=2)
    result = loop.run("一直取帧")
    assert result.turns == 2
    assert len(stub.calls) == 2


def test_unknown_tool_reported_back():
    mock = MockProvider(
        script=[
            AssistantTurn(text="", tool_calls=(ToolCall(id="c1", name="no_such_tool", arguments={}),)),
            AssistantTurn(text="done", tool_calls=()),
        ]
    )
    stub = StubExecutor()
    loop = AgentLoop(provider=mock, executor=stub)
    loop.run("调用未知工具")
    assert stub.calls == []
    last = mock.observed[-1][-1]
    assert "未知工具" in last.text


def test_loop_writes_session_log(tmp_path):
    mock = MockProvider(
        script=[
            AssistantTurn(text="", tool_calls=(_frame_call(3),)),
            AssistantTurn(text="结束", tool_calls=()),
        ]
    )
    stub = StubExecutor()
    log = SessionLog(tmp_path / "sessions" / "s1.jsonl", session_id="sess-1")
    loop = AgentLoop(provider=mock, executor=stub, session=log)
    loop.run("取帧并结束")

    events = replay(log.path)
    kinds = [e.kind for e in events]
    assert kinds == ["user_message", "assistant_turn", "batch_decision", "tool_result", "assistant_turn"]
    # 帧哈希落盘（不落图片本体）
    result_event = next(e for e in events if e.kind == "tool_result")
    assert result_event.payload["image_hashes"] == [_sha256(PNG)]