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':
@@ -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(

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(

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,

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