diff --git a/plugins/platforms/a2a/DESIGN.md b/plugins/platforms/a2a/DESIGN.md
index 035e41f0fc47d..cfb0edc5fe7f0 100644
--- a/plugins/platforms/a2a/DESIGN.md
+++ b/plugins/platforms/a2a/DESIGN.md
@@ -19,10 +19,13 @@ must not touch core files.** A2A now lives entirely under
 ### Outbound — client tools (`a2a` toolset)
 - `a2a_discover(url)` — fetch + summarize a peer's Agent Card (v1.0
   `supportedInterfaces` aware, tolerates 0.3 cards).
-- `a2a_call(agent, message, context_id?)` — send a JSON-RPC `message/send`
+- `a2a_call(agent, message, context_id?, return_immediately?)` — send a JSON-RPC `message/send`
   task to a peer, return the reply. Multi-turn via `context_id` (carried
   inside the Message per v1.0). Surfaces `TASK_STATE_INPUT_REQUIRED` so the
-  model knows to answer and continue the context.
+  model knows to answer and continue the context. `return_immediately` sets
+  `configuration.returnImmediately` so the peer returns a working task id
+  instead of holding the HTTP call until the job finishes.
+- `a2a_get_task(agent, task_id)` — JSON-RPC `GetTask` poll for that id.
 - `a2a_list()` — configured peers + persisted conversations + metrics.
 - `a2a_history(context_id, limit?)` — recall a persisted conversation
   (this is the production consumer of the persistence layer).
@@ -47,18 +50,29 @@ Peers resolved from `config.yaml` → `a2a_agents`, or a direct URL.
 - JSON-RPC methods: `message/send`, `message/stream` (SSE), `tasks/get`,
   `tasks/list`, `tasks/cancel`, `tasks/subscribe`,
   `tasks/pushNotificationConfig/create` (legacy `set` names accepted).
+  `message/send` honors `configuration.returnImmediately` (and the older
+  `configuration.blocking: false`): the HTTP call returns a working Task at
+  once, a background waiter records the real result, and `tasks/get` can
+  poll it. Blocking send is unchanged (wait until done or `A2A_REPLY_TIMEOUT`).
+  The orphan watchdog skips task ids that still have a live waiter, so a
+  long job is not marked failed just because it has been working for more
+  than five minutes. The waiter itself stops after 24 hours
+  (`_BACKGROUND_WAIT_SECONDS`) and records the task as failed, so a hung
+  gateway turn cannot pin a daemon thread forever.
 - **Live-session injection (the #11025 insight):** inbound tasks route through
   the normal `MessageEvent` → `handle_message` path keyed by the A2A
   `contextId`, so the agent that answers is the same one serving the user —
   full memory/context, not a clone. The reply returns through `adapter.send()`,
   which fulfils the pending per-**task** `Future` the HTTP request is blocked
-  on (per-context FIFO, so concurrent same-context requests can't cross-talk);
+  on. The final's reply anchor (the task id) picks the task, and the adapter
+  dispatches one turn per context at a time, so the gateway busy queue never
+  folds two same-context tasks into one turn;
   `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).
+  orphaned tasks after 5 minutes if they have no live waiter (idempotent
+  transitions, no double counting in metrics).
 - **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`.
@@ -124,9 +138,19 @@ outbound client tools (`/metrics` and `a2a_list` report both directions).
 
 ## Persistence (survives compaction)
 A2A conversations are written to `~/.hermes/a2a_conversations/<context>.jsonl`,
-outside the context-compaction pipeline — compaction and restarts can't lose
+outside the context-compaction pipeline. Compaction and restarts can't lose
 them (#11025 requirement). The `a2a_history` tool recalls them by context id.
 
+Outbound `a2a_call` with `return_immediately` writes the prompt at send time.
+The peer reply is written only when `a2a_get_task` sees a terminal state. An
+unpolled job therefore has a user line and no agent line in `a2a_history`.
+That is a known property, not a missing persist. Inbound tasks still write
+both sides, because the background waiter records the reply itself.
+
+`a2a_call` and `a2a_get_task` cache the peer Agent Card for 60 seconds so a
+polling loop does not refetch it on every GetTask. `a2a_discover` always
+fetches a fresh card.
+
 ## Requirements traced to the cluster
 
 | Source | Requirement | Where |
diff --git a/plugins/platforms/a2a/README.md b/plugins/platforms/a2a/README.md
index 2d18bc3f7c8ad..6f5ce727c1df0 100644
--- a/plugins/platforms/a2a/README.md
+++ b/plugins/platforms/a2a/README.md
@@ -31,10 +31,11 @@ a2a_agents:
 
 ## Outbound — call other agents
 
-The agent gets five tools:
+The agent gets these tools:
 
 - `a2a_discover(url)` — what can this agent do?
-- `a2a_call(agent, message, context_id?)` — send it a task, get the reply.
+- `a2a_call(agent, message, context_id?, return_immediately?)` — send it a task, get the reply. Set `return_immediately` for long jobs to get a task id back at once.
+- `a2a_get_task(agent, task_id)` — check a running task (state and, when done, the reply).
 - `a2a_list()` — configured peers, saved conversations, metrics.
 - `a2a_history(context_id)` — recall a saved A2A conversation.
 - `a2a_orchestrate(capability, message, mode?)` — fan-out a task to every
diff --git a/plugins/platforms/a2a/adapter.py b/plugins/platforms/a2a/adapter.py
index d692045b89aa0..7133f35f7ffde 100644
--- a/plugins/platforms/a2a/adapter.py
+++ b/plugins/platforms/a2a/adapter.py
@@ -18,6 +18,7 @@
 import urllib.parse
 import urllib.request
 from collections import deque
+from contextvars import copy_context
 from concurrent.futures import Future
 from concurrent.futures import TimeoutError as FuturesTimeout
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -32,6 +33,7 @@
 
 logger = logging.getLogger(__name__)
 
+_BACKGROUND_WAIT_SECONDS = 24 * 60 * 60
 _DEFAULT_PORT = 9900
 # seconds: orphan grace floor / ceiling / watchdog period. The ceiling keeps the sweep
 # meaningful when A2A_REPLY_TIMEOUT is absurd (1e18 would never fail an orphan).
@@ -291,10 +293,14 @@ def __init__(self, config, **kwargs):
         self._profile_sessions: Dict[tuple[str, str, str], str] = {}
         self._profile_session_locks: Dict[tuple[str, str, str], threading.Lock] = {}
         self._profile_session_locks_guard = threading.Lock()
-        # Pending reply futures: task_id -> (context_id, Future). _pending_order keeps per-context
-        # FIFO so adapter.send() — which only knows the context — resolves the oldest task.
+        # Pending reply futures: task_id -> (context_id, Future). send() resolves the task named by
+        # the final's reply anchor (the gateway anchors on the inbound message id == task id).
         self._pending: Dict[str, tuple[str, Future]] = {}
-        self._pending_order: Dict[str, deque[str]] = {}
+        # One dispatched turn per context. The gateway's busy queue merges or replaces queued text
+        # for a session, which would fold two tasks into one turn: one gets the other's answer and
+        # the other never settles. Later tasks wait here until the in-flight one is popped.
+        self._inflight: Dict[str, str] = {}
+        self._queued: Dict[str, deque[tuple[str, MessageEvent]]] = {}
         # Request ownership outlives reply Futures and also covers synchronous profile forwards.
         self._active_tasks: set[str] = set()
         self._pending_lock = threading.Lock()
@@ -346,7 +352,8 @@ async def disconnect(self) -> None:
             for tid in list(self._pending):
                 self._resolve_locked(tid, protocol.STATE_FAILED, "[agent shutting down]")
             self._pending.clear()
-            self._pending_order.clear()
+            self._inflight.clear()
+            self._queued.clear()
             self._active_tasks.clear()
 
     def _watchdog_loop(self) -> None:
@@ -480,7 +487,6 @@ def _add_pending(self, task_id: str, context_id: str) -> Future:
         with self._pending_lock:
             self._active_tasks.add(task_id)
             self._pending[task_id] = (context_id, fut)
-            self._pending_order.setdefault(context_id, deque()).append(task_id)
         return fut
 
     def _activate_task(self, task_id: str) -> None:
@@ -491,11 +497,38 @@ def _pop_pending(self, task_id: str) -> None:
         with self._pending_lock:
             self._active_tasks.discard(task_id)
             entry = self._pending.pop(task_id, None)
-            order = self._pending_order.get(entry[0]) if entry else None
-            if order and task_id in order:
-                order.remove(task_id)
-            if order is not None and not order:
-                self._pending_order.pop(entry[0], None)
+            nxt = self._advance_context_locked(entry[0]) if entry and self._inflight.get(entry[0]) == task_id else None
+        if nxt is not None:
+            self._dispatch_queued(*nxt)
+
+    def _claim_context(self, context_id: str, task_id: str, event: MessageEvent) -> bool:
+        """True if ``task_id`` may dispatch now; otherwise it waits behind the in-flight task."""
+        with self._pending_lock:
+            if context_id in self._inflight:
+                self._queued.setdefault(context_id, deque()).append((task_id, event))
+                return False
+            self._inflight[context_id] = task_id
+            return True
+
+    def _advance_context_locked(self, context_id: str) -> Optional[tuple[str, MessageEvent]]:
+        """Hand the context to the next queued task still waiting (cancelled ones were popped)."""
+        queue = self._queued.get(context_id)
+        while queue:
+            task_id, event = queue.popleft()
+            entry = self._pending.get(task_id)
+            if entry and not entry[1].done():
+                self._inflight[context_id] = task_id
+                return task_id, event
+        self._queued.pop(context_id, None)
+        self._inflight.pop(context_id, None)
+        return None
+
+    def _dispatch_queued(self, task_id: str, event: MessageEvent) -> None:
+        try:
+            asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)
+        except Exception as e:
+            # Its waiter finalizes the failure and pops it, which hands the context on.
+            self._resolve_task(task_id, protocol.STATE_FAILED, security.redact_outbound(f"Dispatch failed: {e}"))
 
     def _resolve_locked(self, task_id: str, state: str, text: str) -> bool:
         entry = self._pending.get(task_id)
@@ -508,9 +541,14 @@ def _resolve_task(self, task_id: str, state: str, text: str) -> bool:
         with self._pending_lock:
             return self._resolve_locked(task_id, state, text)
 
-    def _resolve_oldest_for_context(self, context_id: str, state: str, text: str) -> bool:
+    def _resolve_final(self, context_id: str, anchor: Optional[str], text: str) -> bool:
+        """Resolve the task that owns a final: its anchor, else the context's in-flight task. A
+        late final anchored on a finished task settles nothing, so it can't answer a sibling."""
         with self._pending_lock:
-            return any(self._resolve_locked(tid, state, text) for tid in self._pending_order.get(context_id, ()))
+            task_id = anchor or self._inflight.get(context_id)
+            entry = self._pending.get(task_id or "")
+            return bool(entry and entry[0] == context_id
+                        and self._resolve_locked(task_id, protocol.STATE_COMPLETED, text))
 
     def _scope_for_agent(self, agent: Optional[dict]) -> tuple[str, str]:
         return tuple(str((agent or self._agents[""]).get(k) or "") for k in ("slug", "tenant"))
@@ -525,7 +563,7 @@ def _end_task(self, rec: dict, state: str, text: str, stored_reply: str = "") ->
         protocol.metrics.tasks_failed += state == protocol.STATE_FAILED
         return protocol.build_task(rec["task_id"], rec["context_id"], state, text, created_at=rec["created_iso"]), None
 
-    def _prepare_task(self, params: dict, peer: str, agent: Optional[dict] = None) -> tuple[Optional[dict], Optional[dict]]:
+    def _prepare_task(self, params: dict, peer: str, agent: Optional[dict] = None, *, defer_forward: bool = False) -> tuple[Optional[dict], Optional[dict]]:
         """Validate, register, and dispatch an inbound message (HTTP worker thread). Returns
         (terminal_task, None) when it ends immediately, else (None, pending) with the future to wait on."""
         agent = agent or self._agents[""]
@@ -549,6 +587,11 @@ def _prepare_task(self, params: dict, peer: str, agent: Optional[dict] = None) -
         self._register_inline_push(task_id, params, agent=agent)
         if not agent.get("local", True):
             self._activate_task(task_id)
+            if defer_forward:
+                self.tasks.set_state(task_id, protocol.STATE_WORKING)
+                return None, {"task_id": task_id, "context_id": context_id, "peer": peer,
+                              "created_iso": rec["created_iso"], "started": time.time(),
+                              "forward": ({**agent, "timeout": _BACKGROUND_WAIT_SECONDS}, peer, context_id, framed)}
             try:
                 reply, state = self._forward_to_profile(agent, peer, context_id, framed)
                 self._record_outcome(task_id, context_id, peer, state, reply)
@@ -561,7 +604,8 @@ def _prepare_task(self, params: dict, peer: str, agent: Optional[dict] = None) -
         event = MessageEvent(text=framed, message_type=MessageType.TEXT, message_id=task_id,
                              source=self.build_source(chat_id=context_id, chat_name=f"a2a:{peer}", chat_type="dm", user_id=peer, user_name=peer))
         try:
-            asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)
+            if self._claim_context(context_id, task_id, event):
+                asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)
         except Exception as e:
             msg = security.redact_outbound(f"Dispatch failed: {e}")
             try:
@@ -628,6 +672,10 @@ def _finalize_task(self, pending: dict, state: str, reply: str) -> tuple[str, st
         """Record a dispatched task's outcome; returns (state, reply) after redaction and
         input-required detection (a leading marker flags a clarification request)."""
         task_id, context_id, peer = pending["task_id"], pending["context_id"], pending["peer"]
+        rec = self.tasks.get(task_id)
+        if rec and rec["state"] in protocol.TERMINAL_STATES:
+            self._pop_pending(task_id)
+            return rec["state"], rec.get("reply") or ""
         try:
             reply = security.redact_outbound(reply or "")
             stripped = reply.lstrip()
@@ -660,8 +708,73 @@ def _await_reply(self, pending: dict, keepalive=None) -> tuple[str, str]:
         return self._await_future(pending["future"], pending["started"] + _reply_timeout(), keepalive,
                                   (protocol.STATE_FAILED, "[agent did not reply in time]"))
 
+    @staticmethod
+    def _config_return_immediately(params: dict) -> bool:
+        """True when the caller asked for a task id now, not a blocking wait.
+
+        A2A v1.0 uses configuration.returnImmediately. Older peers used
+        configuration.blocking = false. Either one is enough.
+        """
+        cfg = params.get("configuration") if isinstance(params, dict) else None
+        if not isinstance(cfg, dict):
+            return False
+        if cfg.get("returnImmediately") is True or cfg.get("return_immediately") is True:
+            return True
+        if cfg.get("blocking") is False:
+            return True
+        return False
+
+    def _wait_in_background(self, pending: dict) -> None:
+        """Wait for the agent, then record the result.
+
+        Used when returnImmediately is set. The HTTP caller already has the
+        working task. A2A_REPLY_TIMEOUT applies only to blocking callers.
+        The wait still stops after _BACKGROUND_WAIT_SECONDS so a hung
+        gateway turn cannot pin a daemon thread and a forever-WORKING task.
+        """
+
+        def _run() -> None:
+            try:
+                try:
+                    if "forward" in pending:
+                        reply, state = self._forward_to_profile(*pending["forward"])
+                    else:
+                        state, reply = pending["future"].result(timeout=_BACKGROUND_WAIT_SECONDS)
+                except FuturesTimeout:
+                    logger.warning(
+                        "A2A: background waiter for task %s hit the %ss ceiling",
+                        pending.get("task_id"),
+                        _BACKGROUND_WAIT_SECONDS,
+                    )
+                    state, reply = (
+                        protocol.STATE_FAILED,
+                        "[agent did not reply in time]",
+                    )
+                except Exception:
+                    state, reply = protocol.STATE_FAILED, "[agent did not reply]"
+                self._finalize_task(pending, state, reply)
+            except Exception:
+                logger.debug("A2A: background waiter failed", exc_info=True)
+                try:
+                    self._finalize_task(
+                        pending, protocol.STATE_FAILED, "[agent did not reply]")
+                except Exception:
+                    pass
+
+        threading.Thread(
+            target=copy_context().run,
+            args=(_run,),
+            name=f"a2a-wait-{pending['task_id'][:12]}",
+            daemon=True,
+        ).start()
+
     def _rpc_message_send(self, req_id: Any, params: dict, peer: str, agent: Optional[dict] = None, v1_response: bool = False) -> dict:
-        task, pending = self._prepare_task(params, peer, agent=agent)
+        options = {"defer_forward": True} if agent and not agent.get("local", True) and self._config_return_immediately(params) else {}
+        task, pending = self._prepare_task(params, peer, agent=agent, **options)
+        if task is None and self._config_return_immediately(params):
+            self._wait_in_background(pending)
+            task = protocol.build_task(pending["task_id"], pending["context_id"], protocol.STATE_WORKING,
+                                       created_at=pending["created_iso"])
         if task is None:
             state, reply = self._finalize_task(pending, *self._await_reply(pending))
             task = protocol.build_task(pending["task_id"], pending["context_id"], state, reply, created_at=pending["created_iso"])
@@ -826,13 +939,15 @@ def fail(msg: str, *args) -> None:
         logger.debug("A2A: push notification sent for task %s", task_id)
 
     async def send(self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None):
-        """Fulfil the oldest pending reply Future for this context (``chat_id`` = A2A context id).
+        """Fulfil the reply Future of the task this final belongs to (``chat_id`` = A2A context id).
         Only sends carrying ``metadata['notify']`` (the base adapter's final-reply marker) satisfy
         the caller; progress/status/preview sends must not."""
+        # Stream-consumer finals carry their anchor in metadata instead of ``reply_to``.
+        anchor = str(reply_to or (metadata or {}).get("reply_to_message_id") or "") or None
         if not (metadata or {}).get("notify"):
             logger.debug("A2A: ignoring non-final send for context %s", chat_id)
-        elif not self._resolve_oldest_for_context(chat_id, protocol.STATE_COMPLETED, content or ""):
-            logger.debug("A2A: send() for context %s had no pending waiter", chat_id)  # late chunk / out-of-band
+        elif not self._resolve_final(chat_id, anchor, content or ""):
+            logger.info("A2A: final for context %s (anchor %s) matched no pending task", chat_id, anchor)
         return SendResult(success=True, message_id=str(int(time.time() * 1000)))
 
     async def send_typing(self, chat_id: str, metadata=None) -> None:
diff --git a/plugins/platforms/a2a/plugin.yaml b/plugins/platforms/a2a/plugin.yaml
index 110267dea5941..3863e9d492838 100644
--- a/plugins/platforms/a2a/plugin.yaml
+++ b/plugins/platforms/a2a/plugin.yaml
@@ -6,10 +6,10 @@ description: >
   A2A (Agent-to-Agent) protocol v1.0 support for Hermes Agent — both directions
   of the open Linux Foundation standard for inter-agent communication.
 
-  OUTBOUND (client tools): a2a_discover, a2a_call, a2a_list, a2a_history, and
-  a2a_orchestrate let the agent fetch another agent's Agent Card and send it
-  tasks over JSON-RPC — works with any A2A-compliant peer (Hermes, LangChain,
-  CrewAI, Google ADK, OpenClaw, ...).
+  OUTBOUND (client tools): a2a_discover, a2a_call, a2a_get_task, a2a_list,
+  a2a_history, and a2a_orchestrate let the agent fetch another agent's Agent
+  Card and send it tasks over JSON-RPC — works with any A2A-compliant peer
+  (Hermes, LangChain, CrewAI, Google ADK, OpenClaw, ...).
 
   INBOUND (platform adapter): exposes Hermes as an A2A-discoverable agent. An
   Agent Card is served at /.well-known/agent-card.json (v1.0 canonical path;
@@ -33,6 +33,7 @@ author: Nous Research
 provides_tools:
   - a2a_discover
   - a2a_call
+  - a2a_get_task
   - a2a_list
   - a2a_history
   - a2a_orchestrate
diff --git a/plugins/platforms/a2a/protocol.py b/plugins/platforms/a2a/protocol.py
index bd44a7e71e20c..d7cca750d2fca 100644
--- a/plugins/platforms/a2a/protocol.py
+++ b/plugins/platforms/a2a/protocol.py
@@ -435,17 +435,50 @@ def _conv_path(context_id: str) -> Path:
     return get_hermes_home() / "a2a_conversations" / f"{safe}.jsonl"
 
 
-def persist_message(context_id: str, role: str, text: str, task_id: str = "") -> None:
+def persist_message(context_id: str, role: str, text: str, task_id: str = "", *,
+                    peer_url: str | None = None, tenant: str = "") -> None:
     """Append one message to the context's on-disk conversation log. Never raises."""
     try:
         path = _conv_path(context_id)
         path.parent.mkdir(parents=True, exist_ok=True)
+        peer = {"peer": peer_url.rstrip("/"), "tenant": tenant} if peer_url is not None else {}
         with path.open("a", encoding="utf-8") as fh:
-            fh.write(json.dumps({"ts": time.time(), "role": role, "text": text, "task_id": task_id}, ensure_ascii=False) + "\n")
+            fh.write(json.dumps({"ts": time.time(), "role": role, "text": text, "task_id": task_id, **peer}, ensure_ascii=False) + "\n")
     except Exception:
         pass
 
 
+_peer_reply_lock = threading.Lock()
+
+
+def persist_peer_reply(context_id: str, text: str, task_id: str, peer_url: str,
+                       tenant: str, state: str) -> None:
+    """Count and persist each peer task outcome once, including after a client restart."""
+    key = {"peer": peer_url.rstrip("/"), "tenant": tenant, "task_id": task_id}
+    try:
+        with _peer_reply_lock:
+            path = _conv_path(context_id)
+            previous = None
+            if path.exists():
+                with path.open(encoding="utf-8") as fh:
+                    for line in fh:
+                        row = json.loads(line)
+                        if (row.get("role") == "user" and row.get("peer") == key["peer"]
+                                and row.get("tenant") == tenant):
+                            previous = None
+                        if row.get("role") == "agent" and all(row.get(k) == v for k, v in key.items()):
+                            previous = row
+            if previous and previous.get("state") == state and previous.get("text") == text:
+                return
+            path.parent.mkdir(parents=True, exist_ok=True)
+            with path.open("a", encoding="utf-8") as fh:
+                fh.write(json.dumps({"ts": time.time(), "role": "agent", "text": text,
+                                     "state": state, **key}, ensure_ascii=False) + "\n")
+            metrics.inbound_total += 1
+    except (OSError, ValueError):
+        pass
+
+
 def load_conversation(context_id: str, limit: int = 50) -> list[dict]:
     """Last *limit* messages for a context (empty list if none / unreadable)."""
     try:
diff --git a/plugins/platforms/a2a/tools.py b/plugins/platforms/a2a/tools.py
index 53859737afb0b..a359c0f1fd86b 100644
--- a/plugins/platforms/a2a/tools.py
+++ b/plugins/platforms/a2a/tools.py
@@ -7,6 +7,8 @@
 import contextlib
 import json
 import logging
+import threading
+import time
 import urllib.error
 import urllib.request
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -18,6 +20,11 @@
 
 logger = logging.getLogger(__name__)
 
+_CARD_CACHE_TTL = 60.0
+_card_cache: dict[tuple[str, str], tuple[float, Optional[dict]]] = {}
+_card_cache_lock = threading.Lock()
+_IN_PROGRESS_STATES = frozenset({protocol.STATE_WORKING, protocol.STATE_SUBMITTED})
+
 _DEFAULT_TIMEOUT = 120
 _ORCHESTRATE_MAX_WORKERS = 6  # max parallel peers for fan-out
 
@@ -75,6 +82,144 @@ def _fetch_card(base_url: str, headers: dict, timeout: int) -> dict:
     return _http_get_json(base + "/.well-known/agent.json", headers, timeout)
 
 
+def _fetch_card_cached(base_url: str, headers: dict, timeout: int) -> Optional[dict]:
+    """Best-effort Agent Card lookup with a short TTL.
+
+    a2a_call and a2a_get_task only need the JSON-RPC URL. Polling every few
+    seconds should not refetch the card each time. a2a_discover still calls
+    _fetch_card so it sees current skills.
+    """
+    key = (base_url.rstrip("/"), headers.get("Authorization", ""))
+    now = time.time()
+    with _card_cache_lock:
+        hit = _card_cache.get(key)
+        if hit is not None and (now - hit[0]) < _CARD_CACHE_TTL:
+            return hit[1]
+    card: Optional[dict] = None
+    try:
+        fetched = _fetch_card(base_url, headers, timeout)
+        if isinstance(fetched, dict):
+            card = fetched
+    except Exception:
+        card = None
+    with _card_cache_lock:
+        _card_cache[key] = (now, card)
+    return card
+
+
+def _interface_tenant(card: Optional[dict], peer: dict) -> str:
+    iface = _select_jsonrpc_interface(card)
+    if iface and iface.get("tenant"):
+        return str(iface["tenant"])
+    return str(peer.get("tenant") or "")
+
+
+def _short_state(state: str) -> str:
+    """TASK_STATE_COMPLETED -> completed (also passes through v0.3 states)."""
+    return state.replace("TASK_STATE_", "").replace("_", "-").lower() if state else ""
+
+
+def _as_bool(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)) and not isinstance(value, bool):
+        return value != 0
+    if isinstance(value, str):
+        return value.strip().lower() in ("1", "true", "yes", "on")
+    return False
+
+
+def _format_peer_task(agent_label: str, reply: str, context_id: str, state: str,
+                      task_id: str = "") -> str:
+    header = f"[{agent_label}"
+    if task_id:
+        header += f" · task {task_id}"
+    if context_id:
+        header += f" · context {context_id}"
+    if state:
+        header += f" · {_short_state(state)}"
+    header += "]"
+    body = reply or "(no text reply)"
+    if state in _IN_PROGRESS_STATES:
+        body = reply or "Task is still running."
+        if task_id:
+            body += (
+                f"\n\nCall a2a_get_task with agent '{agent_label}' and "
+                f"task_id '{task_id}' to check it."
+            )
+        else:
+            body += "\n\nCall a2a_get_task with the same agent to check it."
+    elif state == protocol.STATE_INPUT_REQUIRED:
+        body += (
+            "\n\n(The peer needs more input. Answer by calling a2a_call again "
+            f"with context_id '{context_id}'.)"
+        )
+    return f"{header}\n{body}"
+
+
+def _peer_http_error(agent: str, exc: urllib.error.HTTPError) -> str:
+    if exc.code in (401, 403):
+        return f"Error: peer '{agent}' rejected auth (HTTP {exc.code}). Check the configured token."
+    if exc.code == 429:
+        return f"Error: peer '{agent}' rate limited us (HTTP 429). Retry later."
+    return f"Error: call to '{agent}' failed. HTTP {exc.code}."
+
+
+def a2a_get_task(args: dict, **_: Any) -> str:
+    """Fetch the current state of a peer task by id."""
+    agent = str(args.get("agent") or args.get("agent_name") or args.get("name") or "").strip()
+    task_id = str(args.get("task_id") or args.get("taskId") or args.get("id") or "").strip()
+    if not agent or not task_id:
+        return "Error: both 'agent' and 'task_id' are required."
+
+    peer = _resolve_peer(agent)
+    if not peer or not peer.get("url"):
+        return (
+            f"Error: unknown agent '{agent}'. Configure it under 'a2a_agents' in "
+            f"config.yaml or pass a full http(s):// URL."
+        )
+
+    base_url = peer.get("url", "")
+    headers = _auth_header(peer.get("auth", {}) or {})
+    timeout = int(peer.get("timeout", _DEFAULT_TIMEOUT))
+    card = _fetch_card_cached(base_url, headers, min(timeout, 30))
+
+    rpc_body = {
+        "jsonrpc": "2.0",
+        "id": protocol.new_task_id(),
+        "method": "GetTask",
+        # v1 GetTask params use id. Older peers used taskId. Send both.
+        "params": {"id": task_id, "taskId": task_id},
+    }
+    tenant = _interface_tenant(card, peer)
+    if tenant:
+        rpc_body["params"]["tenant"] = tenant
+
+    try:
+        resp = _http_post_json(_rpc_url(base_url, card), rpc_body, headers, timeout)
+    except urllib.error.HTTPError as e:
+        return _peer_http_error(agent, e)
+    except Exception as e:
+        return f"Error: GetTask on '{agent}' failed. {e}"
+
+    if "error" in resp:
+        err = resp["error"]
+        return f"Error: peer '{agent}' returned an error: {err.get('message', err)}"
+
+    # SendMessage and GetTask both return a Task. Some peers wrap it as
+    # {task: {...}} the same way SendMessage does, so reuse that unwrap.
+    payload = protocol.unwrap_send_message_response(resp.get("result", {}))
+    if not isinstance(payload, dict):
+        return f"Error: peer '{agent}' returned an unexpected GetTask result."
+    reply = _reply_text_from_result(payload)
+    ctx = str(payload.get("contextId") or "")
+    state = (payload.get("status") or {}).get("state", "")
+    got_id = str(payload.get("id") or task_id)
+    if state and state not in _IN_PROGRESS_STATES:
+        protocol.persist_peer_reply(ctx or "unknown", reply, got_id, base_url, tenant, state)
+    return _format_peer_task(agent, reply, ctx, state, got_id)
+
+
 def _select_jsonrpc_interface(card: Optional[dict]) -> Optional[dict]:
     if isinstance(card, dict):
         for iface in card.get("supportedInterfaces", []) or []:
@@ -92,40 +237,67 @@ def _rpc_url(base_url: str, card: Optional[dict]) -> str:
     return base_url.rstrip("/")
 
 
-def _send_task(agent_label: str, peer: dict, message: str, context_id: str) -> tuple[str, str, str]:
-    """One SendMessage to a peer -> (reply_text, context_id, state). Raises urllib errors /
-    ValueError for the caller to format; handles redaction, audit, persistence, metrics."""
+def _send_task(
+    agent_label: str,
+    peer: dict,
+    message: str,
+    context_id: str,
+    return_immediately: bool = False,
+) -> tuple[str, str, str, str]:
+    """Send one message/send to a peer.
+
+    Returns (reply_text, context_id, state, task_id).
+
+    Raises urllib errors / ValueError for the caller to format. Handles
+    outbound redaction, audit, persistence, and metrics.
+    """
     base_url = peer.get("url", "")
     headers = _auth_header(peer.get("auth", {}) or {})
     timeout = int(peer.get("timeout", _DEFAULT_TIMEOUT))
-    try:
-        card = _fetch_card(base_url, headers, min(timeout, 30))  # best-effort, to learn the rpc URL
-    except Exception:
-        card = None
+
+    # Best-effort card fetch (to learn the rpc URL); non-fatal on failure.
+    card = _fetch_card_cached(base_url, headers, min(timeout, 30))
+
     ctx = context_id or protocol.new_context_id()
     safe_message = security.redact_outbound(message)
     # v1.0: contextId lives inside the Message, not at the params top level.
-    rpc_body = {"jsonrpc": "2.0", "id": protocol.new_task_id(), "method": "SendMessage",
-                "params": {"message": protocol.text_message(protocol.ROLE_USER, safe_message, context_id=ctx)}}
-    iface = _select_jsonrpc_interface(card)
-    tenant = str(iface["tenant"]) if iface and iface.get("tenant") else str(peer.get("tenant") or "")
+    params: dict[str, Any] = {
+        "message": protocol.text_message(protocol.ROLE_USER, safe_message, context_id=ctx),
+    }
+    if return_immediately:
+        params["configuration"] = {"returnImmediately": True}
+
+    rpc_body = {
+        "jsonrpc": "2.0",
+        "id": protocol.new_task_id(),
+        "method": "SendMessage",
+        "params": params,
+    }
+
+    tenant = _interface_tenant(card, peer)
     if tenant:
         rpc_body["params"]["tenant"] = tenant
+
     security.audit("outbound", agent_label, rpc_body["id"], safe_message)
-    protocol.persist_message(ctx, "user", safe_message, rpc_body["id"])
+    protocol.persist_message(ctx, "user", safe_message, rpc_body["id"], peer_url=base_url, tenant=tenant)
     protocol.metrics.outbound_total += 1
+
     resp = _http_post_json(_rpc_url(base_url, card), rpc_body, headers, timeout)
     if "error" in resp:
-        raise ValueError(f"Peer '{agent_label}' returned an error: {resp['error'].get('message', resp['error'])}")
-    payload = protocol.unwrap_send_message_response(resp.get("result", {}))
+        err = resp["error"]
+        raise ValueError(f"Peer '{agent_label}' returned an error: {err.get('message', err)}")
+
+    result = resp.get("result", {})
+    payload = protocol.unwrap_send_message_response(result)
     reply = _reply_text_from_result(payload)
-    reply_ctx, state = ctx, ""
+    reply_ctx, state, task_id = ctx, "", ""
     if isinstance(payload, dict):
         reply_ctx = payload.get("contextId", ctx)
         state = (payload.get("status") or {}).get("state", "")
-    protocol.persist_message(reply_ctx, "agent", reply, rpc_body["id"])
-    protocol.metrics.inbound_total += 1
-    return reply, reply_ctx, state
+        task_id = str(payload.get("id") or "")
+    if state not in _IN_PROGRESS_STATES:
+        protocol.persist_peer_reply(reply_ctx, reply, task_id or rpc_body["id"], base_url, tenant, state)
+    return reply, reply_ctx, state, task_id
 
 
 def _reply_text_from_result(result: Any) -> str:
@@ -170,30 +342,42 @@ def a2a_discover(args: dict, **_: Any) -> str:
 
 
 def a2a_call(args: dict, **_: Any) -> str:
-    """Send a task to a peer (configured name or direct URL); ``context_id`` continues a prior exchange."""
+    """Send a task to a peer agent and return its reply.
+
+    ``agent`` is a configured peer name (from ``a2a_agents``) or a direct URL.
+    ``context_id`` continues a prior exchange (multi-turn) when provided.
+    ``return_immediately`` asks the peer for a task id now so long jobs do
+    not hold this call open. Poll the id with a2a_get_task.
+    """
     # Accept common aliases models reach for (observed live: 'agent_name').
     agent = str(args.get("agent") or args.get("agent_name") or args.get("name") or "").strip()
     message = str(args.get("message") or args.get("text") or args.get("task") or "").strip()
     context_id = str(args.get("context_id") or args.get("contextId") or "").strip()
+    return_immediately = _as_bool(
+        args.get("return_immediately") if args.get("return_immediately") is not None
+        else args.get("returnImmediately")
+    )
     if not agent or not message:
         return "Error: both 'agent' and 'message' are required."
+
     peer = _resolve_peer(agent)
     if not peer or not peer.get("url"):
-        return f"Error: unknown agent '{agent}'. Configure it under 'a2a_agents' in config.yaml or pass a full http(s):// URL."
+        return (
+            f"Error: unknown agent '{agent}'. Configure it under 'a2a_agents' in "
+            f"config.yaml or pass a full http(s):// URL."
+        )
+
     try:
-        reply, reply_ctx, state = _send_task(agent, peer, message, context_id)
+        reply, reply_ctx, state, task_id = _send_task(
+            agent, peer, message, context_id, return_immediately=return_immediately)
     except urllib.error.HTTPError as e:
-        return _HTTP_CALL_ERRORS.get(e.code, "Error: call to '{agent}' failed — HTTP {code}.").format(agent=agent, code=e.code)
+        return _peer_http_error(agent, e)
     except ValueError as e:
         return str(e)
     except Exception as e:
-        return f"Error: call to '{agent}' failed — {e}."
-    short_state = state.replace("TASK_STATE_", "").replace("_", "-").lower()  # v0.3 states pass through
-    header = f"[{agent} · context {reply_ctx}" + (f" · {short_state}" if state else "") + "]"
-    body = reply or "(no text reply)"
-    if state == protocol.STATE_INPUT_REQUIRED:
-        body += f"\n\n(The peer needs more input — answer by calling a2a_call again with context_id '{reply_ctx}'.)"
-    return f"{header}\n{body}"
+        return f"Error: call to '{agent}' failed. {e}"
+
+    return _format_peer_task(agent, reply, reply_ctx, state, task_id)
 
 
 def a2a_list(args: dict | None = None, **_: Any) -> str:
@@ -245,7 +429,7 @@ def _match_peers_by_capability(capability: str) -> list[tuple[str, dict]]:
 def _call_peer_sync(agent_name: str, peer_entry: dict, message: str, context_id: str = "") -> tuple[str, str]:
     """Call a single peer synchronously -> (agent_name, reply_text)."""
     try:
-        reply, _ctx, _state = _send_task(agent_name, _peer_from_entry(peer_entry), message, context_id)
+        reply, _ctx, _state, _task_id = _send_task(agent_name, _peer_from_entry(peer_entry), message, context_id)
         return (agent_name, reply or "(no reply)")
     except Exception as e:
         return (agent_name, f"Error: {e}")
@@ -305,8 +489,12 @@ def _str(description: str) -> dict:
                  "reply to continue a multi-turn exchange.",
                  {"agent": _str("Configured peer name (from a2a_agents) or a full http(s):// URL."),
                   "message": _str("The task / message to send the peer, in natural language."),
+                  "return_immediately": {"type": "boolean", "description": "Return a task id now; poll with a2a_get_task."},
                   "context_id": _str("Optional: context id from a prior reply, to continue the conversation.")},
                  ["agent", "message"]),
+    "a2a_get_task": (a2a_get_task, "Poll a peer task by id and return its current state and reply.",
+                     {"agent": _str("Configured peer name or URL."), "task_id": _str("Task id returned by a2a_call.")},
+                     ["agent", "task_id"]),
     "a2a_list": (a2a_list, "List configured A2A peer agents, persisted A2A conversations, and metrics.", {}, []),
     "a2a_history": (a2a_history,
                     "Recall a persisted A2A conversation transcript by context_id (survives restarts and "
diff --git a/tests/hermes_cli/test_deferred_platform_client_tools.py b/tests/hermes_cli/test_deferred_platform_client_tools.py
index 699f148ff625b..69062a2a52b54 100644
--- a/tests/hermes_cli/test_deferred_platform_client_tools.py
+++ b/tests/hermes_cli/test_deferred_platform_client_tools.py
@@ -3,8 +3,8 @@
 Issue #78050: a bundled ``kind: platform`` plugin is registered as a deferred
 loader so ``hermes chat`` doesn't import ~20 gateway SDKs. The a2a plugin ships
 two independent things behind that one deferral — an inbound adapter (heavy)
-and five outbound client tools (``a2a_call``, ``a2a_discover``, ``a2a_list``,
-``a2a_history``, ``a2a_orchestrate``). Deferring the plugin deferred both, so
+and outbound client tools (``a2a_call``, ``a2a_discover``, ``a2a_get_task``,
+``a2a_list``, ``a2a_history``, ``a2a_orchestrate``). Deferring the plugin deferred both, so
 in a CLI/TUI process the client tools never registered at all:
 ``resolve_toolset("a2a")`` returned ``[]`` and the toolset was absent from the
 ``hermes tools`` checklist. The same tools worked in gateway/web processes only
@@ -27,6 +27,7 @@
 A2A_CLIENT_TOOLS = {
     "a2a_call",
     "a2a_discover",
+    "a2a_get_task",
     "a2a_history",
     "a2a_list",
     "a2a_orchestrate",
diff --git a/tests/plugins/test_a2a_async_tasks.py b/tests/plugins/test_a2a_async_tasks.py
new file mode 100644
index 0000000000000..0752ed5db0690
--- /dev/null
+++ b/tests/plugins/test_a2a_async_tasks.py
@@ -0,0 +1,557 @@
+"""A2A long-job path: returnImmediately, GetTask poll, live-task watchdog."""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+from concurrent.futures import Future
+
+from plugins.platforms.a2a import protocol, tools
+from plugins.platforms.a2a import adapter as a2a_adapter
+from plugins.platforms.a2a.adapter import A2AAdapter
+
+
+def _bare_adapter():
+    from gateway.config import PlatformConfig
+    return A2AAdapter(PlatformConfig(enabled=True))
+
+
+def _peer_cfg():
+    return {"a2a_agents": {"fpga": {"url": "http://fpga.local:9900"}}}
+
+
+class TestReturnImmediatelyFlag:
+    def test_v1_flag(self):
+        assert A2AAdapter._config_return_immediately(
+            {"configuration": {"returnImmediately": True}}
+        ) is True
+
+    def test_snake_case_flag(self):
+        assert A2AAdapter._config_return_immediately(
+            {"configuration": {"return_immediately": True}}
+        ) is True
+
+    def test_old_blocking_false(self):
+        assert A2AAdapter._config_return_immediately(
+            {"configuration": {"blocking": False}}
+        ) is True
+
+    def test_default_is_blocking(self):
+        assert A2AAdapter._config_return_immediately({}) is False
+        assert A2AAdapter._config_return_immediately(
+            {"configuration": {"blocking": True}}
+        ) is False
+        assert A2AAdapter._config_return_immediately(
+            {"configuration": {"returnImmediately": False}}
+        ) is False
+
+
+class TestWatchdogSkipsLiveTasks:
+    def test_skip_keeps_running_task(self):
+        store = protocol.TaskStore()
+        store.create("live", "ctx", "peer")
+        store.create("dead", "ctx", "peer")
+        failed = store.fail_orphans(-1, exclude={"live"})
+        assert failed == ["dead"]
+        assert store.get("live")["state"] == protocol.STATE_SUBMITTED
+        assert store.get("dead")["state"] == protocol.STATE_FAILED
+
+    def test_no_skip_fails_old_task(self):
+        store = protocol.TaskStore()
+        store.create("alone", "ctx", "peer")
+        failed = store.fail_orphans(-1)
+        assert failed == ["alone"]
+        assert store.get("alone")["state"] == protocol.STATE_FAILED
+
+    def test_adapter_skips_pending_waiters(self):
+        adapter = _bare_adapter()
+        adapter.tasks.create("live", "ctx", "peer")
+        adapter._add_pending("live", "ctx")
+        try:
+            failed = adapter.tasks.fail_orphans(-1, exclude=set(adapter._active_tasks))
+            assert failed == []
+            assert adapter.tasks.get("live")["state"] == protocol.STATE_SUBMITTED
+        finally:
+            adapter._pop_pending("live")
+
+
+class TestInboundReturnImmediately:
+    def _pending(self, adapter, task_id="t-imm", ctx="c-imm"):
+        adapter.tasks.create(task_id, ctx, "alice")
+        adapter.tasks.set_state(task_id, protocol.STATE_WORKING)
+        fut = adapter._add_pending(task_id, ctx)
+        pending = {
+            "task_id": task_id,
+            "context_id": ctx,
+            "peer": "alice",
+            "future": fut,
+            "created_iso": protocol.now_iso(),
+            "started": time.time(),
+        }
+        return pending, fut
+
+    def test_send_returns_working_without_waiting(self):
+        adapter = _bare_adapter()
+        pending, fut = self._pending(adapter)
+        adapter._prepare_task = lambda params, peer, agent=None: (None, pending)
+
+        resp = adapter._rpc_message_send(
+            1,
+            {"configuration": {"returnImmediately": True}, "message": {}},
+            "alice",
+        )
+        task = resp["result"]
+        assert task["id"] == "t-imm"
+        assert task["status"]["state"] == protocol.STATE_WORKING
+        assert fut.done() is False
+        assert "t-imm" in set(adapter._active_tasks)
+
+        fut.set_result((protocol.STATE_COMPLETED, "bitfile ready"))
+        deadline = time.time() + 2
+        rec = adapter.tasks.get("t-imm")
+        while rec["state"] != protocol.STATE_COMPLETED and time.time() < deadline:
+            time.sleep(0.02)
+            rec = adapter.tasks.get("t-imm")
+        assert rec["state"] == protocol.STATE_COMPLETED
+        assert rec["reply"] == "bitfile ready"
+        assert adapter.tasks.get("t-imm") is not None
+
+    def test_v1_send_wraps_working_task(self):
+        adapter = _bare_adapter()
+        pending, fut = self._pending(adapter, task_id="t-v1", ctx="c-v1")
+        adapter._prepare_task = lambda params, peer, agent=None: (None, pending)
+        try:
+            resp = adapter._rpc_message_send(
+                7,
+                {"configuration": {"returnImmediately": True}, "message": {}},
+                "alice",
+                v1_response=True,
+            )
+            wrapped = resp["result"]
+            assert "task" in wrapped
+            assert wrapped["task"]["status"]["state"] == protocol.STATE_WORKING
+            assert wrapped["task"]["id"] == "t-v1"
+        finally:
+            if not fut.done():
+                fut.set_result((protocol.STATE_FAILED, "test teardown"))
+            deadline = time.time() + 2
+            while "t-v1" in set(adapter._active_tasks) and time.time() < deadline:
+                time.sleep(0.02)
+
+    def test_blocking_send_still_waits(self):
+        adapter = _bare_adapter()
+        pending, fut = self._pending(adapter, task_id="t-block", ctx="c-block")
+        adapter._prepare_task = lambda params, peer, agent=None: (None, pending)
+        awaited = []
+
+        def fake_await(pending_arg, keepalive=None):
+            awaited.append(True)
+            return protocol.STATE_COMPLETED, "done now"
+
+        adapter._await_reply = fake_await
+        resp = adapter._rpc_message_send(1, {"message": {}}, "alice")
+        assert awaited == [True]
+        assert resp["result"]["status"]["state"] == protocol.STATE_COMPLETED
+        assert protocol.extract_text(resp["result"]["artifacts"][0]) == "done now"
+
+    def test_get_task_sees_working_then_completed(self):
+        adapter = _bare_adapter()
+        pending, fut = self._pending(adapter, task_id="t-poll", ctx="c-poll")
+        adapter._prepare_task = lambda params, peer, agent=None: (None, pending)
+        adapter._rpc_message_send(
+            1,
+            {"configuration": {"returnImmediately": True}, "message": {}},
+            "alice",
+        )
+        got = adapter._rpc_tasks_get(2, {"id": "t-poll"})
+        assert got["result"]["status"]["state"] == protocol.STATE_WORKING
+
+        fut.set_result((protocol.STATE_COMPLETED, "synth ok"))
+        deadline = time.time() + 2
+        rec = adapter.tasks.get("t-poll")
+        while rec["state"] != protocol.STATE_COMPLETED and time.time() < deadline:
+            time.sleep(0.02)
+            rec = adapter.tasks.get("t-poll")
+        got = adapter._rpc_tasks_get(3, {"taskId": "t-poll"})
+        assert got["result"]["status"]["state"] == protocol.STATE_COMPLETED
+        assert protocol.extract_text(got["result"]["artifacts"][0]) == "synth ok"
+
+    def test_already_closed_task_is_not_finalized_twice(self):
+        adapter = _bare_adapter()
+        pending, _fut = self._pending(adapter, task_id="t-dup", ctx="c-dup")
+        adapter.tasks.complete("t-dup", protocol.STATE_CANCELED, "stopped")
+        state, reply = adapter._finalize_task(
+            pending, protocol.STATE_COMPLETED, "should ignore")
+        assert state == protocol.STATE_CANCELED
+        assert reply == "stopped"
+        rec = adapter.tasks.get("t-dup")
+        assert rec["state"] == protocol.STATE_CANCELED
+        assert rec["reply"] == "stopped"
+
+    def test_background_waiter_fails_after_ceiling(self, monkeypatch):
+        monkeypatch.setattr(a2a_adapter, "_BACKGROUND_WAIT_SECONDS", 0.05)
+        adapter = _bare_adapter()
+        pending, fut = self._pending(adapter, task_id="t-ceil", ctx="c-ceil")
+        adapter._wait_in_background(pending)
+        deadline = time.time() + 2
+        rec = adapter.tasks.get("t-ceil")
+        while rec["state"] != protocol.STATE_FAILED and time.time() < deadline:
+            time.sleep(0.02)
+            rec = adapter.tasks.get("t-ceil")
+        assert rec["state"] == protocol.STATE_FAILED
+        assert "did not reply in time" in (rec.get("reply") or "")
+        assert "t-ceil" not in set(adapter._active_tasks)
+        fut.set_result((protocol.STATE_COMPLETED, "too late"))
+        rec = adapter.tasks.get("t-ceil")
+        assert rec["state"] == protocol.STATE_FAILED
+
+
+class TestOutboundCallAndGetTask:
+    def setup_method(self):
+        tools._card_cache.clear()
+
+    def test_call_sends_return_immediately_and_prints_task_id(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        monkeypatch.setattr(tools, "_http_get_json", lambda url, h, t: None)
+        captured = {}
+
+        def fake_post(url, body, headers, timeout):
+            captured["body"] = body
+            ctx = body["params"]["message"]["contextId"]
+            return protocol.jsonrpc_result(
+                body["id"],
+                {"task": protocol.build_task("task-99", ctx, protocol.STATE_WORKING)},
+            )
+
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        out = tools.a2a_call({
+            "agent": "fpga",
+            "message": "run synth",
+            "return_immediately": True,
+        })
+        assert captured["body"]["params"]["configuration"]["returnImmediately"] is True
+        assert "task-99" in out
+        assert "a2a_get_task" in out
+        assert "working" in out
+
+    def test_call_string_true_counts_as_flag(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        monkeypatch.setattr(tools, "_http_get_json", lambda url, h, t: None)
+        captured = {}
+
+        def fake_post(url, body, headers, timeout):
+            captured["flag"] = (body["params"].get("configuration") or {}).get(
+                "returnImmediately")
+            ctx = body["params"]["message"]["contextId"]
+            return protocol.jsonrpc_result(
+                body["id"],
+                protocol.build_task("t1", ctx, protocol.STATE_WORKING),
+            )
+
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        tools.a2a_call({
+            "agent": "fpga",
+            "message": "go",
+            "return_immediately": "true",
+        })
+        assert captured["flag"] is True
+
+    def test_default_call_does_not_set_the_flag(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        monkeypatch.setattr(tools, "_http_get_json", lambda url, h, t: None)
+        captured = {}
+
+        def fake_post(url, body, headers, timeout):
+            captured["params"] = body["params"]
+            ctx = body["params"]["message"]["contextId"]
+            return protocol.jsonrpc_result(
+                body["id"],
+                protocol.build_task("t1", ctx, protocol.STATE_COMPLETED, "ok"),
+            )
+
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        out = tools.a2a_call({"agent": "fpga", "message": "ping"})
+        assert "configuration" not in captured["params"]
+        assert "ok" in out
+        assert "completed" in out
+
+    def test_get_task_polls_completed_reply(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        monkeypatch.setattr(tools, "_http_get_json", lambda url, h, t: None)
+        captured = {}
+
+        def fake_post(url, body, headers, timeout):
+            captured["body"] = body
+            return protocol.jsonrpc_result(
+                body["id"],
+                protocol.build_task("task-99", "ctx-1", protocol.STATE_COMPLETED, "bitfile ready"),
+            )
+
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        out = tools.a2a_get_task({"agent": "fpga", "task_id": "task-99"})
+        assert captured["body"]["method"] == "GetTask"
+        assert captured["body"]["params"]["id"] == "task-99"
+        assert captured["body"]["params"]["taskId"] == "task-99"
+        assert "bitfile ready" in out
+        assert "completed" in out
+        assert "task-99" in out
+
+    def test_get_task_still_running(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        monkeypatch.setattr(tools, "_http_get_json", lambda url, h, t: None)
+
+        def fake_post(url, body, headers, timeout):
+            return protocol.jsonrpc_result(
+                body["id"],
+                protocol.build_task("task-99", "ctx-1", protocol.STATE_WORKING),
+            )
+
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        out = tools.a2a_get_task({"agent": "fpga", "task_id": "task-99"})
+        assert "working" in out
+        assert "a2a_get_task" in out
+
+    def test_get_task_requires_args(self):
+        assert "required" in tools.a2a_get_task({"agent": "", "task_id": "x"})
+        assert "required" in tools.a2a_get_task({"agent": "fpga", "task_id": ""})
+
+    def test_get_task_unknown_peer(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", lambda: {"a2a_agents": {}})
+        out = tools.a2a_get_task({"agent": "ghost", "task_id": "t1"})
+        assert "unknown agent" in out
+
+    def test_get_task_accepts_taskId_alias(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        monkeypatch.setattr(tools, "_http_get_json", lambda url, h, t: None)
+        captured = {}
+
+        def fake_post(url, body, headers, timeout):
+            captured["id"] = body["params"]["id"]
+            return protocol.jsonrpc_result(
+                body["id"],
+                protocol.build_task("t-alt", "c", protocol.STATE_COMPLETED, "done"),
+            )
+
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        out = tools.a2a_get_task({"agent": "fpga", "taskId": "t-alt"})
+        assert captured["id"] == "t-alt"
+        assert "done" in out
+
+    def test_get_task_reuses_card_within_ttl(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        gets = []
+
+        def fake_get(url, h, t):
+            gets.append(url)
+            return {"name": "fpga", "url": "http://fpga.local:9900"}
+
+        def fake_post(url, body, headers, timeout):
+            return protocol.jsonrpc_result(
+                body["id"],
+                protocol.build_task("task-99", "ctx-1", protocol.STATE_WORKING),
+            )
+
+        monkeypatch.setattr(tools, "_http_get_json", fake_get)
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        tools.a2a_get_task({"agent": "fpga", "task_id": "task-99"})
+        tools.a2a_get_task({"agent": "fpga", "task_id": "task-99"})
+        assert len(gets) == 1
+
+    def test_get_task_refetches_card_after_ttl(self, monkeypatch):
+        monkeypatch.setattr(tools, "_load_config", _peer_cfg)
+        monkeypatch.setattr(tools, "_CARD_CACHE_TTL", 0)
+        gets = []
+
+        def fake_get(url, h, t):
+            gets.append(url)
+            return {"name": "fpga", "url": "http://fpga.local:9900"}
+
+        def fake_post(url, body, headers, timeout):
+            return protocol.jsonrpc_result(
+                body["id"],
+                protocol.build_task("task-99", "ctx-1", protocol.STATE_WORKING),
+            )
+
+        monkeypatch.setattr(tools, "_http_get_json", fake_get)
+        monkeypatch.setattr(tools, "_http_post_json", fake_post)
+        tools.a2a_get_task({"agent": "fpga", "task_id": "task-99"})
+        tools.a2a_get_task({"agent": "fpga", "task_id": "task-99"})
+        assert len(gets) == 2
+
+def test_peer_polling_records_each_outcome_once(tmp_path, monkeypatch):
+    import json
+    import threading
+    from concurrent.futures import ThreadPoolExecutor
+    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+    reply = {"state": protocol.STATE_INPUT_REQUIRED, "text": "Which target?"}
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *_args):
+            pass
+
+        def do_GET(self):
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(b'{}')
+
+        def do_POST(self):
+            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+            task = protocol.build_task("shared-id", "shared-context", reply["state"], reply["text"])
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(json.dumps(protocol.jsonrpc_result(body["id"], task)).encode())
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    url = f"http://127.0.0.1:{server.server_port}"
+    initial = protocol.metrics.inbound_total
+    try:
+        assert "Which target?" in tools.a2a_call({"agent": url, "message": "Build"})
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            outputs = list(pool.map(lambda _: tools.a2a_get_task({"agent": url, "task_id": "shared-id"}), range(8)))
+        assert all("Which target?" in text for text in outputs)
+        assert protocol.metrics.inbound_total == initial + 1
+        assert "Which target?" in tools.a2a_call({"agent": url, "message": "Not sure", "context_id": "shared-context"})
+        assert "Which target?" in tools.a2a_get_task({"agent": url, "task_id": "shared-id"})
+        assert protocol.metrics.inbound_total == initial + 2
+        reply.update(state=protocol.STATE_COMPLETED, text="Built target")
+        for _ in range(3):
+            assert "Built target" in tools.a2a_get_task({"agent": url, "task_id": "shared-id"})
+        rows = [row for row in protocol.load_conversation("shared-context") if row["role"] == "agent"]
+        assert [row["text"] for row in rows] == ["Which target?", "Which target?", "Built target"]
+        assert protocol.metrics.inbound_total == initial + 3
+        monkeypatch.setattr(tools, "_configured_peers", lambda: {"other": {"url": url, "tenant": "other"}})
+        assert "Built target" in tools.a2a_get_task({"agent": "other", "task_id": "shared-id"})
+        assert protocol.metrics.inbound_total == initial + 4
+        assert len([row for row in protocol.load_conversation("shared-context") if row["role"] == "agent"]) == 4
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+@pytest.mark.parametrize("routed", [False, True])
+def test_background_completion_preserves_profile(tmp_path, monkeypatch, routed):
+    import json
+    import threading
+    from hermes_constants import get_hermes_home, set_hermes_home_override, reset_hermes_home_override
+
+    profile = tmp_path / "profile"
+    process_home = tmp_path / "process"
+    monkeypatch.setenv("HERMES_HOME", str(process_home))
+    token = set_hermes_home_override(profile)
+    release, done = threading.Event(), threading.Event()
+    adapter = _bare_adapter()
+    monkeypatch.setattr(adapter, "_send_push_notification", lambda *_args: done.set())
+    future = None
+    try:
+        if routed:
+            def forward(*_args):
+                assert get_hermes_home() == profile
+                assert release.wait(5)
+                return "profile reply", protocol.STATE_COMPLETED
+            monkeypatch.setattr(adapter, "_forward_to_profile", forward)
+            agent = {"local": False, "profile": "other", "slug": "other", "tenant": "other"}
+        else:
+            pending, future = TestInboundReturnImmediately()._pending(adapter, "profile-task", "profile-context")
+            monkeypatch.setattr(adapter, "_prepare_task", lambda *args, **kwargs: (None, pending))
+            agent = None
+        response = adapter._rpc_message_send(1, {
+            "message": protocol.text_message(protocol.ROLE_USER, "work", context_id="profile-context"),
+            "configuration": {"returnImmediately": True}}, "peer", agent=agent)
+        assert response["result"]["status"]["state"] == protocol.STATE_WORKING
+        if future:
+            future.set_result((protocol.STATE_COMPLETED, "profile reply"))
+        release.set()
+        assert done.wait(5)
+        rows = protocol.load_conversation("profile-context")
+        assert rows[-1]["text"] == "profile reply"
+        audit = [json.loads(line) for line in (profile / "a2a_audit.jsonl").read_text().splitlines()]
+        assert audit[-1]["direction"] == "outbound"
+        assert not (process_home / "a2a_audit.jsonl").exists()
+        assert not (process_home / "a2a_conversations").exists()
+        if routed:
+            # Streaming uses ordinary preparation even if the request includes the async option.
+            task, pending = adapter._prepare_task({
+                "message": protocol.text_message(protocol.ROLE_USER, "stream", context_id="stream-context"),
+                "configuration": {"returnImmediately": True}}, "peer", agent=agent)
+            assert task["status"]["state"] == protocol.STATE_COMPLETED
+            assert pending is None
+    finally:
+        release.set()
+        reset_hermes_home_override(token)
+
+
+def _rpc(port, method, params):
+    import json
+    import urllib.request
+    req = urllib.request.Request(f"http://127.0.0.1:{port}/", json.dumps(
+        {"jsonrpc": "2.0", "id": 1, "method": method, "params": params}).encode(), {"Content-Type": "application/json"})
+    with urllib.request.urlopen(req, timeout=3) as r:
+        return json.load(r).get("result")
+
+
+def test_same_context_tasks_each_get_their_own_turn(tmp_path, monkeypatch):
+    """Three detached tasks in one context, real handle_message: the gateway busy queue must not
+    fold two of them into one turn (one answered with the other's reply, the other stuck)."""
+    import asyncio
+    import socket
+
+    from gateway.config import PlatformConfig
+    import agent.oneshot_footprint as footprint
+
+    for k in ("A2A_BEARER_TOKEN", "A2A_PEER_TOKENS"):
+        monkeypatch.delenv(k, raising=False)
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        port = s.getsockname()[1]
+    monkeypatch.setenv("A2A_PORT", str(port))
+    monkeypatch.setattr(footprint, "is_single_query_session", lambda: False)
+    a = A2AAdapter(PlatformConfig(enabled=True, extra={"port": port}))
+
+    async def model(event):
+        job = event.text.strip().rsplit("\n", 1)[-1]
+        await asyncio.sleep(1.0 if job == "job-0" else 0.2)  # later tasks arrive while job-0 runs
+        return "REPLY " + job
+    a._message_handler = model
+
+    def body():
+        tids = []
+        for i in range(3):
+            tids.append(_rpc(port, "SendMessage", {
+                "message": {"messageId": f"m-{i}", "role": "ROLE_USER", "contextId": "ctx", "parts": [{"text": f"job-{i}"}]},
+                "configuration": {"returnImmediately": True}})["task"]["id"])
+        deadline = time.time() + 15
+        while True:
+            tasks = [_rpc(port, "GetTask", {"id": t}) for t in tids]
+            if time.time() > deadline or all(t["status"]["state"] in protocol.TERMINAL_STATES for t in tasks):
+                return tasks
+            time.sleep(0.5)
+
+    async def main():
+        assert await a.connect()
+        try:
+            return await asyncio.to_thread(body)
+        finally:
+            await a.disconnect()
+
+    tasks = asyncio.run(main())
+    assert [(t["status"]["state"], protocol.extract_text(t["artifacts"][0]) if t.get("artifacts") else None)
+            for t in tasks] == [(protocol.STATE_COMPLETED, f"REPLY job-{i}") for i in range(3)]
+
+
+def test_late_anchored_final_never_answers_a_sibling():
+    import asyncio
+
+    a = _bare_adapter()
+    first = a._add_pending("task-1", "ctx")
+    second = a._add_pending("task-2", "ctx")
+    first.set_result((protocol.STATE_FAILED, "[agent did not reply in time]"))
+
+    asyncio.run(a.send("ctx", "late answer for task 1", reply_to="task-1", metadata={"notify": True}))
+    assert not second.done()
+    asyncio.run(a.send("ctx", "answer 2", metadata={"notify": True, "reply_to_message_id": "task-2"}))
+    assert second.result(timeout=0) == (protocol.STATE_COMPLETED, "answer 2")
diff --git a/tests/plugins/test_a2a_plugin.py b/tests/plugins/test_a2a_plugin.py
index 7e71590879862..164ca6db5e3a4 100644
--- a/tests/plugins/test_a2a_plugin.py
+++ b/tests/plugins/test_a2a_plugin.py
@@ -558,6 +558,9 @@ def register_tool(self, name, toolset, schema, handler, **kw):
         out = registry.dispatch("a2a_call", {"agent": "", "message": ""})
         assert "required" in out and "AttributeError" not in out
 
+        out = registry.dispatch("a2a_get_task", {"agent": "", "task_id": ""})
+        assert "required" in out and "AttributeError" not in out
+
         out = registry.dispatch("a2a_history", {})
         assert "required" in out and "AttributeError" not in out
 
@@ -612,6 +615,7 @@ async def run():
             final = await adapter.send(
                 "ctx-final",
                 "FINAL_PROOF_PAYLOAD",
+                reply_to="task-final",
                 metadata={"notify": True},
             )
             assert final.success is True
@@ -622,19 +626,19 @@ async def run():
         finally:
             adapter._pop_pending("task-final")
 
-    def test_concurrent_same_context_tasks_resolve_fifo(self):
-        """Two in-flight tasks sharing a context must not cross-talk: replies
-        resolve the oldest outstanding task first."""
+    def test_concurrent_same_context_tasks_resolve_by_task_anchor(self):
+        """Two in-flight tasks sharing a context must not cross-talk: each final
+        resolves the task named by its reply anchor, whatever the order."""
         adapter = _bare_adapter()
         fut1 = adapter._add_pending("task-1", "ctx-shared")
         fut2 = adapter._add_pending("task-2", "ctx-shared")
 
         async def run():
-            await adapter.send("ctx-shared", "reply one", metadata={"notify": True})
-            assert fut1.done() and not fut2.done()
-            assert fut1.result(timeout=0)[1] == "reply one"
-            await adapter.send("ctx-shared", "reply two", metadata={"notify": True})
+            await adapter.send("ctx-shared", "reply two", reply_to="task-2", metadata={"notify": True})
+            assert fut2.done() and not fut1.done()
             assert fut2.result(timeout=0)[1] == "reply two"
+            await adapter.send("ctx-shared", "reply one", reply_to="task-1", metadata={"notify": True})
+            assert fut1.result(timeout=0)[1] == "reply one"
 
         try:
             asyncio.run(run())
@@ -668,7 +672,7 @@ def test_on_processing_complete_does_not_clobber_reply(self):
         event = SimpleNamespace(message_id="task-ok")
 
         async def run():
-            await adapter.send("ctx-ok", "real reply", metadata={"notify": True})
+            await adapter.send("ctx-ok", "real reply", reply_to="task-ok", metadata={"notify": True})
             await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS)
 
         try:
@@ -872,7 +876,8 @@ async def fake_handle_message(event):
         else:
             reply = reply_fn(event)
         if reply is not None:
-            await adapter.send(event.source.chat_id, reply, metadata={"notify": True})
+            # The gateway anchors a final on the inbound message id (== A2A task id).
+            await adapter.send(event.source.chat_id, reply, reply_to=event.message_id, metadata={"notify": True})
 
     adapter.handle_message = fake_handle_message  # type: ignore
     adapter._message_handler = object()  # non-None so dispatch proceeds
@@ -1397,7 +1402,7 @@ def fake_post(url, body, headers, timeout):
 
         monkeypatch.setattr(tools, "_http_get_json", fake_get)
         monkeypatch.setattr(tools, "_http_post_json", fake_post)
-        reply, _ctx, _state = tools._send_task(
+        reply, _ctx, _state, _tid = tools._send_task(
             "dev", {"url": "http://peer.example", "auth": {}, "timeout": 5}, "hello", "ctx-1"
         )
         assert reply == "ok"
@@ -1469,7 +1474,7 @@ def fake_post(url, body, headers, timeout):
 
         monkeypatch.setattr(tools, "_http_get_json", fake_get)
         monkeypatch.setattr(tools, "_http_post_json", fake_post)
-        reply, _ctx, state = tools._send_task(
+        reply, _ctx, state, _tid = tools._send_task(
             "dev", {"url": "http://peer.example", "auth": {}, "timeout": 5}, "hello", "ctx-1")
         assert reply == "ok"
         assert state == protocol.STATE_COMPLETED
diff --git a/tests/plugins/test_a2a_tools_gate.py b/tests/plugins/test_a2a_tools_gate.py
index d5e220417c6d5..30eb339f347ef 100644
--- a/tests/plugins/test_a2a_tools_gate.py
+++ b/tests/plugins/test_a2a_tools_gate.py
@@ -46,8 +46,8 @@ def test_config_crash_fails_closed(self):
         with patch.object(a2at, "_load_config", side_effect=RuntimeError("boom")):
             self.assertFalse(a2at._a2a_tools_available())
 
-    def test_all_five_tools_carry_the_gate(self):
-        """Every a2a_* registration must pass the check_fn — a sixth tool
+    def test_every_tool_carries_the_gate(self):
+        """Every a2a_* registration must pass the check_fn — a new tool
         added without it would silently reopen the hole."""
         seen = {}
 
diff --git a/website/docs/user-guide/messaging/a2a.md b/website/docs/user-guide/messaging/a2a.md
index 24b716fa56d7d..d24916e499291 100644
--- a/website/docs/user-guide/messaging/a2a.md
+++ b/website/docs/user-guide/messaging/a2a.md
@@ -46,7 +46,8 @@ With the `a2a` toolset enabled, the agent gets:
 | Tool | What it does |
 |---|---|
 | `a2a_discover(url)` | Fetch and summarize a peer's Agent Card |
-| `a2a_call(agent, message, context_id?)` | Send a task, get the reply; multi-turn via `context_id` |
+| `a2a_call(agent, message, context_id?, return_immediately?)` | Send a task, get the reply. Multi-turn via `context_id`. Set `return_immediately` for long jobs to get a `task_id` back at once. |
+| `a2a_get_task(agent, task_id)` | Poll a running task. Returns state and, when done, the reply. |
 | `a2a_list()` | Configured peers, saved conversations, metrics |
 | `a2a_history(context_id)` | Recall a persisted A2A conversation |
 | `a2a_orchestrate(capability, message, mode?)` | Fan a task out to every peer advertising a capability (`all` / `first` / `best`) |