diff --git a/src/a2a/compat/v0_3/grpc_transport.py b/src/a2a/compat/v0_3/grpc_transport.py index 26faaa3d3..425b0d2f0 100644 --- a/src/a2a/compat/v0_3/grpc_transport.py +++ b/src/a2a/compat/v0_3/grpc_transport.py @@ -33,6 +33,7 @@ from a2a.compat.v0_3.extension_headers import add_legacy_extension_header from a2a.types import a2a_pb2 from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER +from a2a.utils.task import apply_history_length from a2a.utils.telemetry import SpanKind, trace_class @@ -130,8 +131,11 @@ async def send_message( which = resp_proto.WhichOneof('payload') if which == 'task': return a2a_pb2.SendMessageResponse( - task=conversions.to_core_task( - proto_utils.FromProto.task(resp_proto.task) + task=apply_history_length( + conversions.to_core_task( + proto_utils.FromProto.task(resp_proto.task) + ), + request.configuration, ) ) if which == 'msg': @@ -217,7 +221,12 @@ async def get_task( req_proto, metadata=self._get_grpc_metadata(context), ) - return conversions.to_core_task(proto_utils.FromProto.task(resp_proto)) + # v0.3 has no way to ask for an empty history (0 means unlimited), so + # the v1.0 `history_length` semantics are applied to the result here. + return apply_history_length( + conversions.to_core_task(proto_utils.FromProto.task(resp_proto)), + request, + ) @_handle_grpc_exception async def list_tasks( diff --git a/src/a2a/compat/v0_3/jsonrpc_transport.py b/src/a2a/compat/v0_3/jsonrpc_transport.py index f3369910b..e09c0e288 100644 --- a/src/a2a/compat/v0_3/jsonrpc_transport.py +++ b/src/a2a/compat/v0_3/jsonrpc_transport.py @@ -40,6 +40,7 @@ ) from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP +from a2a.utils.task import apply_history_length from a2a.utils.telemetry import SpanKind, trace_class @@ -105,8 +106,11 @@ async def send_message( if kind == 'task': return SendMessageResponse( - task=conversions.to_core_task( - types_v03.Task.model_validate(result_dict) + task=apply_history_length( + conversions.to_core_task( + types_v03.Task.model_validate(result_dict) + ), + request.configuration, ) ) if kind == 'message': @@ -164,8 +168,13 @@ async def get_task( json_rpc_response = JSONRPC20Response(**response_data) if json_rpc_response.error: raise self._create_jsonrpc_error(json_rpc_response.error) - return conversions.to_core_task( - types_v03.Task.model_validate(json_rpc_response.result) + # v0.3 has no way to ask for an empty history (0 means unlimited), so + # the v1.0 `history_length` semantics are applied to the result here. + return apply_history_length( + conversions.to_core_task( + types_v03.Task.model_validate(json_rpc_response.result) + ), + request, ) async def list_tasks( diff --git a/src/a2a/compat/v0_3/rest_transport.py b/src/a2a/compat/v0_3/rest_transport.py index cdeed2867..4d4c22622 100644 --- a/src/a2a/compat/v0_3/rest_transport.py +++ b/src/a2a/compat/v0_3/rest_transport.py @@ -46,6 +46,7 @@ ) from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, MethodNotFoundError +from a2a.utils.task import apply_history_length from a2a.utils.telemetry import SpanKind, trace_class @@ -107,8 +108,11 @@ async def send_message( which = resp_proto.WhichOneof('payload') if which == 'task': return SendMessageResponse( - task=conversions.to_core_task( - proto_utils.FromProto.task(resp_proto.task) + task=apply_history_length( + conversions.to_core_task( + proto_utils.FromProto.task(resp_proto.task) + ), + request.configuration, ) ) if which == 'msg': @@ -165,7 +169,12 @@ async def get_task( resp_proto = ParseDict( response_data, a2a_v0_3_pb2.Task(), ignore_unknown_fields=True ) - return conversions.to_core_task(proto_utils.FromProto.task(resp_proto)) + # v0.3 has no way to ask for an empty history (0 means unlimited), so + # the v1.0 `history_length` semantics are applied to the result here. + return apply_history_length( + conversions.to_core_task(proto_utils.FromProto.task(resp_proto)), + request, + ) async def list_tasks( self, diff --git a/tests/compat/v0_3/test_history_length_compat.py b/tests/compat/v0_3/test_history_length_compat.py new file mode 100644 index 000000000..19b8cf767 --- /dev/null +++ b/tests/compat/v0_3/test_history_length_compat.py @@ -0,0 +1,167 @@ +"""A v0.3 agent cannot be asked for an empty history, so the compat transports +must apply the v1.0 `history_length` semantics to the task they return. + +In v0.3 `history_length = 0` means *unlimited* (see +`tests/compat/v0_3/test_proto_utils.py::test_message_send_configuration_history_length`), +while in v1.0 it means *no history*. Forwarding `0` to a v0.3 agent therefore +returns the whole history, so each compat transport has to truncate locally. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from a2a.compat.v0_3.grpc_transport import CompatGrpcTransport +from a2a.compat.v0_3.jsonrpc_transport import CompatJsonRpcTransport +from a2a.compat.v0_3.rest_transport import CompatRestTransport +from a2a.types.a2a_pb2 import ( + AgentCard, + GetTaskRequest, + Role, + SendMessageRequest, + TaskState, +) + + +URL = 'http://example.com' + + +def _task_dict(proto_style: bool) -> dict: + """A v0.3 task payload carrying three history messages. + + `proto_style` selects the enum spelling used by the protobuf-backed + transports; the JSON-RPC transport validates against the v0.3 pydantic + models, which use the short names. + """ + role = 'ROLE_USER' if proto_style else 'user' + state = 'TASK_STATE_WORKING' if proto_style else 'working' + return { + 'id': 'task-1', + 'contextId': 'ctx-1', + 'status': {'state': state}, + 'history': [ + {'messageId': f'm{i}', 'role': role, 'parts': []} for i in range(3) + ], + } + + +def _rpc_ok(result: dict) -> dict: + return {'id': '0', 'jsonrpc': '2.0', 'result': result} + + +def _rest_transport() -> CompatRestTransport: + t = CompatRestTransport( + httpx_client=AsyncMock(), + agent_card=AgentCard(name='a', version='1'), + url=URL, + ) + + async def _execute_request(method, path, context=None, **kwargs): + return _task_dict(proto_style=True) + + t._execute_request = _execute_request + return t + + +def _jsonrpc_transport() -> CompatJsonRpcTransport: + t = CompatJsonRpcTransport( + httpx_client=AsyncMock(), + agent_card=AgentCard(name='a', version='1'), + url=URL, + ) + + async def _send_request(payload_, context=None): + return _rpc_ok(_task_dict(proto_style=False)) + + t._send_request = _send_request + return t + + +def _grpc_transport() -> CompatGrpcTransport: + import a2a.compat.v0_3.a2a_v0_3_pb2 as v03 + + payload = _task_dict(proto_style=True) + task = v03.Task() + task.id = payload['id'] + task.context_id = payload['contextId'] + setattr(task.status, 'state', TaskState.TASK_STATE_WORKING) + for m in payload['history']: + msg = task.history.add() + msg.message_id = m['messageId'] + msg.role = Role.ROLE_USER + + t = CompatGrpcTransport( + channel=MagicMock(), agent_card=AgentCard(name='a', version='1') + ) + + async def _get_task(req, metadata=None): + return task + + t.stub = MagicMock(GetTask=AsyncMock(side_effect=_get_task)) + return t + + +TRANSPORTS = [_rest_transport, _jsonrpc_transport, _grpc_transport] + + +class TestCompatGetTaskHistoryLength: + @pytest.mark.asyncio + @pytest.mark.parametrize('factory', TRANSPORTS) + async def test_history_length_zero_drops_history(self, factory): + t = factory() + task = await t.get_task(GetTaskRequest(id='task-1', history_length=0)) + assert len(task.history) == 0 + + @pytest.mark.asyncio + @pytest.mark.parametrize('factory', TRANSPORTS) + async def test_history_length_keeps_last_n(self, factory): + t = factory() + task = await t.get_task(GetTaskRequest(id='task-1', history_length=1)) + assert [m.message_id for m in task.history] == ['m2'] + + @pytest.mark.asyncio + @pytest.mark.parametrize('factory', TRANSPORTS) + async def test_unset_history_length_keeps_history(self, factory): + t = factory() + task = await t.get_task(GetTaskRequest(id='task-1')) + assert len(task.history) == 3 + + +class TestCompatSendMessageHistoryLength: + @pytest.mark.asyncio + async def test_rest_send_message_applies_history_length(self): + t = CompatRestTransport( + httpx_client=AsyncMock(), + agent_card=AgentCard(name='a', version='1'), + url=URL, + ) + + async def _execute_request(method, path, context=None, **kwargs): + return {'task': _task_dict(proto_style=True)} + + t._execute_request = _execute_request + req = SendMessageRequest() + req.configuration.history_length = 0 + resp = await t.send_message(req) + assert resp.HasField('task') + assert len(resp.task.history) == 0 + + @pytest.mark.asyncio + async def test_jsonrpc_send_message_applies_history_length(self): + t = CompatJsonRpcTransport( + httpx_client=AsyncMock(), + agent_card=AgentCard(name='a', version='1'), + url=URL, + ) + + async def _send_request(payload_, context=None): + payload = _task_dict(proto_style=False) + payload['kind'] = 'task' + return _rpc_ok(payload) + + t._send_request = _send_request + req = SendMessageRequest() + req.configuration.history_length = 0 + resp = await t.send_message(req) + assert resp.HasField('task') + assert len(resp.task.history) == 0