From 85d6bf20f61ba9fe6707d16d7328862e80bd13b8 Mon Sep 17 00:00:00 2001 From: D41910 <158550716+D41910@users.noreply.github.com> Date: Thu, 15 Oct 2026 09:53:25 +0800 Subject: [PATCH 1/2] fix(compat): apply v1.0 history_length semantics to v0.3 results --- src/a2a/compat/v0_3/grpc_transport.py | 15 +- src/a2a/compat/v0_3/jsonrpc_transport.py | 17 +- src/a2a/compat/v0_3/rest_transport.py | 15 +- .../compat/v0_3/test_history_length_compat.py | 167 ++++++++++++++++++ 4 files changed, 204 insertions(+), 10 deletions(-) create mode 100644 tests/compat/v0_3/test_history_length_compat.py 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 From bea5c321eb88bf460ed3aacc9b1f8ee7bd59a981 Mon Sep 17 00:00:00 2001 From: Winoren <158550716+Winoren@users.noreply.github.com> Date: Sun, 25 Oct 2026 15:04:24 +0800 Subject: [PATCH 2/2] test(compat): colocate history truncation transport cases --- tests/compat/v0_3/test_grpc_transport.py | 30 ++++ .../compat/v0_3/test_history_length_compat.py | 167 ------------------ tests/compat/v0_3/test_jsonrpc_transport.py | 77 ++++++++ tests/compat/v0_3/test_rest_transport.py | 76 ++++++++ 4 files changed, 183 insertions(+), 167 deletions(-) delete mode 100644 tests/compat/v0_3/test_history_length_compat.py diff --git a/tests/compat/v0_3/test_grpc_transport.py b/tests/compat/v0_3/test_grpc_transport.py index 402a57000..f12a86792 100644 --- a/tests/compat/v0_3/test_grpc_transport.py +++ b/tests/compat/v0_3/test_grpc_transport.py @@ -7,6 +7,7 @@ from a2a.compat.v0_3 import a2a_v0_3_pb2 from a2a.compat.v0_3.grpc_transport import CompatGrpcTransport from a2a.types.a2a_pb2 import ( + GetTaskRequest, Message, Role, SendMessageRequest, @@ -66,3 +67,32 @@ def test_compat_grpc_transport_no_extension_metadata(): assert 'a2a-extensions' not in metadata assert 'x-a2a-extensions' not in metadata + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'history_length, expected_ids', + [ + (None, ['m0', 'm1', 'm2']), + (0, []), + (1, ['m2']), + (3, ['m0', 'm1', 'm2']), + (5, ['m0', 'm1', 'm2']), + ], +) +async def test_get_task_history_length( + history_length: int | None, expected_ids: list[str] +) -> None: + transport = CompatGrpcTransport( + channel=AsyncMock(spec=Channel), agent_card=None + ) + task = a2a_v0_3_pb2.Task(id='task-1', context_id='ctx-1') + task.status.state = a2a_v0_3_pb2.TaskState.TASK_STATE_WORKING + for i in range(3): + task.history.add(message_id=f'm{i}', role=a2a_v0_3_pb2.Role.ROLE_USER) + transport.stub = MagicMock(GetTask=AsyncMock(return_value=task)) + req = GetTaskRequest(id='task-1') + if history_length is not None: + req.history_length = history_length + result = await transport.get_task(req) + assert [message.message_id for message in result.history] == expected_ids diff --git a/tests/compat/v0_3/test_history_length_compat.py b/tests/compat/v0_3/test_history_length_compat.py deleted file mode 100644 index 19b8cf767..000000000 --- a/tests/compat/v0_3/test_history_length_compat.py +++ /dev/null @@ -1,167 +0,0 @@ -"""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 diff --git a/tests/compat/v0_3/test_jsonrpc_transport.py b/tests/compat/v0_3/test_jsonrpc_transport.py index 70291f005..923498b6a 100644 --- a/tests/compat/v0_3/test_jsonrpc_transport.py +++ b/tests/compat/v0_3/test_jsonrpc_transport.py @@ -565,3 +565,80 @@ async def test_compat_jsonrpc_transport_mirrors_extension_header( headers = kwargs['headers'] assert headers['A2A-Extensions'] == 'foo,bar' assert headers['X-A2A-Extensions'] == 'foo,bar' + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'history_length, expected_ids', + [ + (None, ['m0', 'm1', 'm2']), + (0, []), + (1, ['m2']), + (3, ['m0', 'm1', 'm2']), + (5, ['m0', 'm1', 'm2']), + ], +) +async def test_get_task_history_length( + transport: CompatJsonRpcTransport, + history_length: int | None, + expected_ids: list[str], +) -> None: + # v0.3 treats zero as unlimited; v1.0 requires an empty local history. + payload = { + 'id': 'task-1', + 'contextId': 'ctx-1', + 'status': {'state': 'working'}, + 'history': [ + {'messageId': f'm{i}', 'role': 'user', 'parts': []} + for i in range(3) + ], + } + with patch.object( + transport, '_send_request', new_callable=AsyncMock + ) as request: + request.return_value = {'jsonrpc': '2.0', 'id': '0', 'result': payload} + req = GetTaskRequest(id='task-1') + if history_length is not None: + req.history_length = history_length + task = await transport.get_task(req) + assert [message.message_id for message in task.history] == expected_ids + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'history_length, expected_ids', + [ + (None, ['m0', 'm1', 'm2']), + (0, []), + (1, ['m2']), + (3, ['m0', 'm1', 'm2']), + (5, ['m0', 'm1', 'm2']), + ], +) +async def test_send_message_history_length( + transport: CompatJsonRpcTransport, + history_length: int | None, + expected_ids: list[str], +) -> None: + # v0.3 treats zero as unlimited; v1.0 requires an empty local history. + payload = { + 'id': 'task-1', + 'contextId': 'ctx-1', + 'status': {'state': 'working'}, + 'history': [ + {'messageId': f'm{i}', 'role': 'user', 'parts': []} + for i in range(3) + ], + } + payload['kind'] = 'task' + with patch.object( + transport, '_send_request', new_callable=AsyncMock + ) as request: + request.return_value = {'jsonrpc': '2.0', 'id': '0', 'result': payload} + req = SendMessageRequest() + if history_length is not None: + req.configuration.history_length = history_length + response = await transport.send_message(req) + assert response.HasField('task') + task = response.task + assert [message.message_id for message in task.history] == expected_ids diff --git a/tests/compat/v0_3/test_rest_transport.py b/tests/compat/v0_3/test_rest_transport.py index 0952aaa80..f265a3f0a 100644 --- a/tests/compat/v0_3/test_rest_transport.py +++ b/tests/compat/v0_3/test_rest_transport.py @@ -690,3 +690,79 @@ async def test_compat_rest_transport_execute_request( mock_send_http_request.assert_called_once_with( transport.httpx_client, mock_request, transport._handle_http_error ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'history_length, expected_ids', + [ + (None, ['m0', 'm1', 'm2']), + (0, []), + (1, ['m2']), + (3, ['m0', 'm1', 'm2']), + (5, ['m0', 'm1', 'm2']), + ], +) +async def test_get_task_history_length( + transport: CompatRestTransport, + history_length: int | None, + expected_ids: list[str], +) -> None: + # v0.3 treats zero as unlimited; v1.0 requires an empty local history. + payload = { + 'id': 'task-1', + 'contextId': 'ctx-1', + 'status': {'state': 'TASK_STATE_WORKING'}, + 'history': [ + {'messageId': f'm{i}', 'role': 'ROLE_USER', 'parts': []} + for i in range(3) + ], + } + with patch.object( + transport, '_execute_request', new_callable=AsyncMock + ) as request: + request.return_value = payload + req = GetTaskRequest(id='task-1') + if history_length is not None: + req.history_length = history_length + task = await transport.get_task(req) + assert [message.message_id for message in task.history] == expected_ids + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'history_length, expected_ids', + [ + (None, ['m0', 'm1', 'm2']), + (0, []), + (1, ['m2']), + (3, ['m0', 'm1', 'm2']), + (5, ['m0', 'm1', 'm2']), + ], +) +async def test_send_message_history_length( + transport: CompatRestTransport, + history_length: int | None, + expected_ids: list[str], +) -> None: + # v0.3 treats zero as unlimited; v1.0 requires an empty local history. + payload = { + 'id': 'task-1', + 'contextId': 'ctx-1', + 'status': {'state': 'TASK_STATE_WORKING'}, + 'history': [ + {'messageId': f'm{i}', 'role': 'ROLE_USER', 'parts': []} + for i in range(3) + ], + } + with patch.object( + transport, '_execute_request', new_callable=AsyncMock + ) as request: + request.return_value = {'task': payload} + req = SendMessageRequest() + if history_length is not None: + req.configuration.history_length = history_length + response = await transport.send_message(req) + assert response.HasField('task') + task = response.task + assert [message.message_id for message in task.history] == expected_ids