test_anthropic_client.py

# (C) Datadog, Inc. 2026-present
# All rights reserved
# Licensed under a 3-clause BSD style license (see LICENSE)
from __future__ import annotations

from types import SimpleNamespace
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock

import anthropic
import pytest

from ddev.ai.agent.anthropic_client import (
    MAX_CONTINUATIONS,
    NATIVE_TOOL_DEFINITIONS,
    WEB_FETCH_VERSION,
    WEB_SEARCH_VERSION,
    AnthropicAgent,
    CompletionResult,
)
from ddev.ai.agent.exceptions import AgentAPIError, AgentConnectionError, AgentError, AgentRateLimitError
from ddev.ai.agent.types import StopReason, ToolResultMessage
from ddev.ai.tools.core.types import ToolResult
from ddev.ai.tools.registry import NATIVE_TOOL_NAMES, ToolRegistry

if TYPE_CHECKING:
    from tests.ai.conftest import FakeToolFactory

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def make_usage(
    input_tokens: int = 10,
    output_tokens: int = 20,
    cache_read: int | None = None,
    cache_creation: int | None = None,
    web_search_requests: int = 0,
    web_fetch_requests: int = 0,
) -> SimpleNamespace:
    server_tool_use = SimpleNamespace(web_search_requests=web_search_requests, web_fetch_requests=web_fetch_requests)
    return SimpleNamespace(
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cache_read_input_tokens=cache_read,
        cache_creation_input_tokens=cache_creation,
        server_tool_use=server_tool_use,
    )


def make_text_block(text: str) -> anthropic.types.TextBlock:
    return anthropic.types.TextBlock(type="text", text=text)


def make_tool_use_block(
    id: str = "toolu_01",
    name: str = "read_file",
    input: dict | None = None,
) -> anthropic.types.ToolUseBlock:
    return anthropic.types.ToolUseBlock(
        type="tool_use",
        id=id,
        name=name,
        input=input or {"path": "/tmp/file.txt"},
    )


def make_response(
    stop_reason: str | None,
    content: list,
    usage: SimpleNamespace | None = None,
) -> SimpleNamespace:
    return SimpleNamespace(
        stop_reason=stop_reason,
        content=content,
        usage=usage or make_usage(),
    )


FAKE_CONTEXT_WINDOW = 200_000


def make_agent(
    tools: ToolRegistry | None = None,
    mock_response: SimpleNamespace | None = None,
) -> tuple[AnthropicAgent, AsyncMock]:
    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(return_value=mock_response or make_response("end_turn", []))
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    registry = tools or ToolRegistry([])
    agent = AnthropicAgent(
        client=client,
        tools=registry,
        system_prompt="You are helpful.",
        name="test-agent",
    )
    return agent, client.messages.create


# ---------------------------------------------------------------------------
# end_turn with a single TextBlock
# ---------------------------------------------------------------------------


async def test_end_turn_single_text_block() -> None:
    content = [make_text_block("Hello!")]
    resp = make_response("end_turn", content)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Hi")

    assert result.stop_reason is StopReason.END_TURN
    assert result.text == "Hello!"
    assert result.tool_calls == []
    assert len(agent.history) == 2
    assert agent.history[0] == {"role": "user", "content": "Hi"}
    assert agent.history[1] == {"role": "assistant", "content": content}


# ---------------------------------------------------------------------------
# tool_use
# ---------------------------------------------------------------------------


async def test_tool_use_single_block() -> None:
    block = make_tool_use_block(id="toolu_42", name="read_file", input={"path": "/etc/hosts"})
    resp = make_response("tool_use", [block])
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Read hosts")

    assert result.stop_reason is StopReason.TOOL_USE
    assert len(result.tool_calls) == 1
    tc = result.tool_calls[0]
    assert tc.id == "toolu_42"
    assert tc.name == "read_file"
    assert tc.input == {"path": "/etc/hosts"}


# ---------------------------------------------------------------------------
# mixed TextBlock + ToolUseBlock
# ---------------------------------------------------------------------------


async def test_mixed_text_and_tool_use() -> None:
    content = [
        make_text_block("I'll read the file for you."),
        make_tool_use_block(id="toolu_01", name="read_file"),
    ]
    resp = make_response("tool_use", content)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Read a file")

    assert result.text == "I'll read the file for you."
    assert len(result.tool_calls) == 1


# ---------------------------------------------------------------------------
# Multiple TextBlocks are concatenated
# ---------------------------------------------------------------------------


async def test_multiple_text_blocks_are_concatenated() -> None:
    content = [make_text_block("Hello, "), make_text_block("world!")]
    resp = make_response("end_turn", content)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Hi")

    assert result.text == "Hello, \nworld!"


# ---------------------------------------------------------------------------
# max_tokens is a normal response (not an error)
# ---------------------------------------------------------------------------


async def test_max_tokens_is_not_an_error() -> None:
    resp = make_response("max_tokens", [make_text_block("Truncated...")])
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Tell me everything")

    assert result.stop_reason is StopReason.MAX_TOKENS
    assert len(agent.history) == 2


# ---------------------------------------------------------------------------
# allowed_tools filtering
# ---------------------------------------------------------------------------


async def test_allowed_tools_filters_to_subset(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool(n) for n in ["read_file", "grep", "mkdir"]])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi", allowed_tools=["read_file"])

    sent_names = [t["name"] for t in create_mock.call_args.kwargs["tools"]]
    assert sent_names == ["read_file"]


async def test_allowed_tools_none_passes_all(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool(n) for n in ["a", "b"]])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi", allowed_tools=None)

    sent_names = [t["name"] for t in create_mock.call_args.kwargs["tools"]]
    assert sent_names == ["a", "b"]


@pytest.mark.parametrize("allowed_tools", [[], ["nonexistent_tool"]])
async def test_allowed_tools_passes_not_given(allowed_tools: list[str]) -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(mock_response=resp)

    await agent.send("Hi", allowed_tools=allowed_tools)

    assert create_mock.call_args.kwargs["tools"] is anthropic.NOT_GIVEN


# ---------------------------------------------------------------------------
# API errors map to the correct AgentError subclass
# ---------------------------------------------------------------------------


def _make_error_agent(side_effect: Exception) -> AnthropicAgent:
    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(side_effect=side_effect)
    return AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")


async def test_connection_error_maps_to_agent_connection_error() -> None:
    agent = _make_error_agent(anthropic.APIConnectionError(request=MagicMock()))

    with pytest.raises(AgentConnectionError) as exc_info:
        await agent.send("Hi")

    assert "Connection failed" in str(exc_info.value)
    assert agent.history == []


async def test_rate_limit_error_maps_to_agent_rate_limit_error() -> None:
    agent = _make_error_agent(
        anthropic.RateLimitError(
            message="rate limit",
            response=MagicMock(status_code=429, headers={}),
            body=None,
        )
    )

    with pytest.raises(AgentRateLimitError) as exc_info:
        await agent.send("Hi")

    assert "Rate limit exceeded" in str(exc_info.value)
    assert agent.history == []


async def test_api_status_error_maps_to_agent_api_error() -> None:
    agent = _make_error_agent(
        anthropic.APIStatusError(
            message="internal server error",
            response=MagicMock(status_code=500),
            body=None,
        )
    )

    with pytest.raises(AgentAPIError) as exc_info:
        await agent.send("Hi")

    assert exc_info.value.status_code == 500
    assert agent.history == []


async def test_response_validation_error_maps_to_agent_error() -> None:
    agent = _make_error_agent(anthropic.APIResponseValidationError(response=MagicMock(), body=None))

    with pytest.raises(AgentError) as exc_info:
        await agent.send("Hi")

    assert "Response validation failed" in str(exc_info.value)
    assert agent.history == []


# ---------------------------------------------------------------------------
# Unknown stop_reason raises AgentError, history unchanged
# ---------------------------------------------------------------------------


async def test_unknown_stop_reason_raises_agent_error() -> None:
    resp = make_response("totally_unknown_reason", [])
    agent, _ = make_agent(mock_response=resp)

    with pytest.raises(AgentError) as exc_info:
        await agent.send("Hi")

    assert agent.history == []
    assert "Unknown stop_reason" in str(exc_info.value)
    assert "totally_unknown_reason" in str(exc_info.value)


# ---------------------------------------------------------------------------
# cache_read_input_tokens=None defaults to 0
# ---------------------------------------------------------------------------


async def test_cache_tokens_none_defaults_to_zero() -> None:
    usage = make_usage(cache_read=None, cache_creation=None)
    resp = make_response("end_turn", [make_text_block("ok")], usage=usage)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Hi")

    assert result.usage.cache_read_input_tokens == 0
    assert result.usage.cache_creation_input_tokens == 0


# ---------------------------------------------------------------------------
# ContextUsage fields
# ---------------------------------------------------------------------------


async def test_context_usage_fields() -> None:
    usage = make_usage(input_tokens=1000, cache_read=500, cache_creation=200)
    resp = make_response("end_turn", [make_text_block("ok")], usage=usage)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Hi")

    ctx = result.usage.context_usage
    assert ctx.window_size == FAKE_CONTEXT_WINDOW
    assert ctx.used_tokens == 1700  # 1000 + 500 + 200
    assert ctx.context_pct == pytest.approx(1700 / FAKE_CONTEXT_WINDOW * 100)
    assert ctx.remaining_tokens == FAKE_CONTEXT_WINDOW - 1700


# ---------------------------------------------------------------------------
# context_window is fetched once and cached across multiple sends
# ---------------------------------------------------------------------------


async def test_context_window_fetched_once() -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, _ = make_agent(mock_response=resp)
    agent._client.messages.create = AsyncMock(return_value=resp)

    await agent.send("First")
    await agent.send("Second")

    agent._client.models.retrieve.assert_awaited_once()


# ---------------------------------------------------------------------------
# Multi-turn — send str then send tool results → history has 4 entries
# ---------------------------------------------------------------------------


async def test_multi_turn_history_grows_correctly() -> None:
    tool_resp = make_response("tool_use", [make_tool_use_block(id="toolu_01")])
    text_resp = make_response("end_turn", [make_text_block("Done.")])

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(side_effect=[tool_resp, text_resp])
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    first = await agent.send("Do X")
    assert first.stop_reason is StopReason.TOOL_USE
    assert len(agent.history) == 2

    tool_results = [ToolResultMessage(tool_call_id="toolu_01", result=ToolResult(success=True, data="result"))]
    second = await agent.send(tool_results)
    assert second.stop_reason is StopReason.END_TURN
    assert len(agent.history) == 4
    assert agent.history[2]["role"] == "user"
    assert agent.history[3]["role"] == "assistant"


# ---------------------------------------------------------------------------
# history property returns a copy
# ---------------------------------------------------------------------------


async def test_history_property_returns_copy() -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, _ = make_agent(mock_response=resp)
    await agent.send("Hi")

    snapshot = agent.history
    snapshot.clear()

    assert len(agent.history) == 2


# ---------------------------------------------------------------------------
# reset() clears history
# ---------------------------------------------------------------------------


async def test_reset_clears_history() -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, _ = make_agent(mock_response=resp)
    await agent.send("Hi")
    assert len(agent.history) == 2

    agent.reset()
    assert agent.history == []


# ---------------------------------------------------------------------------
# Native (server) tool injection
# ---------------------------------------------------------------------------


def test_native_tool_definitions_cover_all_native_names() -> None:
    """Every name in NATIVE_TOOL_NAMES must have an entry in NATIVE_TOOL_DEFINITIONS."""
    assert set(NATIVE_TOOL_DEFINITIONS.keys()) == set(NATIVE_TOOL_NAMES)


async def test_native_tool_injected_into_request() -> None:
    registry = ToolRegistry([], native_tool_names=["web_search"])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Search for something")

    sent_tools = create_mock.call_args.kwargs["tools"]
    assert any(t.get("type") == WEB_SEARCH_VERSION for t in sent_tools)


async def test_web_search_max_uses_stays_below_continuation_budget() -> None:
    registry = ToolRegistry([], native_tool_names=["web_search"])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Search for something")

    web_search = next(t for t in create_mock.call_args.kwargs["tools"] if t.get("type") == WEB_SEARCH_VERSION)
    assert web_search["max_uses"] == MAX_CONTINUATIONS - 1


async def test_web_fetch_injected_with_citations_enabled() -> None:
    registry = ToolRegistry([], native_tool_names=["web_fetch"])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Fetch a page")

    web_fetch = next(t for t in create_mock.call_args.kwargs["tools"] if t.get("name") == "web_fetch")
    assert web_fetch["type"] == WEB_FETCH_VERSION
    assert web_fetch["citations"] == {"enabled": True}
    assert web_fetch["max_uses"] == MAX_CONTINUATIONS - 1


async def test_both_native_tools_injected_together(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool("read_file")], native_tool_names=["web_search", "web_fetch"])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi")

    sent_tools = create_mock.call_args.kwargs["tools"]
    sent_names = [t["name"] for t in sent_tools]
    assert sent_names == ["read_file", "web_search", "web_fetch"]
    # Static cache breakpoint only on the last entry.
    assert sent_tools[-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}
    assert all("cache_control" not in t for t in sent_tools[:-1])


async def test_create_until_complete_returns_completion_result() -> None:
    final = make_response("end_turn", [make_text_block("done")])
    agent, _ = make_agent(mock_response=final)

    result = await agent._create_until_complete(request_messages=[], system_param=[], tool_defs=[])

    assert isinstance(result, CompletionResult)
    assert result.final_response is final
    assert result.paused_turns == []
    assert result.all_responses == [final]


async def test_native_tool_appended_after_client_tools(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool("read_file")], native_tool_names=["web_search"])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi")

    sent_tools = create_mock.call_args.kwargs["tools"]
    assert sent_tools[0]["name"] == "read_file"
    assert sent_tools[-1]["name"] == "web_search"
    # Cache breakpoint must be on the last tool (web_search), not on read_file.
    assert "cache_control" not in sent_tools[0]
    assert sent_tools[-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}


async def test_allowed_tools_gates_native_tool(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool("read_file")], native_tool_names=["web_search"])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi", allowed_tools=["read_file"])

    sent_names = [t["name"] for t in create_mock.call_args.kwargs["tools"]]
    assert "web_search" not in sent_names
    assert "read_file" in sent_names


async def test_allowed_tools_none_passes_all_including_native(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool("read_file")], native_tool_names=["web_search"])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi", allowed_tools=None)

    sent_names = [t["name"] for t in create_mock.call_args.kwargs["tools"]]
    assert "read_file" in sent_names
    assert "web_search" in sent_names


async def test_no_native_tools_request_unchanged(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool("read_file")])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi")

    sent_names = [t["name"] for t in create_mock.call_args.kwargs["tools"]]
    assert sent_names == ["read_file"]


async def test_server_result_blocks_preserved_in_history_not_parsed() -> None:
    server_use = SimpleNamespace(type="server_tool_use", id="srvtoolu_01", name="web_search")
    search_result = SimpleNamespace(type="web_search_tool_result", tool_use_id="srvtoolu_01", content=[])
    content = [
        make_text_block("Searching..."),
        server_use,
        search_result,
        make_text_block("Here are the results."),
    ]
    resp = make_response("end_turn", content)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Search for X")

    assert result.tool_calls == []
    assert "Searching..." in result.text
    assert "Here are the results." in result.text
    assert agent.history[-1] == {"role": "assistant", "content": content}


def make_server_tool_use(id: str, query: str) -> anthropic.types.ServerToolUseBlock:
    return anthropic.types.ServerToolUseBlock(type="server_tool_use", id=id, name="web_search", input={"query": query})


def make_web_search_result(tool_use_id: str, result_count: int) -> anthropic.types.WebSearchToolResultBlock:
    results = [
        anthropic.types.WebSearchResultBlock(
            type="web_search_result", encrypted_content="x", page_age=None, title=f"r{i}", url=f"http://x/{i}"
        )
        for i in range(result_count)
    ]
    return anthropic.types.WebSearchToolResultBlock(
        type="web_search_tool_result", tool_use_id=tool_use_id, content=results
    )


async def test_web_searches_surfaced_with_query_and_result_count() -> None:
    content = [
        make_text_block("Searching..."),
        make_server_tool_use("srv1", "weather in Tuvalu"),
        make_web_search_result("srv1", result_count=3),
        make_text_block("Here is the forecast."),
    ]
    resp = make_response("end_turn", content)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Search for the weather")

    assert result.tool_calls == []
    assert len(result.web_activity.searches) == 1
    assert result.web_activity.searches[0].query == "weather in Tuvalu"
    assert result.web_activity.searches[0].result_count == 3
    assert result.web_activity.searches[0].error is None


async def test_web_search_error_surfaced() -> None:
    error = anthropic.types.WebSearchToolResultError(
        type="web_search_tool_result_error", error_code="max_uses_exceeded"
    )
    content = [
        make_server_tool_use("srv1", "too many"),
        anthropic.types.WebSearchToolResultBlock(type="web_search_tool_result", tool_use_id="srv1", content=error),
    ]
    agent, _ = make_agent(mock_response=make_response("end_turn", content))

    result = await agent.send("Search")

    assert len(result.web_activity.searches) == 1
    assert result.web_activity.searches[0].error == "max_uses_exceeded"
    assert result.web_activity.searches[0].result_count == 0


async def test_no_web_searches_yields_empty_activity() -> None:
    agent, _ = make_agent(mock_response=make_response("end_turn", [make_text_block("hi")]))

    result = await agent.send("Hi")

    assert result.web_activity.searches == []
    assert result.web_activity.fetches == []
    assert result.web_activity.citations == []


def make_server_fetch_use(id: str, url: str) -> anthropic.types.ServerToolUseBlock:
    return anthropic.types.ServerToolUseBlock(type="server_tool_use", id=id, name="web_fetch", input={"url": url})


def make_web_fetch_result(tool_use_id: str, url: str, retrieved_at: str) -> anthropic.types.WebFetchToolResultBlock:
    doc = anthropic.types.DocumentBlock(
        type="document",
        source=anthropic.types.PlainTextSource(type="text", media_type="text/plain", data="page body"),
    )
    fetched = anthropic.types.WebFetchBlock(type="web_fetch_result", url=url, retrieved_at=retrieved_at, content=doc)
    return anthropic.types.WebFetchToolResultBlock(
        type="web_fetch_tool_result", tool_use_id=tool_use_id, content=fetched
    )


async def test_web_fetches_surfaced_with_url_and_retrieved_at() -> None:
    content = [
        make_text_block("Fetching..."),
        make_server_fetch_use("srv1", "https://example.com/doc"),
        make_web_fetch_result("srv1", "https://example.com/doc", "2026-01-01T00:00:00Z"),
        make_text_block("Here is the content."),
    ]
    resp = make_response("end_turn", content)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Fetch the doc")

    assert result.tool_calls == []
    assert len(result.web_activity.fetches) == 1
    fetch = result.web_activity.fetches[0]
    assert fetch.url == "https://example.com/doc"
    assert fetch.retrieved_at == "2026-01-01T00:00:00Z"
    assert fetch.error is None


async def test_web_fetch_error_surfaced() -> None:
    error = anthropic.types.WebFetchToolResultErrorBlock(
        type="web_fetch_tool_result_error", error_code="url_not_accessible"
    )
    content = [
        make_server_fetch_use("srv1", "https://example.com/missing"),
        anthropic.types.WebFetchToolResultBlock(type="web_fetch_tool_result", tool_use_id="srv1", content=error),
    ]
    agent, _ = make_agent(mock_response=make_response("end_turn", content))

    result = await agent.send("Fetch")

    assert len(result.web_activity.fetches) == 1
    assert result.web_activity.fetches[0].error == "url_not_accessible"
    assert result.web_activity.fetches[0].retrieved_at is None


async def test_web_fetch_requests_summed_into_usage() -> None:
    resp = make_response("end_turn", [make_text_block("done")], usage=make_usage(web_fetch_requests=2))
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Fetch")

    assert result.usage.web_fetch_requests == 2


async def test_web_fetch_result_preserved_in_history_not_parsed() -> None:
    content = [
        make_text_block("Fetching..."),
        make_server_fetch_use("srv1", "https://example.com/doc"),
        make_web_fetch_result("srv1", "https://example.com/doc", "2026-01-01T00:00:00Z"),
        make_text_block("Done."),
    ]
    resp = make_response("end_turn", content)
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Fetch")

    assert result.tool_calls == []
    assert agent.history[-1] == {"role": "assistant", "content": content}


async def test_web_search_citations_extracted_from_text_block() -> None:
    citation = anthropic.types.CitationsWebSearchResultLocation(
        type="web_search_result_location",
        url="https://example.com/page",
        title="Example Page",
        cited_text="some cited excerpt",
        encrypted_index="enc123",
    )
    block = anthropic.types.TextBlock(type="text", text="Here is the answer.", citations=[citation])
    resp = make_response("end_turn", [block])
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("What is X?")

    assert len(result.web_activity.citations) == 1
    c = result.web_activity.citations[0]
    assert c.url == "https://example.com/page"
    assert c.title == "Example Page"
    assert c.cited_text == "some cited excerpt"


async def test_web_fetch_char_citation_surfaced() -> None:
    char_citation = anthropic.types.CitationCharLocation(
        type="char_location",
        cited_text="some cited text",
        document_index=0,
        document_title="Fetched Doc",
        end_char_index=10,
        start_char_index=0,
    )
    block = anthropic.types.TextBlock(type="text", text="Based on the doc.", citations=[char_citation])
    resp = make_response("end_turn", [block])
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Hi")

    assert len(result.web_activity.citations) == 1
    c = result.web_activity.citations[0]
    assert c.url is None
    assert c.title == "Fetched Doc"
    assert c.cited_text == "some cited text"


async def test_citations_across_multiple_text_blocks() -> None:
    citation1 = anthropic.types.CitationsWebSearchResultLocation(
        type="web_search_result_location",
        url="https://a.com",
        title="A",
        cited_text="text a",
        encrypted_index="e1",
    )
    citation2 = anthropic.types.CitationsWebSearchResultLocation(
        type="web_search_result_location",
        url="https://b.com",
        title="B",
        cited_text="text b",
        encrypted_index="e2",
    )
    block1 = anthropic.types.TextBlock(type="text", text="First.", citations=[citation1])
    block2 = anthropic.types.TextBlock(type="text", text="Second.", citations=[citation2])
    resp = make_response("end_turn", [block1, block2])
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Hi")

    assert len(result.web_activity.citations) == 2
    assert result.web_activity.citations[0].url == "https://a.com"
    assert result.web_activity.citations[1].url == "https://b.com"


# ---------------------------------------------------------------------------
# pause_turn continuation loop
# ---------------------------------------------------------------------------


async def test_pause_turn_triggers_continuation() -> None:
    pause_content = [make_text_block("Searching...")]
    pause_resp = make_response("pause_turn", pause_content)
    final_resp = make_response("end_turn", [make_text_block("Done.")])

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(side_effect=[pause_resp, final_resp])
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    result = await agent.send("Hi")

    assert client.messages.create.await_count == 2
    assert result.stop_reason is StopReason.END_TURN
    # Paused turn must appear in history before the final assistant turn.
    assert agent.history[-2] == {"role": "assistant", "content": pause_content}
    assert agent.history[-1] == {"role": "assistant", "content": final_resp.content}


async def test_pause_turn_second_call_includes_paused_turn_in_messages() -> None:
    pause_content = [make_text_block("interim")]
    pause_resp = make_response("pause_turn", pause_content)
    final_resp = make_response("end_turn", [make_text_block("done")])

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(side_effect=[pause_resp, final_resp])
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    await agent.send("Hi")

    second_call_messages = client.messages.create.call_args_list[1].kwargs["messages"]
    assert second_call_messages[-1] == {"role": "assistant", "content": pause_content}


async def test_multiple_consecutive_pause_turns() -> None:
    pause1_content = [make_text_block("p1")]
    pause2_content = [make_text_block("p2")]
    final_content = [make_text_block("done")]
    pause1 = make_response("pause_turn", pause1_content)
    pause2 = make_response("pause_turn", pause2_content)
    final = make_response("end_turn", final_content)

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(side_effect=[pause1, pause2, final])
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    result = await agent.send("Hi")

    assert client.messages.create.await_count == 3
    assert result.stop_reason is StopReason.END_TURN
    # Both paused turns appear in history in order.
    assert agent.history[-3] == {"role": "assistant", "content": pause1_content}
    assert agent.history[-2] == {"role": "assistant", "content": pause2_content}
    assert agent.history[-1] == {"role": "assistant", "content": final_content}


async def test_error_during_paused_continuation_leaves_history_unchanged() -> None:
    ok_resp = make_response("end_turn", [make_text_block("ok")])
    pause_resp = make_response("pause_turn", [make_text_block("pausing")])

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(
        side_effect=[
            ok_resp,
            pause_resp,
            anthropic.APIConnectionError(request=MagicMock()),
        ]
    )
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    await agent.send("First")
    history_after_first = agent.history[:]

    with pytest.raises(AgentConnectionError):
        await agent.send("Second")

    assert agent.history == history_after_first


async def test_pause_turn_token_usage_summed_across_calls() -> None:
    pause_usage = make_usage(input_tokens=100, output_tokens=50, web_search_requests=2)
    final_usage = make_usage(input_tokens=200, output_tokens=80, web_search_requests=1)
    pause_resp = make_response("pause_turn", [make_text_block("p")], usage=pause_usage)
    final_resp = make_response("end_turn", [make_text_block("done")], usage=final_usage)

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(side_effect=[pause_resp, final_resp])
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    result = await agent.send("Hi")

    assert result.usage.input_tokens == 300  # 100 + 200
    assert result.usage.output_tokens == 130  # 50 + 80
    assert result.usage.web_search_requests == 3  # 2 + 1


# ---------------------------------------------------------------------------
# refusal and stop_sequence map to StopReason.OTHER
# ---------------------------------------------------------------------------


async def test_refusal_maps_to_other() -> None:
    resp = make_response("refusal", [make_text_block("I can't do that")])
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Do something bad")

    assert result.stop_reason is StopReason.OTHER


async def test_stop_sequence_maps_to_other() -> None:
    resp = make_response("stop_sequence", [make_text_block("Stopped.")])
    agent, _ = make_agent(mock_response=resp)

    result = await agent.send("Hi")

    assert result.stop_reason is StopReason.OTHER


# ---------------------------------------------------------------------------
# Error mid-conversation leaves history unchanged
# ---------------------------------------------------------------------------


async def test_error_mid_conversation_leaves_history_unchanged() -> None:
    ok_resp = make_response("end_turn", [make_text_block("ok")])
    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(
        side_effect=[
            ok_resp,
            anthropic.APIConnectionError(request=MagicMock()),
        ]
    )
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    await agent.send("First message")
    history_after_first = agent.history[:]

    with pytest.raises(AgentConnectionError):
        await agent.send("Second message")

    assert agent.history == history_after_first


# ---------------------------------------------------------------------------
# Prompt caching: static breakpoints (system + last tool, 1h TTL)
# ---------------------------------------------------------------------------


async def test_system_prompt_sent_as_block_with_static_cache_control() -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(mock_response=resp)

    await agent.send("Hi")

    assert create_mock.call_args.kwargs["system"] == [
        {
            "type": "text",
            "text": "You are helpful.",
            "cache_control": {"type": "ephemeral", "ttl": "1h"},
        }
    ]


@pytest.mark.parametrize(
    "tool_names",
    [["only"], ["a", "b"], ["a", "b", "c", "d"]],
    ids=["single_tool", "two_tools", "four_tools"],
)
async def test_only_last_tool_carries_static_cache_control(tool_names: list[str], fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool(n) for n in tool_names])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi")

    sent_tools = create_mock.call_args.kwargs["tools"]
    assert all("cache_control" not in t for t in sent_tools[:-1])
    assert sent_tools[-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}


async def test_allowed_tools_subset_places_cache_control_on_last_of_subset(fake_tool: FakeToolFactory) -> None:
    registry = ToolRegistry([fake_tool(n) for n in ["a", "b", "c"]])
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(tools=registry, mock_response=resp)

    await agent.send("Hi", allowed_tools=["a", "b"])

    sent_tools = create_mock.call_args.kwargs["tools"]
    assert [t["name"] for t in sent_tools] == ["a", "b"]
    assert "cache_control" not in sent_tools[0]
    assert sent_tools[-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}


# ---------------------------------------------------------------------------
# Prompt caching: sliding breakpoint on the last user message block (default TTL)
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "content",
    [
        pytest.param("Hi there", id="str"),
        pytest.param(
            [ToolResultMessage(tool_call_id="t1", result=ToolResult(success=True, data="r1"))],
            id="single_tool_result",
        ),
        pytest.param(
            [ToolResultMessage(tool_call_id=f"t{i}", result=ToolResult(success=True, data=f"r{i}")) for i in range(3)],
            id="multiple_tool_results",
        ),
    ],
)
async def test_sliding_cache_control_on_last_user_block_only(
    content: str | list[ToolResultMessage],
) -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(mock_response=resp)

    await agent.send(content)

    blocks = create_mock.call_args.kwargs["messages"][-1]["content"]
    assert isinstance(blocks, list) and blocks
    assert all("cache_control" not in b for b in blocks[:-1])
    assert blocks[-1]["cache_control"] == {"type": "ephemeral"}
    assert "ttl" not in blocks[-1]["cache_control"]


async def test_empty_tool_results_produces_empty_content_block() -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, create_mock = make_agent(mock_response=resp)

    await agent.send([])

    blocks = create_mock.call_args.kwargs["messages"][-1]["content"]
    assert blocks == []


# ---------------------------------------------------------------------------
# Prompt caching: history must not retain cache_control markers
# (otherwise multi-turn requests would exceed the 4-marker limit)
# ---------------------------------------------------------------------------


async def test_history_str_message_keeps_raw_str_form() -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, _ = make_agent(mock_response=resp)

    await agent.send("Hi")

    assert agent.history[0] == {"role": "user", "content": "Hi"}


async def test_history_tool_result_blocks_have_no_cache_control() -> None:
    resp = make_response("end_turn", [make_text_block("ok")])
    agent, _ = make_agent(mock_response=resp)

    tool_results = [
        ToolResultMessage(tool_call_id=f"t{i}", result=ToolResult(success=True, data=f"r{i}")) for i in range(2)
    ]
    await agent.send(tool_results)

    blocks = agent.history[0]["content"]
    assert all("cache_control" not in b for b in blocks)


async def test_multi_turn_only_latest_user_message_in_request_has_cache_control() -> None:
    first_resp = make_response("tool_use", [make_tool_use_block(id="t1")])
    second_resp = make_response("end_turn", [make_text_block("done")])

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(side_effect=[first_resp, second_resp])
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="sp", name="t")

    await agent.send("First")
    await agent.send([ToolResultMessage(tool_call_id="t1", result=ToolResult(success=True, data="r"))])

    first_call_messages = client.messages.create.call_args_list[0].kwargs["messages"]
    assert first_call_messages[-1]["content"] == [
        {"type": "text", "text": "First", "cache_control": {"type": "ephemeral"}}
    ]

    second_call_messages = client.messages.create.call_args_list[1].kwargs["messages"]
    assert second_call_messages[0] == {"role": "user", "content": "First"}
    latest_blocks = second_call_messages[-1]["content"]
    assert all("cache_control" not in b for b in latest_blocks[:-1])
    assert latest_blocks[-1]["cache_control"] == {"type": "ephemeral"}


# ---------------------------------------------------------------------------
# pause_turn continuation cap
# ---------------------------------------------------------------------------


async def test_pause_turn_raises_after_max_continuations() -> None:
    pause_resp = make_response("pause_turn", [make_text_block("still searching...")])

    client = MagicMock(spec=anthropic.AsyncAnthropic)
    client.messages = MagicMock()
    client.messages.create = AsyncMock(return_value=pause_resp)
    client.models = MagicMock()
    client.models.retrieve = AsyncMock(return_value=SimpleNamespace(max_input_tokens=FAKE_CONTEXT_WINDOW))
    agent = AnthropicAgent(client=client, tools=ToolRegistry([]), system_prompt="", name="t")

    with pytest.raises(AgentError, match=f"pause_turn did not resolve after {MAX_CONTINUATIONS} continuations"):
        await agent.send("Hi")

    assert client.messages.create.await_count == MAX_CONTINUATIONS