Merge branch '1.0-dev' into ishymko/678-push-notifications

Author: ishymko

src/a2a/server/request_handlers/default_request_handler.py

@@ -128,8 +128,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,
@@ -142,7 +141,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)
 
@@ -381,9 +380,7 @@ async def push_notification_callback(event: Event) -> 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)
 
         return result
 
@@ -552,11 +549,15 @@ async def on_subscribe_to_task(
 
         if task.status.state in TERMINAL_TASK_STATES:
             raise ServerError(
-                error=InvalidParamsError(
+                error=UnsupportedOperationError(
                     message=f'Task {task.id} is in terminal state: {task.status.state}'
                 )
             )
 
+        # The operation MUST return a Task object as the first event in the stream
+        # https://a2a-protocol.org/latest/specification/#316-subscribe-to-task
+        yield task
+
         task_manager = TaskManager(
             task_id=task.id,
             context_id=task.context_id,

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/server/request_handlers/test_default_request_handler.py

@@ -1506,7 +1506,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
@@ -2713,7 +2716,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/server/request_handlers/test_jsonrpc_handler.py

@@ -703,7 +703,9 @@ async def streaming_coro():
             collected_events: list[Any] = []
             async for event in response:
                 collected_events.append(event)
-            assert len(collected_events) == len(events)
+            assert (
+                len(collected_events) == len(events) + 1
+            )  # First event is task itself
             assert mock_task.history is not None and len(mock_task.history) == 0
 
     async def test_on_subscribe_no_existing_task_error(self) -> None:

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()