diff --git a/src/a2a_handler/cli/task.py b/src/a2a_handler/cli/task.py index 1bc5019..5a5443a 100644 --- a/src/a2a_handler/cli/task.py +++ b/src/a2a_handler/cli/task.py @@ -13,6 +13,8 @@ reject_control_chars, reject_unknown_keys, validate_agent_url, + validate_history_length, + validate_page_size, validate_resource_id, validate_webhook_url, ) @@ -20,6 +22,7 @@ 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,14 +209,14 @@ 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() @@ -226,16 +224,24 @@ async def do_list() -> None: 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") diff --git a/src/a2a_handler/common/input_validation.py b/src/a2a_handler/common/input_validation.py index cae441b..6c36512 100644 --- a/src/a2a_handler/common/input_validation.py +++ b/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") diff --git a/src/a2a_handler/mcp/server.py b/src/a2a_handler/mcp/server.py index 3889207..08a55ae 100644 --- a/src/a2a_handler/mcp/server.py +++ b/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() diff --git a/src/a2a_handler/service.py b/src/a2a_handler/service.py index 3eb95a4..f7630e3 100644 --- a/src/a2a_handler/service.py +++ b/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. diff --git a/tests/cli/test_task.py b/tests/cli/test_task.py index 9e18f56..d793201 100644 --- a/tests/cli/test_task.py +++ b/tests/cli/test_task.py @@ -1,5 +1,6 @@ """Tests for task CLI commands.""" +import json import os import pytest @@ -16,8 +17,9 @@ TaskStatus, ) +from a2a_handler.cli import cli from a2a_handler.cli.task import task -from a2a_handler.service import StreamEvent +from a2a_handler.service import StreamEvent, TaskListing from tests.factories import make_push_config @@ -400,7 +402,7 @@ def test_task_id_from_json_params_alone(self, runner): class TestTaskList: """Tests for task list command.""" - def _invoke_list(self, runner, args, tasks): + def _invoke_list(self, runner, args, tasks, truncated=False): with ( patch("a2a_handler.cli.task.build_http_client") as mock_client, patch("a2a_handler.cli.task.A2AService") as mock_service_cls, @@ -411,7 +413,9 @@ def _invoke_list(self, runner, args, tasks): mock_client.return_value = mock_http mock_service = AsyncMock() - mock_service.list_all_tasks.return_value = tasks + mock_service.list_all_tasks.return_value = TaskListing( + tasks=tasks, truncated=truncated + ) mock_service_cls.return_value = mock_service result = runner.invoke(task, ["list", *args]) @@ -501,6 +505,52 @@ def test_task_list_rejects_non_positive_page_size(self, runner): assert result.exit_code != 0 assert "at least 1" in result.output + def test_task_list_rejects_negative_history_length(self, runner): + """A negative history length fails before any network call.""" + result = runner.invoke( + task, + ["list", "--url", "http://localhost:8000", "--history-length", "-1"], + ) + assert result.exit_code != 0 + assert "must not be negative" in result.output + + def test_task_list_warns_when_listing_is_truncated(self, runner): + """A listing cut short says so instead of looking complete.""" + tasks = [_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")] + result, _ = self._invoke_list( + runner, ["--url", "http://localhost:8000"], tasks, truncated=True + ) + + assert result.exit_code == 0 + assert "task-1" in result.output + assert "may not be all of them" in result.output + + def test_task_list_reports_truncation_in_json(self, runner): + """The JSON envelope carries the truncation flag for machines.""" + tasks = [_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")] + with ( + patch("a2a_handler.cli.task.build_http_client") as mock_client, + patch("a2a_handler.cli.task.A2AService") as mock_service_cls, + ): + mock_http = AsyncMock() + mock_http.__aenter__.return_value = mock_http + mock_http.__aexit__.return_value = None + mock_client.return_value = mock_http + mock_service = AsyncMock() + mock_service.list_all_tasks.return_value = TaskListing( + tasks=tasks, truncated=True + ) + mock_service_cls.return_value = mock_service + + result = runner.invoke( + cli, ["--output", "json", "task", "list", "--url", "http://x.test"] + ) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["truncated"] is True + assert payload["count"] == 1 + def test_task_list_rejects_invalid_context_id(self, runner): """A malformed context ID fails before any network call.""" result = runner.invoke( diff --git a/tests/core/test_service.py b/tests/core/test_service.py index 5e1bb86..22b3ba7 100644 --- a/tests/core/test_service.py +++ b/tests/core/test_service.py @@ -688,9 +688,10 @@ async def test_list_all_tasks_follows_pages(self): fake_client = _FakeListTasksClient(pages) service = self._service_with(fake_client) - tasks = await service.list_all_tasks(page_size=1) + listing = await service.list_all_tasks(page_size=1) - assert [item.id for item in tasks] == ["task-1", "task-2"] + assert [item.id for item in listing.tasks] == ["task-1", "task-2"] + assert listing.truncated is False assert len(fake_client.requests) == 2 assert fake_client.requests[1].page_token == "page-2" @@ -702,12 +703,13 @@ async def test_list_all_tasks_stops_on_repeated_token_without_duplicates(self): fake_client = _FakeListTasksClient([page, page]) service = self._service_with(fake_client) - tasks = await service.list_all_tasks() + listing = await service.list_all_tasks() # First call, then one follow-up with the token; the replayed page # stops the loop and its tasks are deduplicated, not double-counted. assert len(fake_client.requests) == 2 - assert [item.id for item in tasks] == ["task-1"] + assert [item.id for item in listing.tasks] == ["task-1"] + assert listing.truncated is True async def test_list_all_tasks_is_capped_against_fresh_token_loops( self, monkeypatch @@ -725,10 +727,11 @@ async def test_list_all_tasks_is_capped_against_fresh_token_loops( fake_client = _FakeListTasksClient(pages) service = self._service_with(fake_client) - tasks = await service.list_all_tasks() + listing = await service.list_all_tasks() assert len(fake_client.requests) == 3 - assert [item.id for item in tasks] == ["task-0", "task-1", "task-2"] + assert [item.id for item in listing.tasks] == ["task-0", "task-1", "task-2"] + assert listing.truncated is True async def test_list_tasks_rejects_non_positive_page_size(self): service = self._service_with(_FakeListTasksClient([])) @@ -739,6 +742,72 @@ async def test_list_tasks_rejects_non_positive_page_size(self): with pytest.raises(InputValidationError): await service.list_tasks(page_size=-5) + async def test_list_tasks_rejects_page_size_above_the_spec_maximum(self): + # The spec caps a page at 100; catching it locally saves a round trip + # that could only come back as an InvalidParams error. + service = self._service_with(_FakeListTasksClient([])) + with pytest.raises(InputValidationError) as exc_info: + await service.list_tasks(page_size=101) + assert isinstance(exc_info.value, InputValidationError) + assert exc_info.value.code == "invalid_page_size" + assert "at most 100" in exc_info.value.message + + async def test_list_tasks_accepts_the_maximum_page_size(self): + fake_client = _FakeListTasksClient([ListTasksResponse(tasks=[])]) + service = self._service_with(fake_client) + + await service.list_tasks(page_size=100) + + assert fake_client.requests[0].page_size == 100 + + async def test_list_tasks_rejects_negative_history_length(self): + service = self._service_with(_FakeListTasksClient([])) + with pytest.raises(InputValidationError) as exc_info: + await service.list_tasks(history_length=-1) + assert isinstance(exc_info.value, InputValidationError) + assert exc_info.value.code == "invalid_history_length" + + async def test_list_tasks_leaves_unset_paging_fields_off_the_wire(self): + # page_size and history_length carry explicit presence, so omitting + # them must send nothing at all rather than a zero the server would + # have to interpret. + page = ListTasksResponse(tasks=[]) + fake_client = _FakeListTasksClient([page]) + service = self._service_with(fake_client) + + await service.list_tasks() + + request = fake_client.requests[0] + assert request.HasField("page_size") is False + assert request.HasField("history_length") is False + + async def test_list_all_tasks_stops_on_token_cycle_that_never_repeats(self): + # A server alternating between two tokens never hands back the token + # just sent, so the repeated-token guard alone would loop until the + # page cap. No page contributes a new task, which is what stops it. + pages = [ + ListTasksResponse( + tasks=[_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")], + next_page_token="token-a", + ), + ListTasksResponse( + tasks=[_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")], + next_page_token="token-b", + ), + ListTasksResponse( + tasks=[_make_task(TaskState.TASK_STATE_COMPLETED, task_id="task-1")], + next_page_token="token-a", + ), + ] + fake_client = _FakeListTasksClient(pages) + service = self._service_with(fake_client) + + listing = await service.list_all_tasks() + + assert len(fake_client.requests) == 2 + assert [item.id for item in listing.tasks] == ["task-1"] + assert listing.truncated is True + class _FakeStreamingClient: def __init__(self, events): diff --git a/tests/mcp/test_server.py b/tests/mcp/test_server.py index 19bdeca..ccfb458 100644 --- a/tests/mcp/test_server.py +++ b/tests/mcp/test_server.py @@ -12,6 +12,7 @@ ) from a2a_handler.mcp.server import create_mcp_server +from a2a_handler.service import TaskListing from a2a_handler.session import AgentSession from a2a_handler.validation import ValidationResult, ValidationSource from tests.factories import make_agent_card, make_push_config @@ -345,7 +346,7 @@ async def test_list_tasks_success() -> None: ] mock_service = AsyncMock() - mock_service.list_all_tasks.return_value = tasks + mock_service.list_all_tasks.return_value = TaskListing(tasks=tasks) with ( patch("a2a_handler.mcp.server._build_http_client", return_value=_mock_http()), @@ -366,6 +367,41 @@ async def test_list_tasks_success() -> None: history_length=None, include_artifacts=False, ) + assert resp["truncated"] is False + + +@pytest.mark.asyncio +async def test_list_tasks_reports_truncation() -> None: + """A listing cut short is flagged rather than passed off as complete.""" + server = create_mcp_server() + fn = _tool_fn(server, "list_tasks") + + mock_service = AsyncMock() + mock_service.list_all_tasks.return_value = TaskListing( + tasks=[_make_task(task_id="task-1", state=TaskState.TASK_STATE_COMPLETED)], + truncated=True, + ) + + with ( + patch("a2a_handler.mcp.server._build_http_client", return_value=_mock_http()), + patch("a2a_handler.mcp.server.A2AService", return_value=mock_service), + ): + resp = await fn(agent_url="http://localhost:8000") + + assert resp["truncated"] is True + assert resp["count"] == 1 + + +@pytest.mark.asyncio +async def test_list_tasks_rejects_negative_history_length() -> None: + """A negative history length fails before any network call.""" + server = create_mcp_server() + fn = _tool_fn(server, "list_tasks") + + with pytest.raises(Exception) as exc_info: + await fn(agent_url="http://localhost:8000", history_length=-1) + + assert "must not be negative" in str(exc_info.value) @pytest.mark.asyncio