diff --git a/plugins/platforms/a2a/DESIGN.md b/plugins/platforms/a2a/DESIGN.md index 035e41f0fc47d..ffd750b5ae2bc 100644 --- a/plugins/platforms/a2a/DESIGN.md +++ b/plugins/platforms/a2a/DESIGN.md @@ -56,9 +56,11 @@ Peers resolved from `config.yaml` → `a2a_agents`, or a direct URL. `on_processing_complete` resolves failures/cancellations promptly. - **Task store:** every task (including terminal ones, bounded to the last 500) stays queryable via `tasks/get` / `tasks/list`, and `tasks/subscribe` - reattaches to a running task's stream via store watchers. A watchdog fails - orphaned tasks after 5 minutes (idempotent transitions — no double - counting in metrics). + reattaches to a running task's stream via store watchers. Observation + timeout is **not** terminal (A2A §3.1.3): `_await_reply` keeps waiting for + the live session instead of writing `TASK_STATE_FAILED` / `[agent did not + reply in time]`. A watchdog may fail **orphaned** tasks after 5 minutes, + but skips ids that still have a live pending waiter. - **input-required:** the platform hint tells the agent to start a reply with `[INPUT_REQUIRED]` when it needs clarification; the adapter maps that to `TASK_STATE_INPUT_REQUIRED` with the question in `status.message`. diff --git a/plugins/platforms/a2a/adapter.py b/plugins/platforms/a2a/adapter.py index 79842c88c6763..1c185834360ca 100644 --- a/plugins/platforms/a2a/adapter.py +++ b/plugins/platforms/a2a/adapter.py @@ -473,7 +473,9 @@ def _watchdog_loop(self) -> None: """Background thread that fails orphaned tasks (keeps them queryable).""" while not self._watchdog_stop.wait(_WATCHDOG_INTERVAL): try: - for tid in self.tasks.fail_orphans(_ORPHAN_TIMEOUT): + with self._pending_lock: + live = set(self._pending) + for tid in self.tasks.fail_orphans(_ORPHAN_TIMEOUT, skip_ids=live): logger.warning("A2A: orphaned task %s marked failed (timeout %ds)", tid, _ORPHAN_TIMEOUT) protocol.metrics.tasks_failed += 1 except Exception: @@ -928,27 +930,39 @@ def _finalize_task(self, pending: dict, state: str, reply: str) -> tuple[str, st return state, reply def _await_reply(self, pending: dict, keepalive=None) -> tuple[str, str]: - """Block until the task's future resolves (or times out). + """Block until the task's future resolves. - ``keepalive`` is an optional zero-arg callable invoked every - _SSE_KEEPALIVE seconds while waiting (used by the SSE paths); if it + An observation-window miss is not a terminal A2A state (§3.1.3): the + live session may still answer. ``keepalive`` is an optional zero-arg + callable invoked every ``_SSE_KEEPALIVE`` seconds (SSE paths); if it raises, the client is gone and we stop waiting. """ fut: Future = pending["future"] deadline = pending["started"] + _reply_timeout() + logged_observation_timeout = False while True: + remaining = deadline - time.time() + if keepalive: + wait = _SSE_KEEPALIVE + elif remaining > 0: + wait = remaining + else: + wait = _SSE_KEEPALIVE try: - return fut.result(timeout=_SSE_KEEPALIVE if keepalive else max(0.0, deadline - time.time())) + return fut.result(timeout=wait) except FuturesTimeout: - if time.time() >= deadline: - return (protocol.STATE_FAILED, "[agent did not reply in time]") + if remaining <= 0 and not logged_observation_timeout: + logger.info( + "A2A: observation window elapsed; task stays non-terminal until the session replies" + ) + logged_observation_timeout = True if keepalive: try: keepalive() except Exception: return (protocol.STATE_FAILED, "[client disconnected]") except Exception: - return (protocol.STATE_FAILED, "[agent did not reply in time]") + return (protocol.STATE_FAILED, "[agent processing failed]") def _rpc_message_send(self, req_id: Any, params: dict, peer: str, agent: Optional[dict] = None, v1_response: bool = False) -> dict: terminal, pending = self._prepare_task(params, peer, agent=agent) diff --git a/plugins/platforms/a2a/protocol.py b/plugins/platforms/a2a/protocol.py index f1522fccb1d59..381c7a54508dc 100644 --- a/plugins/platforms/a2a/protocol.py +++ b/plugins/platforms/a2a/protocol.py @@ -30,7 +30,7 @@ from concurrent.futures import Future from datetime import datetime, timezone from pathlib import Path -from typing import Any, Optional +from typing import Any, Optional, Collection PROTOCOL_VERSION = "1.0" @@ -748,12 +748,18 @@ def list( return page, next_offset, total return page, next_offset - def fail_orphans(self, timeout_seconds: int = 300) -> list[str]: + def fail_orphans( + self, + timeout_seconds: int = 300, + skip_ids: Optional[Collection[str]] = None, + ) -> list[str]: + skip = set(skip_ids or ()) with self._lock: now = time.time() stale = [ tid for tid, rec in self._tasks.items() if rec["state"] not in TERMINAL_STATES + and tid not in skip and now - rec["created_at"] > timeout_seconds ] failed = [] diff --git a/tests/plugins/test_a2a_plugin.py b/tests/plugins/test_a2a_plugin.py index 346284b27796d..76866c5b752af 100644 --- a/tests/plugins/test_a2a_plugin.py +++ b/tests/plugins/test_a2a_plugin.py @@ -16,6 +16,7 @@ import os import socket import threading +import time import urllib.error import urllib.request from concurrent.futures import Future @@ -1618,3 +1619,40 @@ def test_forward_to_profile_first_contact_creates_then_resumes_fake_hermes(self, title = con.execute("SELECT title FROM sessions WHERE id='sess-1'").fetchone()[0] con.close() assert title == "a2a-dev-ctx-unsafe-value" + + +# -------------------------------------------------------------------------- +# Observation timeout is not a terminal A2A state (§3.1.3 / #1412) +# -------------------------------------------------------------------------- + +class TestObservationTimeoutNotFailed: + def test_await_reply_keeps_waiting_after_deadline_then_completes(self, monkeypatch): + """A live session that answers late must not be TASK_STATE_FAILED.""" + monkeypatch.setenv("A2A_REPLY_TIMEOUT", "0.05") + adapter = _bare_adapter() + fut = Future() + pending = {"future": fut, "started": time.time() - 1.0} + + def complete_late(): + time.sleep(0.2) + fut.set_result((protocol.STATE_COMPLETED, "late answer")) + + thread = threading.Thread(target=complete_late) + thread.start() + state, reply = adapter._await_reply(pending) + thread.join(timeout=2) + assert (state, reply) == (protocol.STATE_COMPLETED, "late answer") + assert state != protocol.STATE_FAILED + + def test_watchdog_skips_tasks_with_live_waiters(self): + adapter = _bare_adapter() + adapter.tasks.create("task-live", "ctx", "peer") + adapter.tasks._tasks["task-live"]["created_at"] = time.time() - 600 + adapter._add_pending("task-live", "ctx") + with adapter._pending_lock: + live = set(adapter._pending) + failed = adapter.tasks.fail_orphans(timeout_seconds=300, skip_ids=live) + rec = adapter.tasks.get("task-live") + assert failed == [] + assert rec is not None + assert rec["state"] != protocol.STATE_FAILED