"""A2A agents end to end, against a live proxy.

An admin registers an agent whose card pins an A2A protocol version and whose
litellm_params route it through the completion bridge; a caller then discovers the
proxy-owned card and drives it over A2A JSON-RPC. These tests assert the recorded
state (the agent persists, a spend row lands) and the enforced behavior (the served
card points back at the proxy, message/send returns a real completion in the pinned
protocol version, and an unsupported version is refused at registration).
"""

from __future__ import annotations

import pytest

from a2a_client import (
    A2ABridgeParams,
    A2AClient,
    A2ADataPart,
    A2AJsonRpcRequest,
    A2AMessageSendParams,
    A2AOutboundMessage,
    A2ASearchPropertiesParams,
    A2ASkill,
    A2ASkillInvocation,
    A2ATextPart,
    AgentCardParams,
    AgentRegisterBody,
    AgentResponse,
    fetch_agent_card,
)
from e2e_config import unique_marker
from e2e_http import Result, UnknownApiError, unwrap
from lifecycle import ResourceManager

# No api_key: litellm resolves ANTHROPIC_API_KEY from the proxy's own environment
# for this provider, which is what the agent-owner flow relies on. Pinning
# "os.environ/ANTHROPIC_API_KEY" here (#34512) was worse than redundant. The a2a
# bridge forwards litellm_params to acompletion() without expanding "os.environ/"
# indirection, so that literal string was sent upstream as x-api-key and every
# message/send failed with `AnthropicException - invalid x-api-key`. Omitting the
# key lets the normal provider resolution apply. Verified against a live proxy:
# omitted -> 200, "os.environ/..." -> 500 invalid x-api-key, literal key -> 200.
BRIDGE = A2ABridgeParams(
    custom_llm_provider="anthropic",
    model="claude-haiku-4-5",
)

MOVEHOME_AGENT_CARD_URL = "https://movehome.org/.well-known/agent.json"
MOVEHOME_ORIGIN = "https://movehome.org"

pytestmark = pytest.mark.e2e


def _register(client: A2AClient, resources: ResourceManager, protocol_version: str) -> AgentResponse:
    marker = unique_marker()
    body = AgentRegisterBody(
        agent_name=f"e2e-a2a-{marker}",
        agent_card_params=AgentCardParams(
            protocol_version=protocol_version,
            name=f"E2E A2A {marker}",
            description="e2e agent backed by the litellm completion bridge",
            version="1.0.0",
            skills=[A2ASkill(id="chat", name="Chat", description="general chat", tags=["chat"])],
        ),
        litellm_params=BRIDGE,
    )
    agent = unwrap(client.register_agent(body))
    resources.defer(lambda: client.delete_agent(agent.agent_id))
    return agent


def _register_rejection(client: A2AClient, protocol_version: str) -> Result[AgentResponse]:
    marker = unique_marker()
    body = AgentRegisterBody(
        agent_name=f"e2e-a2a-bad-{marker}",
        agent_card_params=AgentCardParams(
            protocol_version=protocol_version,
            name=f"E2E A2A bad {marker}",
            description="rejected at registration",
            version="1.0.0",
            skills=[A2ASkill(id="chat", name="Chat", description="c", tags=["chat"])],
        ),
        litellm_params=BRIDGE,
    )
    return client.register_agent(body)


def _ask(text: str) -> A2AJsonRpcRequest:
    return A2AJsonRpcRequest(
        id=f"e2e-{unique_marker()}",
        params=A2AMessageSendParams(
            message=A2AOutboundMessage(parts=[A2ATextPart(text=text)], message_id=unique_marker())
        ),
    )


class TestA2AAgentLifecycle:
    @pytest.mark.covers("other.a2a.register.persists")
    def test_register_persists(self, client: A2AClient, resources: ResourceManager) -> None:
        agent = _register(client, resources, "0.3")
        fetched = unwrap(client.get_agent(agent.agent_id))
        assert fetched.agent_id == agent.agent_id
        assert fetched.agent_name == agent.agent_name
        assert fetched.agent_card_params.protocol_version == "0.3"

    @pytest.mark.covers("other.a2a.register.semver_version_accepted")
    def test_semver_protocol_version_registers_and_serves(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None:
        agent = _register(client, resources, "0.3.0")
        assert agent.agent_card_params.protocol_version == "0.3"
        card = unwrap(client.agent_card(agent.agent_id, scoped_key))
        assert card.protocol_version == "0.3"
        assert card.supported_interfaces is not None
        assert card.supported_interfaces[0].protocol_version == "0.3"
        result = unwrap(client.send_message(agent.agent_id, scoped_key, _ask("Say hi in one word"))).result
        assert result is not None
        assert result.text != ""

    @pytest.mark.covers("other.a2a.message_send.real_world_agent_replies")
    def test_real_world_agent_replies_to_property_query(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None:
        upstream = unwrap(fetch_agent_card(MOVEHOME_AGENT_CARD_URL)).model_copy(update={"url": MOVEHOME_ORIGIN})
        assert upstream.protocol_version == "0.3.0"
        marker = unique_marker()
        body = AgentRegisterBody(agent_name=f"e2e-a2a-real-{marker}", agent_card_params=upstream)
        agent = unwrap(client.register_agent(body))
        resources.defer(lambda: client.delete_agent(agent.agent_id))
        assert agent.agent_card_params.protocol_version == "0.3"
        location = "GBLON"
        request = A2AJsonRpcRequest(
            id=f"e2e-{unique_marker()}",
            params=A2AMessageSendParams(
                message=A2AOutboundMessage(
                    parts=[
                        A2ADataPart(
                            data=A2ASkillInvocation(
                                skill="search_properties",
                                params=A2ASearchPropertiesParams(un_locode=location, service_type="long_term", limit=3),
                            )
                        )
                    ],
                    message_id=unique_marker(),
                )
            ),
        )
        response = unwrap(client.send_message(agent.agent_id, scoped_key, request))
        assert response.error is None
        assert response.result is not None
        results = response.result.search_results
        assert results is not None, "agent returned no search_results artifact; skill did not run"
        assert results.total > 0
        assert results.listings, "search_properties matched nothing; agent returned no property cards"
        assert all(listing.raia_id for listing in results.listings)
        assert all(listing.location.un_locode == location for listing in results.listings)

    @pytest.mark.covers("other.a2a.discovery.proxy_fronted_card")
    def test_discovery_card_is_proxy_fronted(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None:
        agent = _register(client, resources, "0.3")
        card = unwrap(client.agent_card(agent.agent_id, scoped_key))
        assert card.url is not None and card.url.endswith(f"/a2a/{agent.agent_id}")
        assert card.security_schemes is not None
        scheme = next(iter(card.security_schemes.values()))
        assert scheme.scheme == "bearer"
        assert card.supported_interfaces is not None
        assert card.supported_interfaces[0].url == card.url

    @pytest.mark.covers("other.a2a.message_send.bridge_invokes")
    def test_message_send_runs_completion_bridge(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None:
        agent = _register(client, resources, "0.3")
        request = _ask("Reply with exactly the word PONG and nothing else")
        response = unwrap(client.send_message(agent.agent_id, scoped_key, request))
        assert response.error is None
        assert response.result is not None
        assert "PONG" in response.result.text.upper()

        rows = client.proxy.poll_logs_for_request_id(request.id)
        assert rows, f"no spend log row landed for a2a request {request.id}"
        assert rows[0].call_type == "asend_message"
        assert rows[0].model == f"a2a_agent/{agent.agent_card_params.name}"

    @pytest.mark.covers("other.a2a.version.serves_pinned_0_3")
    def test_pinned_v0_3_serves_flat_message_shape(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None:
        agent = _register(client, resources, "0.3")
        request = _ask("Say hi in one word")
        result = unwrap(client.send_message(agent.agent_id, scoped_key, request)).result
        assert result is not None
        assert not result.is_nested_v1_shape
        assert result.kind == "message"
        assert result.role == "agent"
        assert result.text != ""

    @pytest.mark.covers("other.a2a.version.serves_pinned_1_0")
    def test_pinned_v1_0_serves_nested_message_shape(self, client: A2AClient, resources: ResourceManager, scoped_key: str) -> None:
        agent = _register(client, resources, "1.0")
        request = _ask("Say hi in one word")
        result = unwrap(client.send_message(agent.agent_id, scoped_key, request)).result
        assert result is not None
        assert result.is_nested_v1_shape
        assert result.message is not None
        assert result.message.role == "ROLE_AGENT"
        assert result.text != ""

    @pytest.mark.covers("other.a2a.register.unsupported_version_rejected")
    def test_unsupported_protocol_version_rejected(self, client: A2AClient) -> None:
        result = _register_rejection(client, "9.9")
        match result:
            case UnknownApiError(status_code=status, body=detail):
                assert status == 400
                assert "protocolVersion" in detail
            case _:
                pytest.fail(f"expected 400 for unsupported protocolVersion, got {result}")

    @pytest.mark.covers("other.a2a.register.malformed_version_rejected")
    def test_malformed_protocol_version_rejected(self, client: A2AClient) -> None:
        result = _register_rejection(client, "0.3.garbage")
        match result:
            case UnknownApiError(status_code=status, body=detail):
                assert status == 400
                assert "Unsupported protocolVersion '0.3.garbage'" in detail
            case _:
                pytest.fail(f"expected 400 for malformed protocolVersion, got {result}")
