fix: properly handle unset and zero history length

According to https://a2a-protocol.org/latest/specification/#324-history-length-semantics.
It changes behavior so the fix was postponed till 1.0.

Fixes #573

Author: ishymko

src/a2a/server/request_handlers/default_request_handler.py

@@ -127,8 +127,7 @@ async def on_get_task(
         if not task:
             raise ServerError(error=TaskNotFoundError())
 
-        # Apply historyLength parameter if specified
-        return apply_history_length(task, params.history_length)
+        return apply_history_length(task, params)
 
     async def on_list_tasks(
         self,
@@ -141,7 +140,7 @@ async def on_list_tasks(
             if not params.include_artifacts:
                 task.ClearField('artifacts')
 
-            updated_task = apply_history_length(task, params.history_length)
+            updated_task = apply_history_length(task, params)
             if updated_task is not task:
                 task.CopyFrom(updated_task)
 
@@ -380,9 +379,7 @@ async def push_notification_callback() -> None:
         if isinstance(result, Task):
             self._validate_task_id_match(task_id, result.id)
             if params.configuration:
-                result = apply_history_length(
-                    result, params.configuration.history_length
-                )
+                result = apply_history_length(result, params.configuration)
 
         await self._send_push_notification_if_needed(task_id, result_aggregator)
 

src/a2a/utils/task.py

@@ -4,6 +4,7 @@
 import uuid
 
 from base64 import b64decode, b64encode
+from typing import Literal, Protocol, runtime_checkable
 
 from a2a.types.a2a_pb2 import (
     Artifact,
@@ -81,27 +82,57 @@ def completed_task(
     )
 
 
-def apply_history_length(task: Task, history_length: int | None) -> Task:
+@runtime_checkable
+class HistoryLengthConfig(Protocol):
+    """Protocol for configuration arguments containing history_length field."""
+
+    history_length: int
+
+    def HasField(self, field_name: Literal['history_length']) -> bool:  # noqa: N802 -- Protobuf generated code
+        """Checks if a field is set.
+
+        This method name matches the generated Protobuf code.
+        """
+        ...
+
+
+def apply_history_length(
+    task: Task, config: HistoryLengthConfig | None
+) -> Task:
     """Applies history_length parameter on task and returns a new task object.
 
     Args:
         task: The original task object with complete history
-        history_length: History length configuration value
+        config: Configuration object containing 'history_length' field and HasField method.
 
     Returns:
         A new task object with limited history
     """
-    # Apply historyLength parameter if specified
-    if history_length is not None and history_length > 0 and task.history:
-        # Limit history to the most recent N messages
-        limited_history = list(task.history[-history_length:])
-        # Create a new task instance with limited history
+    if config is None:
+        return task
+
+    # See https://a2a-protocol.org/latest/specification/#324-history-length-semantics
+
+    if not config.HasField('history_length'):
+        return task
+
+    history_length = config.history_length
+
+    if history_length == 0:
+        task_copy = Task()
+        task_copy.CopyFrom(task)
+        task_copy.ClearField('history')
+        return task_copy
+
+    if history_length > 0 and task.history:
+        if len(task.history) <= history_length:
+            return task
+
         task_copy = Task()
         task_copy.CopyFrom(task)
-        # Clear and re-add history items
-        del task_copy.history[:]
-        task_copy.history.extend(limited_history)
+        del task_copy.history[:-history_length]
         return task_copy
+
     return task
 
 

tests/server/request_handlers/test_default_request_handler.py

@@ -1499,7 +1499,10 @@ async def exec_side_effect(_request, queue: EventQueue):
     # Allow producer to emit the next event
     allow_second_event.set()
 
-    received = await resub_gen.__anext__()
+    first_subscribe_event = await anext(resub_gen)
+    assert first_subscribe_event == task_for_resub
+
+    received = await anext(resub_gen)
     assert received == second_event
 
     # Finish producer to allow cleanup paths to complete
@@ -2706,7 +2709,7 @@ async def test_on_subscribe_to_task_in_terminal_state(terminal_state):
         async for _ in request_handler.on_subscribe_to_task(params, context):
             pass  # pragma: no cover
 
-    assert isinstance(exc_info.value.error, InvalidParamsError)
+    assert isinstance(exc_info.value.error, UnsupportedOperationError)
     assert exc_info.value.error.message
     assert (
         f'Task {task_id} is in terminal state: {terminal_state}'

tests/utils/test_task.py

@@ -5,8 +5,9 @@
 
 import pytest
 
-from a2a.types.a2a_pb2 import Artifact, Message, Part, Role, TaskState
+from a2a.types.a2a_pb2 import Artifact, Message, Part, Role, TaskState, GetTaskRequest, SendMessageConfiguration
 from a2a.utils.task import (
+    apply_history_length,
     completed_task,
     decode_page_token,
     encode_page_token,
@@ -215,3 +216,49 @@ def test_decode_page_token_fails(self):
 
 if __name__ == '__main__':
     unittest.main()
+
+
+
+class TestApplyHistoryLength(unittest.TestCase):
+    def setUp(self):
+        # Create a task with some history
+        self.history = [
+            Message(message_id=str(i), role=Role.ROLE_USER, parts=[Part(text=f'msg {i}')])
+            for i in range(5)
+        ]
+        artifacts = [Artifact(artifact_id='a1', parts=[Part(text='a')])]
+        self.task = completed_task(
+            task_id='t1',
+            context_id='c1',
+            artifacts=artifacts,
+            history=self.history
+        )
+
+    def test_none_config_returns_full_history(self):
+        # Test None (no limit) - config is None
+        result = apply_history_length(self.task, None)
+        self.assertEqual(len(result.history), 5)
+        self.assertEqual(result.history, self.history)
+
+    def test_unset_history_length_returns_full_history(self):
+        # Test unset (HasField returns False)
+        # Using GetTaskRequest as it has history_length field
+        result = apply_history_length(self.task, GetTaskRequest())
+        self.assertEqual(len(result.history), 5)
+        self.assertEqual(result.history, self.history)
+
+    def test_positive_history_length_truncates(self):
+        # Test > 0 (partial)
+        result = apply_history_length(self.task, GetTaskRequest(history_length=2))
+        self.assertEqual(len(result.history), 2)
+        self.assertEqual(result.history, self.history[-2:])
+
+    def test_large_history_length_returns_full_history(self):
+        result = apply_history_length(self.task, GetTaskRequest(history_length=10))
+        self.assertEqual(len(result.history), 5)
+        self.assertEqual(result.history, self.history)
+
+    def test_zero_history_length_returns_empty_history(self):
+        result = apply_history_length(self.task, SendMessageConfiguration(history_length=0))
+        self.assertEqual(len(result.history), 0)
+