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

from __future__ import annotations

import json
import logging

import pytest

from agent import (
    AgentRunner,
    FailureDetail,
    FailureDisposition,
    RunConfig,
    ToolExecutionError,
    ToolResult,
)
from agent.trace.store import FileSystemTraceStore
from agent.tools.registry import ToolRegistry


def _schema(name: str) -> dict:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": name,
            "parameters": {"type": "object", "properties": {}},
        },
    }


def test_failure_detail_is_bounded_json_safe_and_round_trips() -> None:
    failure = FailureDetail(
        code="INPUT_SCOPE_MISMATCH",
        message="x" * 3_000,
        disposition=FailureDisposition.REPLAN_TASK,
        source_tool="save_candidate",
        details={"task_id": "task-1", "scope": ["paragraph", 3]},
    )

    assert len(failure.message) == 2_000
    assert FailureDetail.from_dict(failure.to_dict()) == failure
    assert failure.fingerprint() == FailureDetail(
        code="INPUT_SCOPE_MISMATCH",
        message="different wording",
        disposition=FailureDisposition.REPLAN_TASK,
        source_tool="save_candidate",
        details={"task_id": "task-2", "scope": ["paragraph", 3]},
    ).fingerprint()


def test_failure_detail_rejects_non_json_details() -> None:
    with pytest.raises(ValueError, match="JSON values"):
        FailureDetail(
            code="INVALID",
            message="invalid",
            disposition=FailureDisposition.RETRY_CALL,
            details={"value": object()},
        )


def test_tool_result_exposes_structured_failure_to_model() -> None:
    failure = FailureDetail(
        code="REJECTED",
        message="retry with another value",
        disposition=FailureDisposition.RETRY_CALL,
        source_tool="sample",
    )
    result = ToolResult(title="rejected", output="context", failure=failure)

    payload = json.loads(result.to_llm_message())
    assert payload["failure"] == failure.to_dict()
    assert payload["output"] == "context"
    assert ToolExecutionError(failure).failure == failure


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

    async def sample() -> str:
        raise ToolExecutionError(
            FailureDetail(
                code="EXPECTED_REJECTION",
                message="change the contract",
                disposition=FailureDisposition.REPLAN_TASK,
            )
        )

    registry.register(sample, schema=_schema("sample"))
    with caplog.at_level(logging.WARNING):
        result = await registry.execute("sample", {})

    assert result["_control"]["failure"]["code"] == "EXPECTED_REJECTION"
    assert result["_control"]["failure"]["source_tool"] == "sample"
    assert not any(record.exc_info for record in caplog.records)
    assert registry.get_stats("sample")["sample"]["failure_count"] == 1


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

    async def sample() -> str:
        raise RuntimeError("secret implementation detail")

    registry.register(sample, schema=_schema("sample"))
    with caplog.at_level(logging.ERROR):
        result = await registry.execute("sample", {})

    failure = result["_control"]["failure"]
    assert failure["code"] == "UNEXPECTED_TOOL_ERROR"
    assert "secret implementation detail" not in result["text"]
    assert any(record.exc_info for record in caplog.records)


@pytest.mark.asyncio
@pytest.mark.parametrize("parallel", [False, True])
async def test_runner_hides_unexpected_errors_in_both_tool_paths(
    tmp_path,
    caplog,
    parallel,
) -> None:
    registry = ToolRegistry()

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

    registry.register(sample, schema=_schema("sample"))

    async def llm_call(**_kwargs):
        return {
            "content": "",
            "tool_calls": [{
                "id": "call-sample",
                "type": "function",
                "function": {"name": "sample", "arguments": "{}"},
            }],
            "finish_reason": "tool_calls",
        }

    runner = AgentRunner(
        trace_store=FileSystemTraceStore(str(tmp_path)),
        tool_registry=registry,
        llm_call=llm_call,
    )

    async def broken_executor(*_args, **_kwargs):
        raise RuntimeError("private framework traceback")

    runner._execute_authorized_tool = broken_executor  # type: ignore[method-assign]
    from agent.tools.builtin.knowledge import KnowledgeConfig

    with caplog.at_level(logging.ERROR):
        result = await runner.run_result(
            [{"role": "user", "content": "run"}],
            RunConfig(
                max_iterations=2,
                tools=["sample"],
                tool_groups=[],
                parallel_tool_execution=parallel,
                knowledge=KnowledgeConfig(
                    enable_extraction=False,
                    enable_completion_extraction=False,
                    enable_injection=False,
                ),
            ),
        )

    assert result["status"] == "failed"
    assert result["failure"]["code"] == "UNEXPECTED_TOOL_ERROR"
    messages = await runner.trace_store.get_trace_messages(result["trace_id"])
    assert "private framework traceback" not in "\n".join(str(item.content) for item in messages)
    assert any(record.exc_info for record in caplog.records)