script-build-and-framework-tmp/agent/tests/test_runner_components.py

import pytest

from agent import AgentRunner as PublicAgentRunner
from agent import RunConfig as PublicRunConfig
from agent.core.runner import AgentRunner, RunConfig
from agent.tools.registry import ToolRegistry
from agent.trace.models import Trace


def test_runner_public_imports_remain_stable():
    assert PublicAgentRunner is AgentRunner
    assert PublicRunConfig is RunConfig


@pytest.mark.asyncio
async def test_prompt_runtime_preserves_host_prompt_and_iteration_constraint(tmp_path):
    runner = AgentRunner(skills_dir=str(tmp_path))

    prompt = await runner._build_system_prompt(
        RunConfig(max_iterations=7),
        base_prompt="host contract",
    )

    assert prompt.startswith("host contract\n\n## Execution Constraint")
    assert "最多可以用 7 轮交互" in prompt


@pytest.mark.asyncio
async def test_task_name_runtime_uses_current_utility_model_callback():
    runner = AgentRunner()

    async def utility_llm_call(**_kwargs):
        return {"content": "stable title"}

    # Preserve the historical ability to configure this dependency after
    # construction; the internal component must not capture a stale value.
    runner.utility_llm_call = utility_llm_call

    assert (
        await runner._generate_task_name([{"role": "user", "content": "task"}])
        == "stable title"
    )


@pytest.mark.asyncio
async def test_task_name_runtime_keeps_text_fallback_on_utility_failure():
    async def failing_utility(**_kwargs):
        raise RuntimeError("unavailable")

    runner = AgentRunner(utility_llm_call=failing_utility)
    text = "x" * 60

    assert await runner._generate_task_name([{"role": "user", "content": text}]) == (
        "x" * 50 + "..."
    )


@pytest.mark.asyncio
async def test_tool_runtime_keeps_runner_override_hooks() -> None:
    registry = ToolRegistry()

    async def sample() -> str:
        return "ok"

    registry.register(
        sample,
        schema={
            "type": "function",
            "function": {
                "name": "sample",
                "description": "sample",
                "parameters": {"type": "object", "properties": {}},
            },
        },
    )

    class ExtensibleRunner(AgentRunner):
        policy_calls = 0
        context_calls = 0
        schema_calls = 0

        def _resolve_run_policy(self, config, trace=None):
            self.policy_calls += 1
            return super()._resolve_run_policy(config, trace)

        def _build_protected_tool_context(
            self,
            config,
            trace,
            goal_tree,
            sequence,
            side_branch=None,
        ):
            self.context_calls += 1
            return super()._build_protected_tool_context(
                config,
                trace,
                goal_tree,
                sequence,
                side_branch=side_branch,
            )

        def _get_tool_schemas(
            self,
            tools=None,
            tool_groups=None,
            exclude_tools=None,
        ):
            self.schema_calls += 1
            return super()._get_tool_schemas(
                tools,
                tool_groups,
                exclude_tools,
            )

    runner = ExtensibleRunner(tool_registry=registry)
    runner._get_run_tool_schemas(RunConfig(tools=["sample"], tool_groups=[]))
    result = await runner._execute_authorized_tool(
        "sample",
        {},
        "call-1",
        RunConfig(tools=["sample"], tool_groups=[]),
        Trace(trace_id="legacy", mode="agent", agent_role="legacy"),
        None,
        1,
    )

    assert result == "ok"
    assert runner.policy_calls == 2
    assert runner.context_calls == 1
    assert runner.schema_calls == 1


@pytest.mark.asyncio
async def test_image_runtime_keeps_runner_override_hooks() -> None:
    class ExtensibleRunner(AgentRunner):
        image_calls = 0

        async def _process_image_size(
            self,
            base64_url,
            max_size=512,
            min_size=11,
        ):
            self.image_calls += 1
            return "data:image/jpeg;base64,processed"

    runner = ExtensibleRunner()
    messages = [
        {
            "role": "tool",
            "content": [
                {
                    "type": "image_url",
                    "image_url": {"url": "data:image/png;base64,source"},
                }
            ],
        },
        {"role": "assistant", "content": "observed"},
    ]

    optimized = await runner._optimize_images(messages, "gpt-4o")

    assert runner.image_calls == 1
    assert optimized[0]["content"][0]["image_url"]["url"].endswith("processed")
    assert messages[0]["content"][0]["image_url"]["url"].endswith("source")