fix(a2a): do not fail tasks on observation timeout

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`.

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)

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 = []

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