fix(a2a): make list_tasks sorting and filtering safe across naive and aware datetimes

lib/crewai/src/crewai/a2a/utils/task.py

@@ -5,7 +5,7 @@
 import asyncio
 import base64
 from collections.abc import Callable, Coroutine
-from datetime import datetime
+from datetime import datetime, timezone
 from functools import wraps
 import json
 import logging
@@ -503,6 +503,21 @@ async def cancel(
     return None
 
 
+def _parse_status_timestamp(value: str | None) -> datetime:
+    """Parse a task status timestamp as an aware datetime for safe comparisons.
+
+    Tasks without a timestamp sort as the oldest instant; naive timestamps
+    are assumed to be UTC. Mixing naive and aware values would otherwise
+    raise TypeError during sorting or filtering.
+    """
+    if value is None:
+        return datetime.min.replace(tzinfo=timezone.utc)
+    ts = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if ts.tzinfo is None:
+        ts = ts.replace(tzinfo=timezone.utc)
+    return ts
+
+
 def list_tasks(
     tasks: list[A2ATask],
     context_id: str | None = None,
@@ -536,24 +551,21 @@ def list_tasks(
         - total_count: Total number of tasks matching filters (before pagination).
     """
     filtered: list[A2ATask] = []
+    if status_timestamp_after is not None and status_timestamp_after.tzinfo is None:
+        status_timestamp_after = status_timestamp_after.replace(tzinfo=timezone.utc)
     for task in tasks:
         if context_id and task.context_id != context_id:
             continue
         if status and task.status.state != status:
             continue
         if status_timestamp_after and task.status.timestamp:
-            ts = datetime.fromisoformat(task.status.timestamp.replace("Z", "+00:00"))
-            if ts <= status_timestamp_after:
+            if _parse_status_timestamp(task.status.timestamp) <= status_timestamp_after:
                 continue
         filtered.append(task)
 
-    def get_timestamp(t: A2ATask) -> datetime:
-        """Extract timestamp from task status for sorting."""
-        if t.status.timestamp is None:
-            return datetime.min
-        return datetime.fromisoformat(t.status.timestamp.replace("Z", "+00:00"))
-
-    filtered.sort(key=get_timestamp, reverse=True)
+    filtered.sort(
+        key=lambda t: _parse_status_timestamp(t.status.timestamp), reverse=True
+    )
     total = len(filtered)
 
     start = 0

lib/crewai/tests/a2a/utils/test_task.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+from datetime import datetime, timedelta, timezone
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -12,7 +13,7 @@
 from a2a.server.events import EventQueue
 from a2a.types import Message, Task as A2ATask, TaskState, TaskStatus
 
-from crewai.a2a.utils.task import cancel, cancellable, execute
+from crewai.a2a.utils.task import cancel, cancellable, execute, list_tasks
 
 
 @pytest.fixture
@@ -372,4 +373,61 @@ async def slow_task(**kwargs: Any) -> str:
             await cancel(mock_context, mock_event_queue)
 
             with pytest.raises(asyncio.CancelledError):
-                await execute_task
+                await execute_task
+
+
+def _completed_task(task_id: str, context_id: str, timestamp: str | None) -> A2ATask:
+    return A2ATask(
+        id=task_id,
+        context_id=context_id,
+        status=TaskStatus(state=TaskState.completed, timestamp=timestamp),
+    )
+
+
+def test_list_tasks_sorts_aware_and_missing_timestamps():
+    """Tasks without a timestamp must not crash sorting against aware ones."""
+    now = datetime.now(timezone.utc)
+    tasks = [
+        _completed_task("missing-ts", "ctx-1", None),
+        _completed_task("older", "ctx-1", (now - timedelta(hours=2)).isoformat()),
+        _completed_task("newer", "ctx-1", now.isoformat()),
+    ]
+
+    result, _next_token, total = list_tasks(tasks=tasks, context_id="ctx-1")
+
+    assert total == 3
+    assert [task.id for task in result] == ["newer", "older", "missing-ts"]
+
+
+def test_list_tasks_accepts_naive_status_timestamp_after():
+    now = datetime.now(timezone.utc)
+    tasks = [
+        _completed_task("old", "ctx-1", (now - timedelta(hours=2)).isoformat()),
+        _completed_task("new", "ctx-1", now.isoformat()),
+    ]
+
+    result, _next_token, total = list_tasks(
+        tasks=tasks,
+        context_id="ctx-1",
+        status_timestamp_after=datetime(2000, 1, 1),
+    )
+
+    assert total == 2
+    assert [task.id for task in result] == ["new", "old"]
+
+
+def test_list_tasks_filters_out_tasks_older_than_cutoff():
+    now = datetime.now(timezone.utc)
+    tasks = [
+        _completed_task("old", "ctx-1", (now - timedelta(hours=2)).isoformat()),
+        _completed_task("new", "ctx-1", now.isoformat()),
+    ]
+
+    result, _next_token, total = list_tasks(
+        tasks=tasks,
+        context_id="ctx-1",
+        status_timestamp_after=now - timedelta(hours=1),
+    )
+
+    assert total == 1
+    assert result[0].id == "new"