fix: harden task listing against partial and out-of-range results

src/a2a_handler/cli/task.py

@@ -13,13 +13,16 @@
     reject_control_chars,
     reject_unknown_keys,
     validate_agent_url,
+    validate_history_length,
+    validate_page_size,
     validate_resource_id,
     validate_webhook_url,
 )
 from a2a.types import Task
 from a2a_handler.service import (
     A2AService,
     TASK_STATE_LABELS,
+    TaskListing,
     protocol_dump,
     push_config_dump,
     response_state,
@@ -192,13 +195,8 @@ def task_list(
             validate_resource_id(context_id, "context_id")
         if status:
             status_value = task_state_from_label(status)
-        if page_size is not None and page_size < 1:
-            raise InputValidationError(
-                code="invalid_page_size",
-                message="--page-size must be at least 1",
-                suggestion="Omit --page-size to use the default of 50",
-                details={"field": "page_size"},
-            )
+        validate_page_size(page_size, "--page-size")
+        validate_history_length(history_length, "--history-length")
     except InputValidationError as error:
         handle_validation_error(error, output)
         raise click.Abort() from error
@@ -211,31 +209,39 @@ async def do_list() -> None:
         try:
             async with build_http_client(credentials=credentials) as http_client:
                 service = A2AService(http_client, resolved_url, credentials=credentials)
-                tasks = await service.list_all_tasks(
+                listing = await service.list_all_tasks(
                     context_id=context_id,
                     status=status_value,
                     page_size=page_size,
                     history_length=history_length,
                     include_artifacts=include_artifacts,
                 )
-                _format_task_list(tasks, output)
+                _format_task_list(listing, output)
         except Exception as e:
             handle_client_error(e, resolved_url, output)
             raise click.Abort()
 
     asyncio.run(do_list())
 
 
-def _format_task_list(tasks: list[Task], output: Output) -> None:
-    """Format a task listing for structured or human-readable output."""
+def _format_task_list(listing: TaskListing, output: Output) -> None:
+    """Format a task listing for structured or human-readable output.
+
+    A listing cut short by a pagination defense is reported in every format,
+    so no consumer mistakes a partial result for the whole set.
+    """
+    tasks = listing.tasks
     if output.output_format == "ndjson":
         for task_item in tasks:
             output.json(protocol_dump(task_item))
+        if listing.truncated:
+            output.json({"type": "truncated", "count": len(tasks)})
         return
     if output.output_format == "json":
         output.json(
             {
                 "count": len(tasks),
+                "truncated": listing.truncated,
                 "tasks": [protocol_dump(task_item) for task_item in tasks],
             }
         )
@@ -249,6 +255,11 @@ def _format_task_list(tasks: list[Task], output: Output) -> None:
         if task_item.context_id:
             line += f"  context={task_item.context_id}"
         output.text(line)
+    if listing.truncated:
+        output.text(
+            f"Warning: listing stopped early; showing {len(tasks)} task(s), "
+            "which may not be all of them."
+        )
 
 
 @task.command("cancel")

src/a2a_handler/common/input_validation.py

@@ -61,6 +61,56 @@ def validate_resource_id(value: str, field_name: str) -> str:
     return value
 
 
+# The A2A spec bounds a ListTasks page size to [1, 100]; a server rejects
+# anything outside it, so catch it locally instead of paying a round trip.
+# https://a2a-protocol.org/latest/specification/#314-list-tasks
+MAX_LIST_TASKS_PAGE_SIZE = 100
+
+
+def validate_page_size(page_size: int | None, label: str = "page_size") -> None:
+    """Reject an out-of-range page size before it reaches the wire.
+
+    ``label`` names the input in the message so a CLI caller can show the
+    flag the user typed; ``details.field`` stays the machine name either way.
+    """
+    if page_size is None:
+        return
+    if page_size < 1:
+        raise InputValidationError(
+            code="invalid_page_size",
+            message=f"{label} must be at least 1",
+            suggestion=f"Omit {label} to use the server's default page size",
+            details={"field": "page_size"},
+        )
+    if page_size > MAX_LIST_TASKS_PAGE_SIZE:
+        raise InputValidationError(
+            code="invalid_page_size",
+            message=f"{label} must be at most {MAX_LIST_TASKS_PAGE_SIZE}",
+            suggestion=(
+                f"Use a page size up to {MAX_LIST_TASKS_PAGE_SIZE}; every page "
+                "is fetched regardless, so a smaller one loses nothing"
+            ),
+            details={"field": "page_size", "max": MAX_LIST_TASKS_PAGE_SIZE},
+        )
+
+
+def validate_history_length(
+    history_length: int | None, label: str = "history_length"
+) -> None:
+    """Reject a negative history length before it reaches the wire.
+
+    ``label`` names the input in the message; ``details.field`` stays the
+    machine name, as in :func:`validate_page_size`.
+    """
+    if history_length is not None and history_length < 0:
+        raise InputValidationError(
+            code="invalid_history_length",
+            message=f"{label} must not be negative",
+            suggestion=f"Omit {label} to use the server's default",
+            details={"field": "history_length"},
+        )
+
+
 def validate_webhook_url(url: str) -> str:
     """Validate webhook callback URLs used for push notifications."""
     reject_control_chars(url, "webhook_url")

src/a2a_handler/mcp/server.py

@@ -20,6 +20,8 @@
     reject_control_chars,
     validate_agent_url,
     validate_header_name,
+    validate_history_length,
+    validate_page_size,
     validate_resource_id,
     validate_webhook_url,
 )
@@ -423,6 +425,8 @@ async def list_tasks(
         Returns:
             A dictionary containing:
             - count: Number of tasks returned
+            - truncated: True when a pagination defense stopped the crawl
+              early, meaning the listing is incomplete
             - tasks: The tasks in A2A wire format
         """
         logger.info("Listing tasks at %s", agent_url)
@@ -433,13 +437,8 @@ async def list_tasks(
                 validate_resource_id(context_id, "context_id")
             if status:
                 status_value = task_state_from_label(status)
-            if page_size is not None and page_size < 1:
-                raise InputValidationError(
-                    code="invalid_page_size",
-                    message="page_size must be at least 1",
-                    suggestion="Omit page_size to use the default of 50",
-                    details={"field": "page_size"},
-                )
+            validate_page_size(page_size)
+            validate_history_length(history_length)
             if bearer_token:
                 reject_control_chars(bearer_token, "bearer_token")
             if api_key:
@@ -459,7 +458,7 @@ async def list_tasks(
 
         async with _build_http_client(credentials=credentials) as http_client:
             service = A2AService(http_client, agent_url, credentials=credentials)
-            tasks = await service.list_all_tasks(
+            listing = await service.list_all_tasks(
                 context_id=context_id,
                 status=status_value,
                 page_size=page_size,
@@ -468,8 +467,9 @@ async def list_tasks(
             )
 
             return {
-                "count": len(tasks),
-                "tasks": [protocol_dump(task) for task in tasks],
+                "count": len(listing.tasks),
+                "truncated": listing.truncated,
+                "tasks": [protocol_dump(task) for task in listing.tasks],
             }
 
     @mcp.tool()

src/a2a_handler/service.py

@@ -54,6 +54,8 @@
     InputValidationError,
     reject_control_chars,
     validate_agent_url,
+    validate_history_length,
+    validate_page_size,
     validate_resource_id,
     validate_webhook_url,
 )
@@ -64,14 +66,24 @@
 # path locally so Handler can still fall back to it for older servers.
 LEGACY_AGENT_CARD_WELL_KNOWN_PATH = "/.well-known/agent.json"
 
-# Tasks fetched per ListTasks request when the caller does not choose a page
-# size. Servers reject a zero (i.e. unset) page size outright.
-DEFAULT_LIST_TASKS_PAGE_SIZE = 50
-
 # Hard bound on pagination so a server minting a fresh continuation token on
 # every response cannot spin list_all_tasks forever.
 MAX_LIST_TASKS_PAGES = 1000
 
+
+@dataclass(frozen=True, slots=True)
+class TaskListing:
+    """Every task a full listing produced, plus whether it was cut short.
+
+    ``truncated`` is True when a pagination defense (page cap or a
+    non-progressing server) stopped the crawl before the server ran out of
+    continuation tokens; the tasks list is then incomplete.
+    """
+
+    tasks: list[Task]
+    truncated: bool = False
+
+
 TERMINAL_TASK_STATES = {
     TaskState.TASK_STATE_COMPLETED,
     TaskState.TASK_STATE_CANCELED,
@@ -867,6 +879,8 @@ async def get_task(
 
         Returns the raw A2A Task object.
         """
+        validate_history_length(history_length)
+
         client = await self._get_or_create_client()
 
         request = GetTaskRequest(id=task_id)
@@ -902,9 +916,10 @@ async def list_tasks(
             context_id: Only return tasks in this context
             status: Only return tasks in this ``TaskState``
             page_size: Maximum tasks per page (server may return fewer);
-                defaults to ``DEFAULT_LIST_TASKS_PAGE_SIZE``
+                left unset the server chooses its own default
             page_token: Continuation token from a previous page's
-                ``next_page_token``
+                ``next_page_token``. Opaque server data; it is sent back
+                verbatim, not validated as user input.
             history_length: Number of history messages to include per task
             include_artifacts: Whether to include task artifacts
 
@@ -913,26 +928,21 @@ async def list_tasks(
         """
         if context_id:
             validate_resource_id(context_id, "context_id")
-        if page_token:
-            reject_control_chars(page_token, "page_token")
-        if page_size is not None and page_size < 1:
-            raise InputValidationError(
-                code="invalid_page_size",
-                message="page_size must be at least 1",
-                suggestion="Omit page_size to use the default",
-                details={"field": "page_size"},
-            )
+        validate_page_size(page_size)
+        validate_history_length(history_length)
 
         client = await self._get_or_create_client()
 
-        # Servers commonly reject a zero page size and some treat an unset
-        # field as zero, so always send an explicit one.
+        # page_size and history_length carry explicit presence, so they are
+        # only set when the caller chose a value; unset means the server's
+        # default applies.
         request = ListTasksRequest(
             context_id=context_id or "",
-            page_size=page_size or DEFAULT_LIST_TASKS_PAGE_SIZE,
             page_token=page_token or "",
             include_artifacts=include_artifacts,
         )
+        if page_size is not None:
+            request.page_size = page_size
         if status is not None:
             request.status = cast("TaskState", status)
         if history_length is not None:
@@ -954,19 +964,23 @@ async def list_all_tasks(
         page_size: int | None = None,
         history_length: int | None = None,
         include_artifacts: bool = False,
-    ) -> list[Task]:
+    ) -> TaskListing:
         """List tasks across every page, following continuation tokens.
 
-        Defenses against misbehaving servers: tasks are deduplicated by ID (a
-        replayed page adds nothing), a token identical to the one just sent
-        stops the loop (requesting again could only repeat), and pagination
-        is capped at ``MAX_LIST_TASKS_PAGES`` so a server minting fresh
-        tokens forever cannot spin the client.
+        Defenses against misbehaving servers: tasks are deduplicated by ID,
+        the loop stops as soon as a page contributes nothing new while still
+        offering a continuation token (which subsumes replayed pages and
+        token cycles of any length), and pagination is capped at
+        ``MAX_LIST_TASKS_PAGES`` so a server minting fresh tokens forever
+        cannot spin the client. A listing cut short by either defense is
+        marked ``truncated`` so consumers are not handed silently
+        incomplete data.
         """
         tasks: list[Task] = []
         seen_task_ids: set[str] = set()
         page_token: str | None = None
         pages_fetched = 0
+        truncated = False
 
         while True:
             response = await self.list_tasks(
@@ -978,32 +992,44 @@ async def list_all_tasks(
                 include_artifacts=include_artifacts,
             )
             pages_fetched += 1
+            new_tasks = 0
             for task in response.tasks:
                 if task.id and task.id in seen_task_ids:
                     continue
                 if task.id:
                     seen_task_ids.add(task.id)
                 tasks.append(task)
+                new_tasks += 1
 
             next_token = response.next_page_token
             if not next_token:
                 break
+            if response.tasks and new_tasks == 0:
+                logger.warning(
+                    "Server offered page token %r but the page held nothing "
+                    "new; stopping pagination",
+                    next_token,
+                )
+                truncated = True
+                break
             if next_token == (page_token or ""):
                 logger.warning(
                     "Server repeated page token %r; stopping pagination",
                     next_token,
                 )
+                truncated = True
                 break
             if pages_fetched >= MAX_LIST_TASKS_PAGES:
                 logger.warning(
-                    "Stopping after %d pages; the task listing may be incomplete",
+                    "Stopping after %d pages; the task listing is incomplete",
                     pages_fetched,
                 )
+                truncated = True
                 break
             page_token = next_token
 
         logger.info("Listed %d task(s) across %d page(s)", len(tasks), pages_fetched)
-        return tasks
+        return TaskListing(tasks=tasks, truncated=truncated)
 
     async def resubscribe(self, task_id: str) -> AsyncIterator[StreamEvent]:
         """Resubscribe to a task's event stream.