fix: properly handle unset and zero history length (#717)

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

After changing to proto passing `.history_length` would not work anymore
due to the way how proto generated code works - optional values are
still translated to language defaults to avoid `None`s, while presence
should be checked via `HasField` - done in this PR.

Fixes #573

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

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
+
+    See Also:
+        https://a2a-protocol.org/latest/specification/#324-history-length-semantics
     """
-    # 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 or not config.HasField('history_length'):
+        return task
+
+    history_length = config.history_length
+
+    if history_length == 0:
+        if not task.history:
+            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)
+        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)
+        del task_copy.history[:-history_length]
+        return task_copy
+
     return task
 
 

tests/utils/test_task.py

@@ -5,8 +5,17 @@
 
 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,
@@ -213,5 +222,54 @@ def test_decode_page_token_fails(self):
         )
 
 
+class TestApplyHistoryLength(unittest.TestCase):
+    def setUp(self):
+        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):
+        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):
+        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):
+        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)
+
+
 if __name__ == '__main__':
     unittest.main()