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

from __future__ import annotations

import json
import logging

import pytest

from agent import FailureDetail, FailureDisposition, ToolExecutionError, ToolResult
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)