fix(compat): apply v1.0 history_length semantics to v0.3 results

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':
@@ -169,11 +173,16 @@ async def send_message_streaming(
             response = await stream.read()
             if response == grpc.aio.EOF:  # type: ignore[attr-defined]
                 break
-            yield conversions.to_core_stream_response(
+            event = conversions.to_core_stream_response(
                 types_v03.SendStreamingMessageSuccessResponse(
                     result=proto_utils.FromProto.stream_response(response)
                 )
             )
+            if event.HasField('task'):
+                event.task.CopyFrom(
+                    apply_history_length(event.task, request.configuration)
+                )
+            yield event
 
     @_handle_grpc_stream_exception
     async def subscribe(
@@ -217,7 +226,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(

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':
@@ -140,6 +144,10 @@ async def send_message_streaming(
             dict(rpc_request.data),
             context,
         ):
+            if event.HasField('task'):
+                event.task.CopyFrom(
+                    apply_history_length(event.task, request.configuration)
+                )
             yield event
 
     async def get_task(
@@ -164,8 +172,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(

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':
@@ -143,6 +147,10 @@ async def send_message_streaming(
             context=context,
             json=MessageToDict(req_proto, preserving_proto_field_name=True),
         ):
+            if event.HasField('task'):
+                event.task.CopyFrom(
+                    apply_history_length(event.task, request.configuration)
+                )
             yield event
 
     async def get_task(
@@ -165,7 +173,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,

tests/compat/v0_3/test_grpc_transport.py

@@ -1,12 +1,14 @@
 from unittest.mock import AsyncMock, MagicMock
 
+import grpc
 import pytest
 
 from a2a.client.client import ClientCallContext
 from a2a.client.optionals import Channel
 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 +68,81 @@ 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
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('history_length', [None, 0, 1, 3, 5])
+@pytest.mark.parametrize('history_size', [0, 3])
+async def test_streaming_task_history_length(
+    history_length: int | None,
+    history_size: int,
+) -> 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(history_size):
+        task.history.add(message_id=f'm{i}', role=a2a_v0_3_pb2.Role.ROLE_USER)
+    task_event = a2a_v0_3_pb2.StreamResponse(task=task)
+    message_event = a2a_v0_3_pb2.StreamResponse(
+        msg=a2a_v0_3_pb2.Message(
+            message_id='reply', role=a2a_v0_3_pb2.Role.ROLE_AGENT
+        )
+    )
+    stream = MagicMock(
+        read=AsyncMock(
+            side_effect=[task_event, message_event, task_event, grpc.aio.EOF]
+        )
+    )
+    transport.stub = MagicMock(
+        SendStreamingMessage=MagicMock(return_value=stream)
+    )
+    request = SendMessageRequest(
+        message=Message(message_id='request', role=Role.ROLE_USER)
+    )
+    if history_length is not None:
+        request.configuration.history_length = history_length
+    events = [
+        event async for event in transport.send_message_streaming(request)
+    ]
+    expected = [f'm{i}' for i in range(history_size)]
+    if history_length is not None:
+        expected = expected[-history_length:] if history_length else []
+    assert len(events) == 3
+    for index in (0, 2):
+        assert [
+            item.message_id for item in events[index].task.history
+        ] == expected
+        assert events[index].task.id == task.id
+    assert events[1].message.message_id == 'reply'
+    assert len(task.history) == history_size

tests/compat/v0_3/test_jsonrpc_transport.py

@@ -1,3 +1,5 @@
+from collections.abc import AsyncIterator
+from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
@@ -565,3 +567,125 @@ 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
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('history_length', [None, 0, 1, 3, 5])
+@pytest.mark.parametrize('history_size', [0, 3])
+async def test_streaming_task_history_length(
+    transport: CompatJsonRpcTransport,
+    monkeypatch: pytest.MonkeyPatch,
+    history_length: int | None,
+    history_size: int,
+) -> None:
+    task = Task(id='task-1', context_id='ctx-1')
+    for i in range(history_size):
+        task.history.add(message_id=f'm{i}', role=Role.ROLE_USER)
+    message = StreamResponse(
+        message=Message(message_id='reply', role=Role.ROLE_AGENT)
+    )
+
+    async def stream(
+        *args: Any, **kwargs: Any
+    ) -> AsyncIterator[StreamResponse]:
+        yield StreamResponse(task=task)
+        yield message
+        yield StreamResponse(task=task)
+
+    monkeypatch.setattr(transport, '_send_stream_request', stream)
+    request = SendMessageRequest(
+        message=Message(message_id='request', role=Role.ROLE_USER)
+    )
+    if history_length is not None:
+        request.configuration.history_length = history_length
+    events = [
+        event async for event in transport.send_message_streaming(request)
+    ]
+    expected = [f'm{i}' for i in range(history_size)]
+    if history_length is not None:
+        expected = expected[-history_length:] if history_length else []
+    assert len(events) == 3
+    for index in (0, 2):
+        assert [
+            item.message_id for item in events[index].task.history
+        ] == expected
+        assert events[index].task.id == task.id
+    assert events[1] == message
+    assert len(task.history) == history_size