diff --git a/CHANGELOG.md b/CHANGELOG.md
index f038705..9645f22 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,6 +21,7 @@
 - Propagate workflow cancellation and close execution-owned resources exactly once.
 - Anonymize orchestration access tokens and bound registry response deadlines.
 - Preserve multi-value results from arbitrary terminal workflow nodes.
+- Map pre-task failures from the standard A2A error envelope while keeping post-creation failures in task state.
 - Publish a compiled host integration example and require sample regression tests.
 - Keep documentation host-neutral; sample class and AgentCard identifiers remain unchanged.
 
diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md
index 6482392..f3db2ca 100644
--- a/docs/en/API_REFERENCE.md
+++ b/docs/en/API_REFERENCE.md
@@ -256,7 +256,7 @@ runtime, diagnostic adapter, or explicit lifecycle requires them.
 | `ProtocolResponses`           | A2A event/result assembly helpers |
 | `ClientEventMapper`           | Stable event projection for callbacks and diagnostics |
 | `WireLog`                     | Correlation context and protocol-observation facade |
-| `RemoteProblemException`      | Structured remote problem response (`status`, `title`, `detail`, `type`, `timestamp`) |
+| `RemoteA2AErrorException`     | Standard A2A HTTP error projection (`code`, `status`, `message`, typed details and safe headers) |
 
 ---
 
@@ -589,11 +589,11 @@ not add private negotiation-state keys to the wire message.
 
 ## Error Handling
 
-- Top-level remote problem responses retain their fields in `RemoteProblemException`;
-  `findIn(Throwable)` finds it through wrappers. Workflow failures use `remote.problem.<status>`,
-  with details in history / TASK_RESPONSE rather than business outputs.
+- Standard top-level A2A error envelopes retain their fields in `RemoteA2AErrorException`;
+  `findIn(Throwable)` also projects wrapped SDK errors. Workflow failures use `a2a.<reason>` or
+  `a2a.http.<status>`, with details in history / TASK_RESPONSE rather than business outputs.
   Executor node events include executionId; task response events and history include the logical taskId.
-  See [integration guidance](INTEGRATION_GUIDE.md#14-remote-problem-responses) for propagation and logging.
+  See [integration guidance](INTEGRATION_GUIDE.md#14-a2a-errors-and-task-failures) for propagation and logging.
 
 - Missing/null/exceptional/timed-out callbacks fail explicitly. Host SDK errors may be mapped to BusinessFailure with
   safe code/details.
@@ -655,7 +655,7 @@ Auto-configures the following beans (all `@ConditionalOnMissingBean`, so you can
 | `eventBusProcessor` | `MainEventBusProcessor`       | Event bus processor                                       |
 | `requestHandler`    | `RequestHandler`              | Default request handler                                   |
 | `restHandler`       | `RestHandler`                 | REST protocol handler                                     |
-| `a2aController`     | `A2AController`               | Spring MVC controller (`message:send` + `message:stream`) |
+| `a2aController`     | `A2AController`               | Spring MVC controller for message and task endpoints     |
 
 ### A2AController
 
@@ -663,6 +663,14 @@ Spring MVC controller that exposes A2A REST endpoints:
 
 - `POST {path-prefix}/message:send` — blocking send
 - `POST {path-prefix}/message:stream` — SSE streaming
+- `GET {path-prefix}/tasks/{id}` — query a task
+- `GET {path-prefix}/tasks` — list tasks with optional filters
+- `POST {path-prefix}/tasks/{id}:cancel` — cancel a task
+- `POST {path-prefix}/tasks/{id}:subscribe` — subscribe to task updates over SSE
+
+HTTP+JSON responses use `application/a2a+json`; successful streaming responses use `text/event-stream`. When a stream
+closes, the controller cancels its publisher subscription and notifies the SDK event consumer so that server-side
+polling does not remain active.
 
 ### Usage
 
diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md
index dac010d..58fea3c 100644
--- a/docs/en/INTEGRATION_GUIDE.md
+++ b/docs/en/INTEGRATION_GUIDE.md
@@ -549,28 +549,34 @@ Cleanup is enabled and fail-fast by default so stale tasks cannot silently lead
 the demo may cancel active tasks created by another installation; use an isolated identity or disable cleanup after
 providing an equivalent ownership-aware policy.
 
-## 14. Remote problem responses
-
-A successful HTTP envelope does not guarantee a successful task: SSE data can contain a top-level
-problem object such as `{"status":429,"detail":"Active task limit reached","type":""}`.
-Top-level numeric status 400–599 with title/detail becomes `RemoteProblemException`, preserving
-status, title, detail, type and timestamp. Both synchronous and streaming responses are checked.
-Nested business data is not classified as a protocol error.
-Detection is limited to bodies actually delivered by the SDK. A non-2xx HTTP response may be rejected
-first as an HTTP/authentication error by the SDK or adapter, without exposing its problem body;
-such failures retain the generic failure path rather than fabricating errorDetails.
-
-The error fails the call without fabricating a successful task, invoking onNegotiation, or automatically
-resubmitting it. For 400, inspect business inputs; for 429, inspect dispatched-agent capacity and active tasks or
-subscriptions before deciding whether to retry. Negotiation still requires a valid Negotiation-T Propose.
-Use `RemoteProblemException.findIn(error)` to inspect the cause chain (null means no problem found);
-workflow execution follows its existing failure path.
-Task results and execution history / TASK_RESPONSE events expose `remote.problem.400`,
-`remote.problem.429`, etc. as errorCode, the reason as error, the five fields above as errorDetails,
-and empty outputs. Known credential fields use the protocol logger's redaction rules;
-unrecognized SDK exceptions still expose only the exception type.
-Independent authorization/subscription failures do not determine the workflow result.
-Pretty logging does not modify the received problem payload.
+## 14. A2A errors and task failures
+
+Keep failures before and after task creation separate. A request rejected before a task is created uses a
+non-2xx HTTP response with the standard A2A `google.rpc.Status` JSON envelope:
+
+```json
+{"error":{"code":400,"status":"INVALID_ARGUMENT","message":"A required parameter is missing","details":[{"@type":"type.googleapis.com/google.rpc.ErrorInfo","reason":"INVALID_PARAMS","domain":"a2a-protocol.org","metadata":{"field":"port"}}]}}
+```
+
+`RemoteA2AErrorException` preserves the observed HTTP status, envelope code/status/message, typed details,
+ErrorInfo reason/domain and safe response headers. `findIn(Throwable)` also projects typed A2A Java SDK
+errors. The direct transport checks ordinary responses and defensively detects a top-level error envelope if
+an SDK exposes a pre-stream rejection as SSE data. Nested `error` objects inside a task, message or artifact
+remain business content.
+
+If an SSE call closes before delivering any A2A event, the transport fails immediately instead of waiting for the
+workflow send timeout. The underlying SDK does not expose the HTTP status of an empty streaming response, so this case
+is reported as a transport/protocol failure rather than a fabricated HTTP error code.
+
+Workflow history and TASK_RESPONSE expose a stable code such as `a2a.invalid_params`; unknown HTTP errors
+fall back to `a2a.http.<status>`. `errorDetails` retains protocol facts and `retryAfter` when observed. Such an
+error creates no successful output, does not invoke onNegotiation and is not retried automatically.
+
+After a task has been created, a business execution failure is not an HTTP protocol error. The agent returns
+HTTP 200 with a Task or status update in `TASK_STATE_FAILED`; the TaskStatus message, task metadata and
+artifacts carry the available failure evidence. The engine marks the workflow task failed, keeps that evidence
+in `receivedMessages`, and does not interpret an extension-specific result schema. Independent authorization
+and subscription failures remain independent of workflow outcome.
 
 For an `ALL_SUCCESS` step, one failed task does not retroactively cancel a peer call that was already dispatched: the
 peer result or timeout is collected before the workflow returns `success=false`. Downstream steps are not invoked;
@@ -584,9 +590,9 @@ The host agent decides how to expose a failed execution to its caller. It must n
 success artifact. A persistent host owns notification subscriptions independently of workflow results.
 
 Logging responsibilities: PROTOCOL captures observed traffic, correlated by requestId rather than
-adjacent log entries; REMOTE_PROBLEM reports transport rejection; TASK_FAILED identifies the execution,
+adjacent log entries; A2A_ERROR reports a protocol rejection; TASK_FAILED identifies the execution,
 step, logical task, agent, code and reason; WORKFLOW_STOPPED lists unexecuted steps.
-Sample TASK_RESPONSE logs bridge contextId and executionId. Recognized remote problems use WARN summaries,
+Sample TASK_RESPONSE logs bridge contextId and executionId. Recognized A2A errors use WARN summaries,
 while unexpected exceptions retain ERROR stack traces. Known credentials and Bearer/Basic values are redacted.
 Logging configuration, pretty display and observer callbacks do not determine task outcomes.
 
diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md
index 749c06c..96b0c50 100644
--- a/docs/zh/API_REFERENCE.md
+++ b/docs/zh/API_REFERENCE.md
@@ -244,7 +244,7 @@ Map<String, Object> normalized = AgentCardNormalizer.normalize(rawMap);
 | `ProtocolResponses`           | A2A 事件与结果组装辅助方法 |
 | `ClientEventMapper`           | 面向回调和诊断的稳定事件投影 |
 | `WireLog`                     | 关联上下文与协议观测门面 |
-| `RemoteProblemException`      | 结构化远程问题响应（`status`、`title`、`detail`、`type`、`timestamp`） |
+| `RemoteA2AErrorException`     | 标准 A2A HTTP 错误投影（`code`、`status`、`message`、类型化详情及安全响应头） |
 
 ---
 
@@ -560,10 +560,10 @@ getReceivedMessages() 是保留层级的响应来源，getOutputs() 为便利投
 
 ## 错误处理
 
-- 顶层远端 problem 错误通过 `RemoteProblemException` 保留；可用 `findIn(Throwable)` 查找包装异常中的原因。
-  工作流任务失败信息为 `remote.problem.<status>`，详情进入 history / TASK_RESPONSE，不作为业务 outputs。
+- 顶层标准 A2A 错误信封通过 `RemoteA2AErrorException` 保留；`findIn(Throwable)` 也会投影包装的 SDK 错误。
+  工作流任务失败信息为 `a2a.<reason>` 或 `a2a.http.<status>`，详情进入 history / TASK_RESPONSE，不作为业务 outputs。
   引擎节点事件携带 executionId，任务结果事件和 history 携带逻辑 taskId。
-  失败传播、并行步骤失败规则和日志职责见 [集成指南](INTEGRATION_GUIDE.md#14-远端错误响应)。
+  失败传播、并行步骤失败规则和日志职责见 [集成指南](INTEGRATION_GUIDE.md#14-a2a-错误与任务失败)。
 
 - 回调异常／空值／超时明确失败；SDK 内容错误由宿主转换为 BusinessFailure，必要时保留安全的 code/details。
 - maxNegotiationExchanges 耗尽或本地 Stop 只结束本地交互，不自动发送 Abort。业务 Send(Abort) 不算任务成功。
@@ -621,7 +621,7 @@ a2at:
 | `eventBusProcessor` | `MainEventBusProcessor`       | 事件总线处理器                                         |
 | `requestHandler`    | `RequestHandler`              | 默认请求处理器                                         |
 | `restHandler`       | `RestHandler`                 | REST 协议处理器                                        |
-| `a2aController`     | `A2AController`               | Spring MVC 控制器（`message:send` + `message:stream`） |
+| `a2aController`     | `A2AController`               | 消息和任务端点的 Spring MVC 控制器                    |
 
 ### A2AController
 
@@ -629,6 +629,13 @@ a2at:
 
 - `POST {path-prefix}/message:send` — 阻塞式发送
 - `POST {path-prefix}/message:stream` — SSE 流式
+- `GET {path-prefix}/tasks/{id}` — 查询任务
+- `GET {path-prefix}/tasks` — 按可选条件分页查询任务
+- `POST {path-prefix}/tasks/{id}:cancel` — 取消任务
+- `POST {path-prefix}/tasks/{id}:subscribe` — 通过 SSE 订阅任务更新
+
+HTTP+JSON 响应使用 `application/a2a+json`，成功建立的流式响应使用 `text/event-stream`。连接结束时，控制器会取消
+Publisher 订阅并通知 SDK 停止事件消费，避免服务端轮询继续占用资源。
 
 ### 用法
 
diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md
index fdfe976..c0eceb1 100644
--- a/docs/zh/INTEGRATION_GUIDE.md
+++ b/docs/zh/INTEGRATION_GUIDE.md
@@ -514,30 +514,31 @@ Demo 在打开独立协议通道和启动工作流之前，先查询每个被调
 `A2A_TASK_CLEANUP_MAX_TASKS` 配置。任务查询受认证身份权限约束；若多个实例共用同一身份，Demo 可能取消其他实例创建的活跃任务。
 应使用隔离身份，或在提供等价的任务归属清理策略后关闭该功能。
 
-## 14. 远端错误响应
+## 14. A2A 错误与任务失败
 
-响应头成功不代表任务成功。HTTP 200 的 SSE data 也可能返回顶层错误对象，例如：
+任务创建前和创建后的失败必须分开处理。请求在任务创建前被拒绝时，返回非 2xx HTTP 状态和标准
+A2A `google.rpc.Status` JSON 错误信封：
 
 ```json
-{"status":429,"detail":"Current active tasks have reached the maximum limit 10.","type":""}
+{"error":{"code":400,"status":"INVALID_ARGUMENT","message":"缺少必填参数","details":[{"@type":"type.googleapis.com/google.rpc.ErrorInfo","reason":"INVALID_PARAMS","domain":"a2a-protocol.org","metadata":{"field":"port"}}]}}
 ```
 
-通信层识别顶层数字 status 为 400–599 且携带 title/detail 的错误对象，
-以 `RemoteProblemException` 保留 status、title、detail、type、timestamp。
-同步响应和流式响应均检查；普通任务内容或 artifact 内嵌的同名字段不被当成协议错误。
-该识别以 SDK 实际交付的正文为限：真实 HTTP 非 2xx 可能先被 SDK/适配器作为 HTTP 或认证错误拒绝，
-不保证可取得 problem 正文；没有该正文时仍走原有通用失败路径，不伪造 errorDetails。
-错误不生成成功结果、不触发 onNegotiation，也不自动重新提交请求。400 应核对业务参数；
-429 应检查被调度智能体的容量、活跃任务和订阅，再由业务决定是否重试。
-只有合法 Negotiation-T Propose 才进入协商回调。
-
-直接调用客户端时可用 `RemoteProblemException.findIn(error)` 检查 cause 链（没有则返回 null）；
-工作流执行时沿现有失败路径报告错误：
-任务结果及执行 history / TASK_RESPONSE 事件中的 errorCode 为 `remote.problem.400`、
-`remote.problem.429` 等，error 包含原因，errorDetails 包含上述五个字段，outputs 为空。
-错误内容中的已知凭据字段按协议日志的规则脱敏；未识别的 SDK 异常仍只暴露异常类型。
-预置授权、通知订阅的错误仍由其独立生命周期处理，不决定工作流执行结果。
-协议 pretty 日志仅用于展示，不改写原始收到的错误对象。
+`RemoteA2AErrorException` 保留实际观察到的 HTTP 状态、信封 code/status/message、类型化 details、
+ErrorInfo reason/domain 及安全响应头；`findIn(Throwable)` 也会投影 A2A Java SDK 的类型化错误。
+直连传输检查普通响应，并防御性识别被 SDK 作为 SSE data 暴露的建流前顶层错误信封。
+Task、Message 或 Artifact 内嵌的 `error` 对象仍属于业务内容，不会被误判为协议错误。
+
+若 SSE 调用未产生任何 A2A 事件便结束，传输会立即失败，不再等待工作流发送超时。底层 SDK 不暴露空流响应的 HTTP
+状态，因此该场景报告为传输/协议失败，不会虚构 HTTP 错误码。
+
+工作流 history 和 TASK_RESPONSE 使用 `a2a.invalid_params` 等稳定错误码；未知 HTTP 错误回退为
+`a2a.http.<status>`。`errorDetails` 保留协议事实及实际取得的 `retryAfter`。这类错误不生成成功输出、
+不触发 onNegotiation，也不自动重试。
+
+任务创建成功后的业务执行失败不是 HTTP 协议错误。被调度智能体返回 HTTP 200，并通过 Task 或
+StatusUpdate 的 `TASK_STATE_FAILED` 表示失败；TaskStatus 消息、任务 metadata 和 artifact 携带失败证据。
+引擎将工作流任务标记为失败，把证据保留在 `receivedMessages`，但不解释扩展协议的业务结果 schema。
+独立授权和通知订阅的失败仍按各自生命周期处理，不决定工作流结果。
 
 对于 `ALL_SUCCESS` 步骤，一个任务失败不会追溯取消已下发的并行任务；引擎在另一任务返回或超时后报告 `success=false`。后续步骤不执行，成功的并行结果保留在 history，不伪造整体成功结果。引擎不自动重试、排队或部分成功汇总；`ANY_SUCCESS` 节点仍遵循声明的任一成功语义。
 
@@ -547,9 +548,9 @@ Demo 在打开独立协议通道和启动工作流之前，先查询每个被调
 宿主智能体决定如何向调用方暴露失败，但不得把失败执行转换为成功 artifact。常驻宿主以独立生命周期管理通知订阅。
 
 **日志职责**：PROTOCOL 记录已观察到的请求/响应，按 requestId 关联并发报文；
-REMOTE_PROBLEM 记录通信层识别的远端拒绝，TASK_FAILED 记录执行 ID、节点、逻辑任务 ID、
+A2A_ERROR 记录协议拒绝，TASK_FAILED 记录执行 ID、节点、逻辑任务 ID、
 智能体、错误码和原因，WORKFLOW_STOPPED 列出未执行节点。示例 TASK_RESPONSE 日志同时包含
-contextId 与 executionId，供跨层定位。已识别的远端问题使用 WARN 摘要，
+contextId 与 executionId，供跨层定位。已识别的 A2A 错误使用 WARN 摘要，
 未知异常保留 ERROR 堆栈；错误详情中的已知凭据及 Bearer/Basic 值脱敏。
 日志开关、pretty 展示和观察回调不会改变任务结果。
 
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java b/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java
index 68a325b..e36de61 100644
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java
+++ b/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java
@@ -45,7 +45,9 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLParameters;
@@ -54,10 +56,12 @@
 import org.a2aproject.sdk.grpc.utils.ProtoUtils;
 import org.a2aproject.sdk.server.AgentCardCacheMetadata;
 import org.a2aproject.sdk.server.ServerCallContext;
+import org.a2aproject.sdk.server.auth.TaskOperation;
 import org.a2aproject.sdk.server.agentexecution.AgentExecutor;
 import org.a2aproject.sdk.server.events.InMemoryQueueManager;
 import org.a2aproject.sdk.server.events.MainEventBus;
 import org.a2aproject.sdk.server.events.MainEventBusProcessor;
+import org.a2aproject.sdk.server.extensions.A2AExtensions;
 import org.a2aproject.sdk.server.requesthandlers.DefaultRequestHandler;
 import org.a2aproject.sdk.server.requesthandlers.RequestHandler;
 import org.a2aproject.sdk.server.tasks.BasePushNotificationSender;
@@ -65,14 +69,20 @@
 import org.a2aproject.sdk.server.tasks.InMemoryTaskStore;
 import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore;
 import org.a2aproject.sdk.server.tasks.PushNotificationSender;
+import org.a2aproject.sdk.server.version.A2AVersionValidator;
 import org.a2aproject.sdk.spec.A2AError;
 import org.a2aproject.sdk.spec.AgentCard;
+import org.a2aproject.sdk.spec.InternalError;
+import org.a2aproject.sdk.spec.InvalidRequestError;
 import org.a2aproject.sdk.spec.Message;
 import org.a2aproject.sdk.spec.StreamingEventKind;
 import org.a2aproject.sdk.spec.Task;
 import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent;
+import org.a2aproject.sdk.spec.TaskIdParams;
 import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
+import org.a2aproject.sdk.spec.UnsupportedOperationError;
 import org.a2aproject.sdk.transport.rest.handler.RestHandler;
+import org.a2aproject.sdk.transport.rest.handler.RestHandler.HTTPRestResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -101,6 +111,8 @@ public class JdkHttpA2AServer implements AutoCloseable {
 
   private final ExecutorService agentExecutorService;
 
+  private final Thread eventBusThread;
+
   private final Map<String, Object> agentCardMap;
 
   private final String agentName;
@@ -141,7 +153,8 @@ public JdkHttpA2AServer(
               t.setDaemon(true);
               return t;
             });
-    RestHandler restHandler = initServerComponents(agentCard, agentExecutor, typedCard);
+    ServerComponents components = initServerComponents(agentExecutor, typedCard);
+    this.eventBusThread = components.eventBusThread();
     this.executorService =
         new ThreadPoolExecutor(
             THREAD_COUNT,
@@ -154,8 +167,8 @@ public JdkHttpA2AServer(
               t.setDaemon(true);
               return t;
             });
-    this.server = HttpsServer.create(new InetSocketAddress(host, port), 0);
     SSLContext sslContext = createSslContext();
+    this.server = HttpsServer.create(new InetSocketAddress(host, port), 0);
     this.server.setHttpsConfigurator(
         new HttpsConfigurator(sslContext) {
           @Override
@@ -166,13 +179,14 @@ public void configure(HttpsParameters params) {
         });
     this.server.setExecutor(executorService);
     this.server.createContext(
-        pathPrefix.isEmpty() ? "/" : pathPrefix, exchange -> handleExchange(exchange, restHandler));
+        pathPrefix.isEmpty() ? "/" : pathPrefix,
+        exchange -> handleExchange(exchange, components, typedCard));
     if (hasSecuritySchemes(agentCard)) {
       this.server.createContext("/rest/plat/smapp/v1/oauth/token", this::handleLogin);
       log.info("[{}] Auth login endpoint enabled", agentName);
     }
 
-    log.info("[{}] A2A server started on https://{}:{}/", agentName, host, port);
+    log.info("[{}] A2A server configured on https://{}:{}/", agentName, host, port);
   }
 
   private static String formatSse(long seq, String payload) {
@@ -209,28 +223,27 @@ private static StreamResponse toStreamResponse(StreamingEventKind event) {
     throw new IllegalArgumentException("Unsupported event: " + event);
   }
 
-  private static RequestHandler extractRequestHandler(RestHandler restHandler) {
-    try {
-      var f = RestHandler.class.getDeclaredField("requestHandler");
-      f.setAccessible(true);
-      return (RequestHandler) f.get(restHandler);
-    } catch (ReflectiveOperationException e) {
-      throw new IllegalStateException("Failed to access request handler", e);
-    }
-  }
-
   private static ServerCallContext buildCallContext(HttpExchange exchange) {
     Map<String, String> headers = new LinkedHashMap<>();
     exchange
         .getRequestHeaders()
         .forEach((k, v) -> headers.put(k.toLowerCase(Locale.ROOT), String.join(",", v)));
     String ext = firstHeader(headers, "A2A-Extensions", "X-A2A-Extensions");
-    Set<String> exts = ext.isBlank() ? Set.of() : Set.of(ext.split(","));
+    Set<String> exts;
+    if (ext.isBlank()) {
+      exts = Set.of();
+    } else {
+      java.util.LinkedHashSet<String> parsed = new java.util.LinkedHashSet<>();
+      for (String value : ext.split(",")) {
+        if (!value.isBlank()) parsed.add(value.strip());
+      }
+      exts = java.util.Collections.unmodifiableSet(parsed);
+    }
     return new ServerCallContext(
         null,
         Map.of("headers", headers),
         exts,
-        firstNullableHeader(headers, "A2A-Protocol-Version", "A2A-Version"));
+        firstNullableHeader(headers, "A2A-Version", "A2A-Protocol-Version"));
   }
 
   private static String firstHeader(Map<String, String> h, String a, String b) {
@@ -257,6 +270,29 @@ private static void sendJson(HttpExchange exchange, int status, Object data) thr
     }
   }
 
+  private static void sendRestResponse(HttpExchange exchange, HTTPRestResponse response)
+      throws IOException {
+    byte[] bytes = response.getBody().getBytes(StandardCharsets.UTF_8);
+    exchange
+        .getResponseHeaders()
+        .set("Content-Type", "application/a2a+json");
+    exchange.sendResponseHeaders(response.getStatusCode(), bytes.length);
+    try (OutputStream os = exchange.getResponseBody()) {
+      os.write(bytes);
+    }
+  }
+
+  private static void sendA2AError(
+      HttpExchange exchange, RestHandler restHandler, A2AError error) throws IOException {
+    var response = restHandler.createErrorResponse(error);
+    byte[] bytes = response.getBody().getBytes(StandardCharsets.UTF_8);
+    exchange.getResponseHeaders().set("Content-Type", "application/a2a+json");
+    exchange.sendResponseHeaders(response.getStatusCode(), bytes.length);
+    try (OutputStream os = exchange.getResponseBody()) {
+      os.write(bytes);
+    }
+  }
+
   private static AgentCard toTypedAgentCard(Map<String, Object> card) {
     return cardMapper.convertValue(card, AgentCard.class);
   }
@@ -288,16 +324,6 @@ private static String extractPathPrefix(Map<String, Object> agentCard) {
     return "";
   }
 
-  private static void startEventBus(MainEventBusProcessor proc) {
-    try {
-      var m = MainEventBusProcessor.class.getDeclaredMethod("start");
-      m.setAccessible(true);
-      m.invoke(proc);
-    } catch (ReflectiveOperationException e) {
-      throw new IllegalStateException("Failed to start event bus", e);
-    }
-  }
-
   /** Parses exactly one encoded task-id path segment and rejects ambiguous task routes. */
   static String taskIdFromPath(String path, String operationSuffix) {
     String prefix = "/tasks/";
@@ -345,15 +371,31 @@ private SSLContext createSslContext() throws IOException {
   }
 
   public void start() {
-    server.start();
+    eventBusThread.start();
+    try {
+      server.start();
+      log.info(
+          "[{}] A2A server started on https://{}:{}/",
+          agentName,
+          server.getAddress().getHostString(),
+          server.getAddress().getPort());
+    } catch (RuntimeException error) {
+      eventBusThread.interrupt();
+      throw error;
+    }
   }
 
   @Override
   public void close() {
     server.stop(0);
+    eventBusThread.interrupt();
     agentExecutorService.shutdownNow();
     executorService.shutdownNow();
     try {
+      eventBusThread.join(1_000);
+      if (eventBusThread.isAlive()) {
+        log.warn("[{}] Event bus processor did not stop within 1 second", agentName);
+      }
       agentExecutorService.awaitTermination(1, TimeUnit.SECONDS);
       executorService.awaitTermination(1, TimeUnit.SECONDS);
     } catch (InterruptedException e) {
@@ -370,8 +412,11 @@ public String getAgentName() {
     return agentName;
   }
 
-  private RestHandler initServerComponents(
-      Map<String, Object> agentCard, AgentExecutor agentExecutor, AgentCard typedCard) {
+  boolean isEventBusProcessorAlive() {
+    return eventBusThread.isAlive();
+  }
+
+  private ServerComponents initServerComponents(AgentExecutor agentExecutor, AgentCard typedCard) {
     InMemoryTaskStore taskStore = new InMemoryTaskStore();
     MainEventBus mainEventBus = new MainEventBus();
     InMemoryQueueManager queueManager = new InMemoryQueueManager(taskStore, mainEventBus);
@@ -379,7 +424,11 @@ private RestHandler initServerComponents(
     PushNotificationSender pushSender = new BasePushNotificationSender(pushStore);
     MainEventBusProcessor eventBusProc =
         new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager);
-    startEventBus(eventBusProc);
+    // Manual wiring: no CDI/Spring container invokes the @PostConstruct start(), and
+    // ensureStarted() is intentionally empty in the SDK (it only forces CDI proxy resolution).
+    // Without this thread the event bus never drains and inbound tasks hang forever.
+    Thread eventBusThread = new Thread(eventBusProc, "MainEventBusProcessor");
+    eventBusThread.setDaemon(true);
     RequestHandler requestHandler =
         DefaultRequestHandler.builder()
             .agentExecutor(agentExecutor)
@@ -390,11 +439,18 @@ private RestHandler initServerComponents(
             .executor(agentExecutorService)
             .eventConsumerExecutor(agentExecutorService)
             .build();
-    return new RestHandler(
-        typedCard, new AgentCardCacheMetadata(typedCard, null), requestHandler, Runnable::run);
+    RestHandler restHandler =
+        new RestHandler(
+            typedCard, new AgentCardCacheMetadata(typedCard, null), requestHandler, Runnable::run);
+    return new ServerComponents(restHandler, requestHandler, eventBusThread);
   }
 
-  private void handleExchange(HttpExchange exchange, RestHandler restHandler) throws IOException {
+  private record ServerComponents(
+      RestHandler restHandler, RequestHandler requestHandler, Thread eventBusThread) {}
+
+  private void handleExchange(
+      HttpExchange exchange, ServerComponents components, AgentCard typedCard) throws IOException {
+    RestHandler restHandler = components.restHandler();
     String fullPath = exchange.getRequestURI().getRawPath();
     String method = exchange.getRequestMethod();
     String path =
@@ -410,13 +466,18 @@ private void handleExchange(HttpExchange exchange, RestHandler restHandler) thro
 
       if ("POST".equalsIgnoreCase(method) && "/message:send".equals(path)) {
         var resp = restHandler.sendMessage(buildCallContext(exchange), "", readBody(exchange));
-        sendJson(exchange, resp.getStatusCode(), resp.getBody());
+        sendRestResponse(exchange, resp);
         return;
       }
 
       if ("POST".equalsIgnoreCase(method) && "/message:stream".equals(path)) {
 
-        handleStream(exchange, restHandler, readBody(exchange));
+        handleStream(
+            exchange,
+            restHandler,
+            components.requestHandler(),
+            typedCard,
+            readBody(exchange));
         return;
       }
       if ("GET".equalsIgnoreCase(method) && "/tasks".equals(path)) {
@@ -432,34 +493,36 @@ private void handleExchange(HttpExchange exchange, RestHandler restHandler) thro
                 integerParam(query, "historyLength"),
                 query.get("statusTimestampAfter"),
                 booleanParam(query, "includeArtifacts"));
-        sendJson(exchange, resp.getStatusCode(), resp.getBody());
+        sendRestResponse(exchange, resp);
         return;
       }
       // Task routes (?? A2A: ??/??/??)
       String taskId = taskIdFromPath(path, null);
       if ("GET".equalsIgnoreCase(method) && taskId != null) {
         var resp = restHandler.getTask(buildCallContext(exchange), "", taskId, null);
-        sendJson(exchange, resp.getStatusCode(), resp.getBody());
+        sendRestResponse(exchange, resp);
         return;
       }
       taskId = taskIdFromPath(path, ":cancel");
       if ("POST".equalsIgnoreCase(method) && taskId != null) {
         var resp =
             restHandler.cancelTask(buildCallContext(exchange), "", readBody(exchange), taskId);
-        sendJson(exchange, resp.getStatusCode(), resp.getBody());
+        sendRestResponse(exchange, resp);
         return;
       }
       taskId = taskIdFromPath(path, ":subscribe");
       if ("POST".equalsIgnoreCase(method) && taskId != null) {
-        handleSubscribe(exchange, restHandler, taskId);
+        handleSubscribe(exchange, restHandler, components.requestHandler(), typedCard, taskId);
         return;
       }
       exchange.sendResponseHeaders(404, -1);
+    } catch (A2AError e) {
+      sendA2AError(exchange, restHandler, e);
     } catch (IllegalArgumentException e) {
-      sendJson(exchange, 400, Map.of("error", Map.of("code", 400, "message", e.getMessage())));
+      sendA2AError(exchange, restHandler, new InvalidRequestError("Invalid A2A request"));
     } catch (Exception e) {
-      log.error("[{}] Handler error: {}", agentName, e.getMessage(), e);
-      sendJson(exchange, 500, Map.of("error", Map.of("code", 500, "message", e.getMessage())));
+      log.error("[{}] Handler error: {}", agentName, e.getClass().getSimpleName(), e);
+      sendA2AError(exchange, restHandler, new InternalError("A2A request handling failed"));
     } finally {
       exchange.close();
     }
@@ -523,120 +586,156 @@ private void handleLogin(HttpExchange exchange) throws IOException {
   }
 
   @SuppressWarnings("unchecked")
-  private void handleSubscribe(HttpExchange exchange, RestHandler restHandler, String taskId)
+  private void handleSubscribe(
+      HttpExchange exchange,
+      RestHandler restHandler,
+      RequestHandler requestHandler,
+      AgentCard typedCard,
+      String taskId)
       throws IOException {
     try {
-      var resp = restHandler.subscribeToTask(buildCallContext(exchange), "", taskId);
-      if (resp instanceof RestHandler.HTTPRestStreamingResponse streaming) {
-        exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
-        exchange.sendResponseHeaders(200, 0);
-        AtomicLong seq = new AtomicLong(0);
-        CountDownLatch done = new CountDownLatch(1);
-        OutputStream os = exchange.getResponseBody();
-        streaming
-            .getPublisher()
-            .subscribe(
-                new Flow.Subscriber<String>() {
-                  private Flow.Subscription sub;
-
-                  public void onSubscribe(Flow.Subscription s) {
-                    sub = s;
-                    s.request(1);
-                  }
-
-                  public void onNext(String item) {
-                    try {
-                      String sse =
-                          String.format(Locale.ROOT, "id:%d%n", seq.incrementAndGet())
-                              + "data:"
-                              + item
-                              + "\n\n";
-                      os.write(sse.getBytes(StandardCharsets.UTF_8));
-                      os.flush();
-                      sub.request(1);
-                    } catch (IOException e) {
-                      sub.cancel();
-                      done.countDown();
-                    }
-                  }
-
-                  public void onError(Throwable t) {
-                    done.countDown();
-                  }
-
-                  public void onComplete() {
-                    done.countDown();
-                  }
-                });
-        try {
-          done.await();
-        } catch (InterruptedException e) {
-          Thread.currentThread().interrupt();
-        } finally {
-          os.close();
-        }
-      } else {
-        sendJson(exchange, resp.getStatusCode(), resp.getBody());
-      }
+      requireStreamingSupport(typedCard);
+      ServerCallContext context = buildCallContext(exchange);
+      requestHandler.authorizeTaskAccess(taskId, context, TaskOperation.SUBSCRIBE_TO_TASK);
+      Flow.Publisher<StreamingEventKind> publisher =
+          requestHandler.onSubscribeToTask(
+              TaskIdParams.builder().id(taskId).tenant("").build(), context);
+      writeStream(exchange, publisher, context);
     } catch (A2AError e) {
-      sendJson(exchange, 500, Map.of("error", e.getMessage()));
+      sendA2AError(exchange, restHandler, e);
     }
   }
 
   @SuppressWarnings("unchecked")
-  private void handleStream(HttpExchange exchange, RestHandler restHandler, String requestBody)
+  private void handleStream(
+      HttpExchange exchange,
+      RestHandler restHandler,
+      RequestHandler requestHandler,
+      AgentCard typedCard,
+      String requestBody)
       throws IOException {
     try {
-      RequestHandler inner = extractRequestHandler(restHandler);
+      requireStreamingSupport(typedCard);
+      ServerCallContext context = buildCallContext(exchange);
+      A2AVersionValidator.validateProtocolVersion(typedCard, context);
+      A2AExtensions.validateRequiredExtensions(typedCard, context);
       SendMessageRequest.Builder builder = SendMessageRequest.newBuilder();
       JsonFormat.parser().merge(requestBody, builder);
       var request = ProtoUtils.FromProto.messageSendParams(builder.build());
+      requestHandler.authorizeTaskAccess(
+          request.message().taskId(), context, TaskOperation.MESSAGE_SEND_STREAM);
       Flow.Publisher<StreamingEventKind> publisher =
-          inner.onMessageSendStream(request, buildCallContext(exchange));
-      exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
-      exchange.sendResponseHeaders(200, 0);
-      AtomicLong seq = new AtomicLong(0);
-      CountDownLatch done = new CountDownLatch(1);
-      OutputStream os = exchange.getResponseBody();
-      publisher.subscribe(
-          new Flow.Subscriber<>() {
-            private Flow.Subscription sub;
-
-            public void onSubscribe(Flow.Subscription s) {
-              sub = s;
-              s.request(1);
-            }
+          requestHandler.onMessageSendStream(request, context);
+      writeStream(exchange, publisher, context);
+    } catch (A2AError e) {
+      sendA2AError(exchange, restHandler, e);
+    } catch (com.google.protobuf.InvalidProtocolBufferException e) {
+      sendA2AError(exchange, restHandler, new InvalidRequestError("Invalid A2A request body"));
+    }
+  }
 
-            public void onNext(StreamingEventKind item) {
-              try {
-                String payload = JsonFormat.printer().print(toStreamResponse(item));
-                os.write(
-                    formatSse(seq.incrementAndGet(), payload).getBytes(StandardCharsets.UTF_8));
-                os.flush();
-                sub.request(1);
-              } catch (IOException e) {
-                sub.cancel();
-                done.countDown();
-              }
-            }
+  private static void requireStreamingSupport(AgentCard agentCard)
+      throws UnsupportedOperationError {
+    if (!agentCard.capabilities().streaming()) {
+      throw new UnsupportedOperationError(null, "Streaming is not supported by the agent", null);
+    }
+  }
 
-            public void onError(Throwable t) {
-              done.countDown();
-            }
+  private static void writeStream(
+      HttpExchange exchange,
+      Flow.Publisher<StreamingEventKind> publisher,
+      ServerCallContext context)
+      throws IOException {
+    exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
+    exchange.sendResponseHeaders(200, 0);
+    AtomicLong seq = new AtomicLong(0);
+    CountDownLatch done = new CountDownLatch(1);
+    OutputStream output = exchange.getResponseBody();
+    StreamCancellation cancellation = new StreamCancellation(context);
+    publisher.subscribe(
+        new Flow.Subscriber<>() {
+          @Override
+          public void onSubscribe(Flow.Subscription subscription) {
+            cancellation.register(subscription);
+          }
 
-            public void onComplete() {
+          @Override
+          public void onNext(StreamingEventKind item) {
+            try {
+              String payload = JsonFormat.printer().print(toStreamResponse(item));
+              output.write(
+                  formatSse(seq.incrementAndGet(), payload).getBytes(StandardCharsets.UTF_8));
+              output.flush();
+              cancellation.requestNext();
+            } catch (IOException error) {
+              cancellation.release();
               done.countDown();
             }
-          });
+          }
+
+          @Override
+          public void onError(Throwable error) {
+            cancellation.release();
+            done.countDown();
+          }
+
+          @Override
+          public void onComplete() {
+            cancellation.release();
+            done.countDown();
+          }
+        });
+    try {
+      done.await();
+    } catch (InterruptedException error) {
+      Thread.currentThread().interrupt();
+    } finally {
+      cancellation.release();
+      output.close();
+    }
+  }
+
+  private static final class StreamCancellation {
+    private final ServerCallContext context;
+    private final AtomicReference<Flow.Subscription> subscription = new AtomicReference<>();
+    private final AtomicBoolean released = new AtomicBoolean();
+
+    private StreamCancellation(ServerCallContext context) {
+      this.context = context;
+    }
+
+    private void register(Flow.Subscription candidate) {
+      if (!subscription.compareAndSet(null, candidate) || released.get()) {
+        candidate.cancel();
+      } else {
+        candidate.request(1);
+      }
+    }
+
+    private void requestNext() {
+      Flow.Subscription active = subscription.get();
+      if (active != null && !released.get()) {
+        active.request(1);
+      }
+    }
+
+    private void release() {
+      if (!released.compareAndSet(false, true)) {
+        return;
+      }
+      Flow.Subscription active = subscription.get();
+      if (active != null) {
+        try {
+          active.cancel();
+        } catch (RuntimeException error) {
+          log.warn("[SSE] Failed to cancel publisher subscription: {}", error.getMessage());
+        }
+      }
       try {
-        done.await();
-      } catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-      } finally {
-        os.close();
+        context.invokeEventConsumerCancelCallback();
+      } catch (RuntimeException error) {
+        log.warn("[SSE] Failed to notify SDK event consumer cancellation: {}", error.getMessage());
       }
-    } catch (A2AError e) {
-      sendJson(exchange, 500, Map.of("error", e.getMessage()));
     }
   }
 }
diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java
index 04edc60..a9369b4 100644
--- a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java
+++ b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java
@@ -53,7 +53,8 @@ protected String[] arguments(String city1, String city2) throws Exception {
   @ParameterizedTest
   @ValueSource(ints = {400, 429})
   @org.junit.jupiter.api.Timeout(60)
-  void remoteProblemReachesNorthboundFailureWithoutMergeOrResubmission(int status) throws Exception {
+  void standardA2AErrorReachesNorthboundFailureWithoutMergeOrResubmission(int status)
+      throws Exception {
     var firstCalls = new AtomicInteger();
     var secondCalls = new AtomicInteger();
     HttpServer city1 = omc(status, firstCalls);
@@ -70,7 +71,7 @@ void remoteProblemReachesNorthboundFailureWithoutMergeOrResubmission(int status)
       var error = assertThrows(IllegalStateException.class,
           () -> new SpringSpnDemo().run(arguments(first, second)));
       assertTrue(error.getMessage().contains("TASK_STATE_FAILED"), error.getMessage());
-      assertTrue(error.getMessage().contains("remote.problem." + status), error.getMessage());
+      assertTrue(error.getMessage().contains(errorCode(status)), error.getMessage());
       assertTrue(error.getMessage().contains(reason(status)), error.getMessage());
       assertTrue(error.getMessage().contains("diagnosis_city2"));
       assertEquals(1, firstCalls.get());
@@ -79,11 +80,14 @@ void remoteProblemReachesNorthboundFailureWithoutMergeOrResubmission(int status)
       assertTrue(evidence.contains("OPERATION_FAILED extension=Authorization-T"));
       assertTrue(evidence.contains("OPERATION_FAILED extension=Notification-T"));
       assertTrue(evidence.contains("[Executor] TASK_FAILED executionId="));
-      assertTrue(evidence.contains("errorCode=remote.problem." + status));
+      assertTrue(evidence.contains("errorCode=" + errorCode(status)));
       assertTrue(evidence.contains("skippedSteps=[merge_analysis]"));
       assertTrue(evidence.contains("[Orchestrator] WORKFLOW_FINISH success=false, history=2"));
-      assertTrue(evidence.contains("=== SSE data (JSON display; not wire text) ==="));
-      assertTrue(evidence.contains("\"status\": " + status));
+      assertTrue(evidence.contains("\"error\""));
+      // The projected errorDetails keep the observed numeric HTTP status; the envelope's
+      // "status" field is the canonical string (INVALID_ARGUMENT), not the number.
+      assertTrue(evidence.contains("\"httpStatus\":" + status));
+      assertTrue(evidence.contains("\"status\":\"" + canonicalStatus(status) + "\""));
       assertTrue(evidence.contains("[SpringWorkbench] TASK_FAILED"));
       assertTrue(evidence.contains("[Demo] SHUTDOWN_DONE success=false"));
       assertFalse(evidence.contains("[ERROR] null"));
@@ -128,6 +132,30 @@ private static String reason(int status) {
         : "Current active tasks have reached the maximum limit 10. Please try again later.";
   }
 
+  private static String errorCode(int status) {
+    return status == 400 ? "a2a.invalid_params" : "a2a.active_task_limit_exceeded";
+  }
+
+  private static String canonicalStatus(int status) {
+    return status == 400 ? "INVALID_ARGUMENT" : "RESOURCE_EXHAUSTED";
+  }
+
+  private static String errorResponse(int status, String message) throws java.io.IOException {
+    String canonical = status == 400 ? "INVALID_ARGUMENT"
+        : status == 429 ? "RESOURCE_EXHAUSTED" : "UNAVAILABLE";
+    String errorReason = status == 400 ? "INVALID_PARAMS"
+        : status == 429 ? "ACTIVE_TASK_LIMIT_EXCEEDED" : "OPERATION_REJECTED";
+    String domain = status == 400 ? "a2a-protocol.org" : "fixture.invalid";
+    return JSON.writeValueAsString(Map.of("error", Map.of(
+        "code", status,
+        "status", canonical,
+        "message", message,
+        "details", List.of(Map.of(
+            "@type", "type.googleapis.com/google.rpc.ErrorInfo",
+            "reason", errorReason,
+            "domain", domain)))));
+  }
+
   private static HttpServer omc(int status, AtomicInteger calls) throws Exception {
     HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
     server.createContext("/", exchange -> {
@@ -150,11 +178,10 @@ private static HttpServer omc(int status, AtomicInteger calls) throws Exception
       String response;
       if (!task) {
         // Deliberate pre-position failures must not prevent the subsequent workflow tasks.
-        response = JSON.writeValueAsString(Map.of("status", 503, "detail", "Independent operation rejected"));
+        response = errorResponse(503, "Independent operation rejected");
       } else if (status != 0) {
         calls.incrementAndGet();
-        response = JSON.writeValueAsString(Map.of(
-            "status", status, "detail", reason(status), "type", "", "timestamp", "2026-08-31T09:07:35Z"));
+        response = errorResponse(status, reason(status));
       } else {
         calls.incrementAndGet();
         response = JSON.writeValueAsString(Map.of("task", Map.of(
@@ -162,9 +189,13 @@ private static HttpServer omc(int status, AtomicInteger calls) throws Exception
             "status", Map.of("state", "TASK_STATE_COMPLETED"),
             "artifacts", List.of(Map.of("artifactId", "diagnosis", "parts", List.of(Map.of("text", "City2 diagnosis")))))));
       }
-      byte[] bytes = ("data: " + response + "\n\n").getBytes(StandardCharsets.UTF_8);
-      exchange.getResponseHeaders().set("Content-Type", "text/event-stream;charset=UTF-8");
-      exchange.sendResponseHeaders(200, 0);
+      boolean failed = !task || status != 0;
+      byte[] bytes = (failed ? response : "data: " + response + "\n\n")
+          .getBytes(StandardCharsets.UTF_8);
+      exchange.getResponseHeaders().set(
+          "Content-Type", failed ? "application/a2a+json" : "text/event-stream;charset=UTF-8");
+      if (status == 429) exchange.getResponseHeaders().set("Retry-After", "5");
+      exchange.sendResponseHeaders(failed ? (!task ? 503 : status) : 200, bytes.length);
       try (var output = exchange.getResponseBody()) { output.write(bytes); }
       finally { exchange.close(); }
     });
diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java
index 1fd25f9..0ee4190 100644
--- a/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java
+++ b/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java
@@ -200,6 +200,101 @@ void testGetAgentCard() throws Exception {
     assertEquals(AGENT_NAME, card.get("name"));
   }
 
+  @Test
+  void closeStopsTheManuallyManagedEventBusProcessor() {
+    assertTrue(server.isEventBusProcessorAlive());
+
+    server.close();
+
+    assertFalse(server.isEventBusProcessorAlive());
+    server = null;
+  }
+
+  @Test
+  void nonStreamingTaskErrorUsesA2AJsonMediaType() throws Exception {
+    HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build();
+    HttpRequest req =
+        HttpRequest.newBuilder()
+            .uri(URI.create("https://127.0.0.1:" + port + "/tasks/missing"))
+            .GET()
+            .build();
+
+    HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());
+
+    assertEquals(404, resp.statusCode());
+    assertEquals(
+        "application/a2a+json", resp.headers().firstValue("Content-Type").orElseThrow());
+  }
+
+  @Test
+  void malformedStreamingRequestReturnsAStandardA2AErrorBeforeSseStarts() throws Exception {
+    HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build();
+    HttpRequest req =
+        HttpRequest.newBuilder()
+            .uri(URI.create("https://127.0.0.1:" + port + "/message:stream"))
+            .header("Content-Type", "application/a2a+json")
+            .header("A2A-Version", "1.0")
+            .POST(HttpRequest.BodyPublishers.ofString("{"))
+            .build();
+    HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());
+    assertEquals(400, resp.statusCode());
+    assertEquals(
+        "application/a2a+json",
+        resp.headers().firstValue("Content-Type").orElseThrow());
+    var root = mapper.readTree(resp.body());
+    assertEquals(400, root.path("error").path("code").asInt());
+    assertEquals("INVALID_ARGUMENT", root.path("error").path("status").asText());
+    assertTrue(root.path("error").path("details").isArray());
+    assertFalse(root.path("error").path("details").isEmpty());
+    assertEquals(
+        "type.googleapis.com/google.rpc.ErrorInfo",
+        root.path("error").path("details").get(0).path("@type").asText());
+  }
+
+  @Test
+  void unsupportedProtocolVersionReturnsAStandardA2AErrorBeforeSseStarts() throws Exception {
+    HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build();
+    HttpRequest req =
+        HttpRequest.newBuilder()
+            .uri(URI.create("https://127.0.0.1:" + port + "/message:stream"))
+            .header("Content-Type", "application/a2a+json")
+            .header("A2A-Version", "2.0")
+            .POST(HttpRequest.BodyPublishers.ofString("{}"))
+            .build();
+
+    HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());
+
+    assertEquals(400, resp.statusCode());
+    assertEquals(
+        "application/a2a+json", resp.headers().firstValue("Content-Type").orElseThrow());
+    var root = mapper.readTree(resp.body());
+    assertEquals("UNIMPLEMENTED", root.path("error").path("status").asText());
+    assertEquals(
+        "VERSION_NOT_SUPPORTED",
+        root.path("error").path("details").get(0).path("reason").asText());
+  }
+
+  @Test
+  void missingTaskSubscriptionReturnsAStandardA2AErrorBeforeSseStarts() throws Exception {
+    HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build();
+    HttpRequest req =
+        HttpRequest.newBuilder()
+            .uri(URI.create("https://127.0.0.1:" + port + "/tasks/missing:subscribe"))
+            .POST(HttpRequest.BodyPublishers.noBody())
+            .build();
+
+    HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());
+
+    assertEquals(404, resp.statusCode());
+    assertEquals(
+        "application/a2a+json", resp.headers().firstValue("Content-Type").orElseThrow());
+    var root = mapper.readTree(resp.body());
+    assertEquals(404, root.path("error").path("code").asInt());
+    assertEquals("NOT_FOUND", root.path("error").path("status").asText());
+    assertEquals(
+        "TASK_NOT_FOUND", root.path("error").path("details").get(0).path("reason").asText());
+  }
+
   @Test
   void testSendMessage() throws Exception {
     SendMessageResult result =
diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java
index 6fe016f..379a95c 100644
--- a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java
+++ b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java
@@ -51,17 +51,17 @@ void failureSummaryPreservesEachCityReasonWithoutPassingPartialOutputsAsDiagnosi
                         "status",
                         "failed",
                         "errorCode",
-                        "remote.problem.429",
+                        "a2a.active_task_limit_exceeded",
                         "error",
                         "OMC capacity reached",
                         "errorDetails",
                         Map.of(
-                            "status",
+                            "httpStatus",
                             429,
-                            "detail",
-                            "capacity reached",
-                            "type",
-                            "",
+                            "status",
+                            "RESOURCE_EXHAUSTED",
+                            "reason",
+                            "ACTIVE_TASK_LIMIT_EXCEEDED",
                             "accessSession",
                             "private-token")),
                     Map.of(
@@ -76,8 +76,8 @@ void failureSummaryPreservesEachCityReasonWithoutPassingPartialOutputsAsDiagnosi
             .build();
     String summary = WorkbenchOrchestrator.buildResultText(result);
     assertTrue(summary.contains("city1"));
-    assertTrue(summary.contains("remote.problem.429"));
-    assertTrue(summary.contains("capacity reached"));
+    assertTrue(summary.contains("a2a.active_task_limit_exceeded"));
+    assertTrue(summary.contains("RESOURCE_EXHAUSTED"));
     assertTrue(summary.contains("city2"));
     org.junit.jupiter.api.Assertions.assertFalse(summary.contains("private-token"));
     org.junit.jupiter.api.Assertions.assertFalse(summary.contains("partial diagnosis"));
diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java
index 2688cd2..03514a6 100644
--- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java
+++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java
@@ -176,8 +176,7 @@ public MainEventBusProcessor eventBusProcessor(
       PushNotificationConfigStore pushStore) {
     PushNotificationSender sender = new BasePushNotificationSender(pushStore);
     MainEventBusProcessor proc = new MainEventBusProcessor(bus, store, sender, qm);
-    proc.ensureStarted();
-    log.info("[A2A] MainEventBusProcessor started");
+    log.info("[A2A] MainEventBusProcessor configured; Spring manages its lifecycle");
     return proc;
   }
 
@@ -211,7 +210,8 @@ public RestHandler restHandler(AgentCard card, RequestHandler handler, ExecutorS
 
   @Bean
   @ConditionalOnMissingBean
-  public A2AController a2aController(RestHandler restHandler, RequestHandler requestHandler) {
-    return new A2AController(restHandler, requestHandler);
+  public A2AController a2aController(
+      RestHandler restHandler, RequestHandler requestHandler, AgentCard agentCard) {
+    return new A2AController(restHandler, requestHandler, agentCard);
   }
 }
diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java
index 132242e..d5ccc15 100644
--- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java
+++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java
@@ -31,18 +31,29 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import org.a2aproject.sdk.grpc.SendMessageRequest;
 import org.a2aproject.sdk.grpc.StreamResponse;
 import org.a2aproject.sdk.grpc.utils.ProtoUtils;
 import org.a2aproject.sdk.server.ServerCallContext;
+import org.a2aproject.sdk.server.auth.TaskOperation;
+import org.a2aproject.sdk.server.extensions.A2AExtensions;
 import org.a2aproject.sdk.server.requesthandlers.RequestHandler;
+import org.a2aproject.sdk.server.version.A2AVersionValidator;
+import org.a2aproject.sdk.spec.A2AError;
+import org.a2aproject.sdk.spec.AgentCard;
+import org.a2aproject.sdk.spec.InternalError;
+import org.a2aproject.sdk.spec.InvalidRequestError;
 import org.a2aproject.sdk.spec.MessageSendParams;
 import org.a2aproject.sdk.spec.StreamingEventKind;
+import org.a2aproject.sdk.spec.TaskIdParams;
+import org.a2aproject.sdk.spec.UnsupportedOperationError;
 import org.a2aproject.sdk.transport.rest.handler.RestHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -69,94 +80,149 @@
  * <p>The path prefix is configurable via {@code a2at.server.path-prefix}. The controller delegates
  * to {@link RestHandler} (non-streaming) and {@link RequestHandler} (streaming). The {@link
  * ServerCallContext} is built from {@link HttpServletRequest} headers, preserving the
- * A2A-Extensions header.
+ * A2A-Extensions header. Streaming endpoints return {@link SseEmitter}; a request rejected before
+ * the stream is created throws {@link A2AError}, which {@link #a2aError(A2AError)} maps to a
+ * non-2xx response with the standard A2A error envelope.
  */
 @RestController
 public class A2AController {
 
   private static final Logger log = LoggerFactory.getLogger(A2AController.class);
+  private static final MediaType A2A_JSON = MediaType.parseMediaType("application/a2a+json");
 
   private final RestHandler restHandler;
   private final RequestHandler requestHandler;
+  private final AgentCard agentCard;
   private final AtomicInteger activeStreams = new AtomicInteger();
   private final Object streamCompletionMonitor = new Object();
 
-  public A2AController(RestHandler restHandler, RequestHandler requestHandler) {
+  public A2AController(
+      RestHandler restHandler, RequestHandler requestHandler, AgentCard agentCard) {
     this.restHandler = restHandler;
     this.requestHandler = requestHandler;
+    this.agentCard = agentCard;
   }
 
   @PostMapping("${a2at.server.path-prefix}/message:send")
   public ResponseEntity<String> sendMessage(HttpServletRequest req, @RequestBody String body) {
     var ctx = buildContext(req);
     var resp = restHandler.sendMessage(ctx, "", body);
-    return ResponseEntity.status(resp.getStatusCode())
-        .contentType(
-            MediaType.parseMediaType(
-                resp.getContentType() != null
-                    ? resp.getContentType()
-                    : MediaType.APPLICATION_JSON_VALUE))
-        .body(resp.getBody());
+    return toResponse(resp);
   }
 
   @PostMapping(
       value = "${a2at.server.path-prefix}/message:stream",
       produces = MediaType.TEXT_EVENT_STREAM_VALUE)
   public SseEmitter streamMessage(HttpServletRequest req, @RequestBody String body) {
-    SseEmitter emitter = track(new SseEmitter(0L));
     try {
+      var ctx = buildContext(req);
+      requireStreamingSupport();
+      A2AVersionValidator.validateProtocolVersion(agentCard, ctx);
+      A2AExtensions.validateRequiredExtensions(agentCard, ctx);
+
       SendMessageRequest.Builder builder = SendMessageRequest.newBuilder();
       JsonFormat.parser().merge(body, builder);
       MessageSendParams params = ProtoUtils.FromProto.messageSendParams(builder.build());
-
-      var ctx = buildContext(req);
+      requestHandler.authorizeTaskAccess(
+          params.message().taskId(), ctx, TaskOperation.MESSAGE_SEND_STREAM);
       Flow.Publisher<StreamingEventKind> publisher =
           requestHandler.onMessageSendStream(params, ctx);
+      return subscribeToEmitter(publisher, ctx);
+    } catch (A2AError e) {
+      log.warn("[SSE] A2A request rejected before stream creation: {}", e.getMessage());
+      throw e;
+    } catch (com.google.protobuf.InvalidProtocolBufferException e) {
+      log.warn("[SSE] Invalid A2A request body: {}", e.getMessage());
+      throw new InvalidRequestError("Invalid A2A request body");
+    } catch (Exception e) {
+      log.error("[SSE] Setup failed before stream creation: {}", e.getClass().getSimpleName(), e);
+      throw new InternalError("Streaming request setup failed");
+    }
+  }
 
-      final AtomicLong seq = new AtomicLong(0);
-      publisher.subscribe(
-          new Flow.Subscriber<>() {
-            private Flow.Subscription sub;
-
-            @Override
-            public void onSubscribe(Flow.Subscription s) {
-              sub = s;
-              s.request(1);
-            }
+  /**
+   * Maps a pre-stream {@link A2AError} to a non-2xx response with the standard A2A error envelope.
+   *
+   * <p>Must be an exception handler rather than a {@code ResponseEntity} return so that streaming
+   * endpoints can declare {@link SseEmitter} as their return type. A wildcard {@code
+   * ResponseEntity<?>} return is not picked up by Spring's emitter handler and the emitter body is
+   * rejected by the message converters, turning every stream into a 500 response.
+   */
+  @ExceptionHandler(A2AError.class)
+  ResponseEntity<String> a2aError(A2AError error) {
+    var response = restHandler.createErrorResponse(error);
+    return toResponse(response);
+  }
 
-            @Override
-            public void onNext(StreamingEventKind item) {
+  private SseEmitter subscribeToEmitter(
+      Flow.Publisher<StreamingEventKind> publisher, ServerCallContext context) {
+    AtomicReference<Flow.Subscription> subscription = new AtomicReference<>();
+    AtomicBoolean released = new AtomicBoolean();
+    Runnable release =
+        () -> {
+          if (released.compareAndSet(false, true)) {
+            Flow.Subscription active = subscription.get();
+            if (active != null) {
               try {
-                StreamResponse sr = ProtoUtils.ToProto.streamResponse(item);
-                String compact = JsonFormat.printer().omittingInsignificantWhitespace().print(sr);
-                emitter.send(
-                    SseEmitter.event()
-                        .id(Long.toString(seq.incrementAndGet()))
-                        .data(compact, MediaType.APPLICATION_JSON));
-              } catch (Exception e) {
-                log.error("[SSE] Write failed: {}", e.getMessage());
-                sub.cancel();
-                emitter.completeWithError(e);
-                return;
+                active.cancel();
+              } catch (RuntimeException error) {
+                log.warn("[SSE] Failed to cancel publisher subscription: {}", error.getMessage());
               }
-              sub.request(1);
             }
-
-            @Override
-            public void onError(Throwable t) {
-              log.error("[SSE] Stream error: {}", t.getMessage());
-              emitter.completeWithError(t);
+            try {
+              context.invokeEventConsumerCancelCallback();
+            } catch (RuntimeException error) {
+              log.warn("[SSE] Failed to notify SDK event consumer cancellation: {}", error.getMessage());
             }
+          }
+        };
+    SseEmitter emitter = track(new SseEmitter(0L), release);
+    final AtomicLong seq = new AtomicLong(0);
+    publisher.subscribe(
+        new Flow.Subscriber<>() {
+          @Override
+          public void onSubscribe(Flow.Subscription s) {
+            if (!subscription.compareAndSet(null, s) || released.get()) {
+              s.cancel();
+            } else {
+              s.request(1);
+            }
+          }
 
-            @Override
-            public void onComplete() {
-              emitter.complete();
+          @Override
+          public void onNext(StreamingEventKind item) {
+            try {
+              StreamResponse sr = ProtoUtils.ToProto.streamResponse(item);
+              String compact = JsonFormat.printer().omittingInsignificantWhitespace().print(sr);
+              emitter.send(
+                  SseEmitter.event()
+                      .id(Long.toString(seq.incrementAndGet()))
+                      .data(compact, MediaType.APPLICATION_JSON));
+            } catch (Exception e) {
+              log.error("[SSE] Write failed: {}", e.getMessage());
+              release.run();
+              emitter.completeWithError(e);
+              return;
             }
-          });
-    } catch (Exception e) {
-      log.error("[SSE] Setup failed: {}", e.getMessage(), e);
-      emitter.completeWithError(e);
-    }
+            Flow.Subscription active = subscription.get();
+            if (active != null && !released.get()) {
+              active.request(1);
+            }
+          }
+
+          @Override
+          public void onError(Throwable t) {
+            log.error("[SSE] Stream error: {}", t.getMessage());
+            release.run();
+            emitter.completeWithError(t);
+          }
+
+          @Override
+          public void onComplete() {
+            release.run();
+            emitter.complete();
+          }
+        });
     return emitter;
   }
 
@@ -164,13 +230,7 @@ public void onComplete() {
   public ResponseEntity<String> getTask(HttpServletRequest req, @PathVariable("id") String taskId) {
     var ctx = buildContext(req);
     var resp = restHandler.getTask(ctx, "", taskId, null);
-    return ResponseEntity.status(resp.getStatusCode())
-        .contentType(
-            MediaType.parseMediaType(
-                resp.getContentType() != null
-                    ? resp.getContentType()
-                    : MediaType.APPLICATION_JSON_VALUE))
-        .body(resp.getBody());
+    return toResponse(resp);
   }
 
   @GetMapping("${a2at.server.path-prefix}/tasks")
@@ -194,13 +254,7 @@ public ResponseEntity<String> listTasks(
             historyLength,
             statusTimestampAfter,
             includeArtifacts);
-    return ResponseEntity.status(resp.getStatusCode())
-        .contentType(
-            MediaType.parseMediaType(
-                resp.getContentType() != null
-                    ? resp.getContentType()
-                    : MediaType.APPLICATION_JSON_VALUE))
-        .body(resp.getBody());
+    return toResponse(resp);
   }
 
   @PostMapping("${a2at.server.path-prefix}/tasks/{id}:cancel")
@@ -210,73 +264,37 @@ public ResponseEntity<String> cancelTask(
       @RequestBody(required = false) String body) {
     var ctx = buildContext(req);
     var resp = restHandler.cancelTask(ctx, "", body != null ? body : "{}", taskId);
-    return ResponseEntity.status(resp.getStatusCode())
-        .contentType(
-            MediaType.parseMediaType(
-                resp.getContentType() != null
-                    ? resp.getContentType()
-                    : MediaType.APPLICATION_JSON_VALUE))
-        .body(resp.getBody());
+    return toResponse(resp);
   }
 
   @PostMapping(
       value = "${a2at.server.path-prefix}/tasks/{id}:subscribe",
       produces = MediaType.TEXT_EVENT_STREAM_VALUE)
-  public SseEmitter subscribeToTask(HttpServletRequest req, @PathVariable("id") String taskId) {
-    SseEmitter emitter = track(new SseEmitter(0L));
+  public SseEmitter subscribeToTask(
+      HttpServletRequest req, @PathVariable("id") String taskId) {
     try {
       var ctx = buildContext(req);
-      var resp = restHandler.subscribeToTask(ctx, "", taskId);
-      if (resp instanceof RestHandler.HTTPRestStreamingResponse streaming) {
-        final AtomicLong seq = new AtomicLong(0);
-        streaming
-            .getPublisher()
-            .subscribe(
-                new Flow.Subscriber<String>() {
-                  private Flow.Subscription sub;
-
-                  @Override
-                  public void onSubscribe(Flow.Subscription s) {
-                    sub = s;
-                    s.request(1);
-                  }
-
-                  @Override
-                  public void onNext(String item) {
-                    try {
-                      emitter.send(
-                          SseEmitter.event()
-                              .id(Long.toString(seq.incrementAndGet()))
-                              .data(item, MediaType.APPLICATION_JSON));
-                    } catch (Exception e) {
-                      log.error("[SSE] Subscribe write failed: {}", e.getMessage());
-                      sub.cancel();
-                      emitter.completeWithError(e);
-                      return;
-                    }
-                    sub.request(1);
-                  }
-
-                  @Override
-                  public void onError(Throwable t) {
-                    log.error("[SSE] Subscribe stream error: {}", t.getMessage());
-                    emitter.completeWithError(t);
-                  }
-
-                  @Override
-                  public void onComplete() {
-                    emitter.complete();
-                  }
-                });
-      } else {
-        emitter.send(SseEmitter.event().id("1").data(resp.getBody(), MediaType.APPLICATION_JSON));
-        emitter.complete();
-      }
+      requireStreamingSupport();
+      requestHandler.authorizeTaskAccess(taskId, ctx, TaskOperation.SUBSCRIBE_TO_TASK);
+      Flow.Publisher<StreamingEventKind> publisher =
+          requestHandler.onSubscribeToTask(TaskIdParams.builder().id(taskId).tenant("").build(), ctx);
+      return subscribeToEmitter(publisher, ctx);
+    } catch (A2AError e) {
+      log.warn("[SSE] Task subscription rejected before stream creation: {}", e.getMessage());
+      throw e;
     } catch (Exception e) {
-      log.error("[SSE] Subscribe setup failed: {}", e.getMessage(), e);
-      emitter.completeWithError(e);
+      log.error(
+          "[SSE] Subscribe setup failed before stream creation: {}",
+          e.getClass().getSimpleName(),
+          e);
+      throw new InternalError("Task subscription setup failed");
+    }
+  }
+
+  private void requireStreamingSupport() throws UnsupportedOperationError {
+    if (!agentCard.capabilities().streaming()) {
+      throw new UnsupportedOperationError(null, "Streaming is not supported by the agent", null);
     }
-    return emitter;
   }
 
   /**
@@ -313,12 +331,13 @@ public int activeStreamCount() {
     return activeStreams.get();
   }
 
-  private <T extends ResponseBodyEmitter> T track(T emitter) {
+  private <T extends ResponseBodyEmitter> T track(T emitter, Runnable cleanup) {
     activeStreams.incrementAndGet();
     AtomicBoolean completed = new AtomicBoolean();
     Runnable finish =
         () -> {
           if (completed.compareAndSet(false, true)) {
+            cleanup.run();
             activeStreams.decrementAndGet();
             synchronized (streamCompletionMonitor) {
               streamCompletionMonitor.notifyAll();
@@ -331,6 +350,12 @@ private <T extends ResponseBodyEmitter> T track(T emitter) {
     return emitter;
   }
 
+  private static ResponseEntity<String> toResponse(RestHandler.HTTPRestResponse response) {
+    return ResponseEntity.status(response.getStatusCode())
+        .contentType(A2A_JSON)
+        .body(response.getBody());
+  }
+
   private ServerCallContext buildContext(HttpServletRequest req) {
     Map<String, Object> state = new LinkedHashMap<>();
     Map<String, String> headers = new LinkedHashMap<>();
diff --git a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java
index 2207bdc..a9f57d0 100644
--- a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java
+++ b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java
@@ -21,13 +21,41 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+import static org.hamcrest.Matchers.containsString;
 
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Flow;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.a2aproject.sdk.server.ServerCallContext;
+import org.a2aproject.sdk.server.auth.TaskOperation;
 import org.a2aproject.sdk.server.config.A2AConfigProvider;
+import org.a2aproject.sdk.server.requesthandlers.RequestHandler;
+import org.a2aproject.sdk.spec.AgentCapabilities;
+import org.a2aproject.sdk.spec.AgentCard;
+import org.a2aproject.sdk.spec.TaskNotFoundError;
+import org.a2aproject.sdk.spec.StreamingEventKind;
+import org.a2aproject.sdk.transport.rest.handler.RestHandler;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.test.context.runner.WebApplicationContextRunner;
+import org.springframework.http.ResponseEntity;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.MvcResult;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -76,11 +104,139 @@ void everyTaskRouteUsesConfiguredA2aPathPrefix() throws Exception {
         placeholder + "/tasks/{id}:subscribe",
         subscribeMethod.getAnnotation(PostMapping.class).value()[0]);
     assertEquals(SseEmitter.class, subscribeMethod.getReturnType());
+    assertEquals(
+        "text/event-stream", subscribeMethod.getAnnotation(PostMapping.class).produces()[0]);
 
     var streamMethod =
         A2AController.class.getMethod(
             "streamMessage", jakarta.servlet.http.HttpServletRequest.class, String.class);
+    // SseEmitter (not ResponseEntity): a wildcard ResponseEntity return is not picked up by
+    // Spring's emitter handler and the stream response degrades to a converter error.
     assertEquals(SseEmitter.class, streamMethod.getReturnType());
+    assertEquals("text/event-stream", streamMethod.getAnnotation(PostMapping.class).produces()[0]);
+  }
+
+  @Test
+  void completedSseReleasesPublisherAndSdkEventConsumer() throws Exception {
+    RestHandler restHandler = mock(RestHandler.class);
+    RequestHandler requestHandler = mock(RequestHandler.class);
+    AgentCard agentCard = mock(AgentCard.class);
+    when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(true).build());
+    AtomicReference<Flow.Subscriber<? super StreamingEventKind>> subscriber =
+        new AtomicReference<>();
+    AtomicInteger subscriptionCancellations = new AtomicInteger();
+    AtomicInteger eventConsumerCancellations = new AtomicInteger();
+    when(requestHandler.onSubscribeToTask(any(), any()))
+        .thenAnswer(
+            invocation -> {
+              ServerCallContext context = invocation.getArgument(1);
+              context.setEventConsumerCancelCallback(eventConsumerCancellations::incrementAndGet);
+              Flow.Publisher<StreamingEventKind> publisher =
+                  receiver -> {
+                    subscriber.set(receiver);
+                    receiver.onSubscribe(
+                        new Flow.Subscription() {
+                          @Override
+                          public void request(long count) {}
+
+                          @Override
+                          public void cancel() {
+                            subscriptionCancellations.incrementAndGet();
+                          }
+                        });
+                  };
+              return publisher;
+            });
+    var controller = new A2AController(restHandler, requestHandler, agentCard);
+    MockMvc mockMvc =
+        MockMvcBuilders.standaloneSetup(controller)
+            .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json")
+            .build();
+
+    MvcResult stream =
+        mockMvc
+            .perform(post("/a2a/json/tasks/task-1:subscribe").accept("text/event-stream"))
+            .andExpect(request().asyncStarted())
+            .andReturn();
+    subscriber.get().onComplete();
+    mockMvc.perform(asyncDispatch(stream)).andExpect(status().isOk());
+
+    assertEquals(1, subscriptionCancellations.get());
+    assertEquals(1, eventConsumerCancellations.get());
+    assertEquals(0, controller.activeStreamCount());
+  }
+
+  @Test
+  void nonStreamingA2AResponseUsesProtocolMediaType() throws Exception {
+    RestHandler restHandler = mock(RestHandler.class);
+    when(restHandler.sendMessage(any(), eq(""), eq("{}")))
+        .thenReturn(new RestHandler.HTTPRestResponse(200, "application/json", "{}"));
+    var controller =
+        new A2AController(restHandler, mock(RequestHandler.class), mock(AgentCard.class));
+    MockMvc mockMvc =
+        MockMvcBuilders.standaloneSetup(controller)
+            .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json")
+            .build();
+
+    mockMvc
+        .perform(post("/a2a/json/message:send").content("{}"))
+        .andExpect(status().isOk())
+        .andExpect(content().contentTypeCompatibleWith("application/a2a+json"));
+  }
+
+  @Test
+  void taskSubscriptionRejectionReturnsAStandardNonStreamingA2AError() throws Exception {
+    RestHandler restHandler = mock(RestHandler.class);
+    RequestHandler requestHandler = mock(RequestHandler.class);
+    var responseBody =
+        "{\"error\":{\"code\":404,\"status\":\"NOT_FOUND\","
+            + "\"message\":\"Task not found\",\"details\":[]}}";
+    when(restHandler.createErrorResponse(any()))
+        .thenReturn(new RestHandler.HTTPRestResponse(404, "application/a2a+json", responseBody));
+    doThrow(new TaskNotFoundError())
+        .when(requestHandler)
+        .authorizeTaskAccess(eq("missing"), any(), eq(TaskOperation.SUBSCRIBE_TO_TASK));
+    AgentCard agentCard = mock(AgentCard.class);
+    when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(true).build());
+    var controller = new A2AController(restHandler, requestHandler, agentCard);
+
+    // Through MockMvc so the exception -> @ExceptionHandler mapping runs exactly as in dispatch.
+    MockMvc mockMvc =
+        MockMvcBuilders.standaloneSetup(controller)
+            .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json")
+            .build();
+
+    mockMvc
+        .perform(post("/a2a/json/tasks/missing:subscribe"))
+        .andExpect(status().is(404))
+        .andExpect(content().contentTypeCompatibleWith("application/a2a+json"))
+        .andExpect(content().string(containsString("\"status\":\"NOT_FOUND\"")));
+  }
+
+  @Test
+  void streamingCapabilityIsCheckedBeforeInvokingBusinessHandler() throws Exception {
+    RestHandler restHandler = mock(RestHandler.class);
+    RequestHandler requestHandler = mock(RequestHandler.class);
+    AgentCard agentCard = mock(AgentCard.class);
+    when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(false).build());
+    when(restHandler.createErrorResponse(any()))
+        .thenReturn(
+            new RestHandler.HTTPRestResponse(
+                501,
+                "application/a2a+json",
+                "{\"error\":{\"code\":-32004,\"status\":\"UNIMPLEMENTED\"}}"));
+    var controller = new A2AController(restHandler, requestHandler, agentCard);
+
+    MockMvc mockMvc =
+        MockMvcBuilders.standaloneSetup(controller)
+            .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json")
+            .build();
+
+    mockMvc
+        .perform(post("/a2a/json/message:stream").content("{}"))
+        .andExpect(status().is(501))
+        .andExpect(content().contentTypeCompatibleWith("application/a2a+json"));
+    verifyNoInteractions(requestHandler);
   }
 
   @Test
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java
similarity index 81%
rename from workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClient.java
rename to workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java
index e4da1a1..c602404 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClient.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java
@@ -5,16 +5,16 @@
  * SPDX-License-Identifier: Apache-2.0
  *
  *    Licensed under the Apache License, Version 2.0 (the "License"); you may
- *    not use this file except in compliance with the License. You may obtain
- *    a copy of the License at
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
  *
  *         http://www.apache.org/licenses/LICENSE-2.0
  *
- *    Unless required by applicable law or agreed to in writing, software
- *    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- *    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- *    License for the specific language governing permissions and limitations
- *    under the License.
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
  */
 package dev.openan.workflow.engine.client;
 
@@ -28,11 +28,11 @@
 import org.a2aproject.sdk.client.http.A2AHttpResponse;
 import org.a2aproject.sdk.client.http.ServerSentEvent;
 
-/** Checks problem responses before the SDK's A2A parser; wire logging remains observational. */
-final class ProblemDetectingHttpClient implements A2AHttpClient {
+/** Detects standard A2A error envelopes before the SDK attempts to parse them as task events. */
+final class A2AErrorDetectingHttpClient implements A2AHttpClient {
   private final A2AHttpClient delegate;
 
-  ProblemDetectingHttpClient(A2AHttpClient delegate) {
+  A2AErrorDetectingHttpClient(A2AHttpClient delegate) {
     this.delegate = delegate;
   }
 
@@ -41,7 +41,8 @@ final class ProblemDetectingHttpClient implements A2AHttpClient {
   @Override public DeleteBuilder createDelete() { return new Delete(delegate.createDelete()); }
 
   private static A2AHttpResponse check(A2AHttpResponse response) {
-    var error = RemoteProblemException.fromPayload(response.body());
+    var error = RemoteA2AErrorException.fromResponse(
+        response.status(), response.body(), response.headers().toMap());
     if (error != null) throw error;
     return response;
   }
@@ -90,6 +91,7 @@ private static CompletableFuture<Void> stream(
     var result = new CompletableFuture<Void>();
     var upstream = new AtomicReference<CompletableFuture<Void>>();
     var ended = new AtomicBoolean();
+    var eventReceived = new AtomicBoolean();
     Consumer<Throwable> fail = error -> {
       if (!ended.compareAndSet(false, true)) return;
       try { errors.accept(error); }
@@ -107,13 +109,18 @@ private static CompletableFuture<Void> stream(
     });
     var active = starter.start(event -> {
       if (ended.get()) return;
-      var problem = RemoteProblemException.fromPayload(event.data());
-      if (problem != null) fail.accept(problem);
+      var remoteError = RemoteA2AErrorException.fromPayload(event.data());
+      if (remoteError != null) fail.accept(remoteError);
       else {
+        eventReceived.set(true);
         try { messages.accept(event); }
         catch (RuntimeException callbackError) { fail.accept(callbackError); }
       }
     }, fail, () -> {
+      if (!eventReceived.get()) {
+        fail.accept(new IOException("A2A streaming response closed without an event"));
+        return;
+      }
       if (ended.compareAndSet(false, true)) {
         try { complete.run(); result.complete(null); }
         catch (RuntimeException error) { result.completeExceptionally(error); }
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java
index b3b9ca0..3b87d04 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java
@@ -488,14 +488,16 @@ CompletableFuture<SendMessageResult> send(
                         .receivedMessages(ProtocolResponses.assemble(events))
                         .build();
                   } catch (Exception e) {
-                    RemoteProblemException problem = RemoteProblemException.findIn(e);
-                    if (problem != null) {
+                    RemoteA2AErrorException remoteError = RemoteA2AErrorException.findIn(e);
+                    if (remoteError != null) {
                       log.warn(
-                          "[Transport] REMOTE_PROBLEM agent={}, contextId={}, status={}, reason={}",
+                          "[Transport] A2A_ERROR agent={}, contextId={}, httpStatus={}, status={}, reason={}, message={}",
                           agentName,
                           contextId,
-                          problem.getStatus(),
-                          problem.getMessage().replace("\r", "\\r").replace("\n", "\\n"));
+                          remoteError.getHttpStatus(),
+                          remoteError.getStatus(),
+                          remoteError.getReason(),
+                          remoteError.getMessage().replace("\r", "\\r").replace("\n", "\\n"));
                     } else {
                       log.error(
                           "[Transport] Failed to send message to {}: {}",
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java
index 4abc929..14ca7bc 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java
@@ -216,18 +216,21 @@ private static void onError(
           error.getMessage());
     } else {
       String msg = error.getMessage() != null ? error.getMessage() : "";
-      RemoteProblemException problem = RemoteProblemException.findIn(error);
-      boolean connectionClosed = problem == null && TransportFailures.isExpectedLocalClose(error);
+      RemoteA2AErrorException remoteError = RemoteA2AErrorException.findIn(error);
+      boolean connectionClosed =
+          remoteError == null && TransportFailures.isExpectedLocalClose(error);
       if (connectionClosed) {
         log.debug("[A2ARuntime] Connection closed for '{}': {}", agentName, msg);
       } else {
         errorRef.set(error);
-        if (problem != null) {
+        if (remoteError != null) {
           log.warn(
-              "[A2ARuntime] REMOTE_PROBLEM agent={}, status={}, reason={}",
+              "[A2ARuntime] A2A_ERROR agent={}, httpStatus={}, status={}, reason={}, message={}",
               agentName,
-              problem.getStatus(),
-              problem.getMessage().replace("\r", "\\r").replace("\n", "\\n"));
+              remoteError.getHttpStatus(),
+              remoteError.getStatus(),
+              remoteError.getReason(),
+              remoteError.getMessage().replace("\r", "\\r").replace("\n", "\\n"));
         } else {
           log.error(
               "[A2ARuntime] Error callback for '{}': {}", agentName, error.getMessage(), error);
@@ -695,7 +698,8 @@ private A2AHttpClient createHttpClient() {
             crlPath,
             Duration.ofSeconds(60),
             httpClientExecutor);
-    return new ProblemDetectingHttpClient(new JdkA2AHttpClient(new ObservedHttpClient(httpClient)));
+    return new A2AErrorDetectingHttpClient(
+        new JdkA2AHttpClient(new ObservedHttpClient(httpClient)));
   }
 
   private record StreamClientKey(String agentName, String contextId) {}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java
index 3a42077..ab78645 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java
@@ -19,6 +19,7 @@
 
 package dev.openan.workflow.engine.client;
 
+import dev.openan.workflow.engine.util.SensitiveDataRedactor;
 import java.io.IOException;
 import java.net.*;
 import java.net.http.*;
@@ -145,6 +146,8 @@ private final class Exchange {
     final Map<String, String> context = WireLog.context();
     final java.util.concurrent.atomic.AtomicBoolean failureLogged =
         new java.util.concurrent.atomic.AtomicBoolean();
+    final java.util.concurrent.atomic.AtomicBoolean cancelledBySubscriber =
+        new java.util.concurrent.atomic.AtomicBoolean();
 
     Exchange(HttpRequest request) {
       this.request = request;
@@ -174,15 +177,20 @@ void emit(
     }
 
     void failure(Throwable error) {
+      if (cancelledBySubscriber.get()) return;
       // Body subscriber and sendAsync completion can report the same exchange failure.
       if (!failureLogged.compareAndSet(false, true)) return;
+      String message = SensitiveDataRedactor.redact(Objects.toString(error.getMessage(), ""))
+          .replace("\r", "\\r")
+          .replace("\n", "\\n");
       emit(
           "FAILURE",
           null,
           Map.of(),
           "transport-error",
           "",
-          "errorType=" + error.getClass().getSimpleName());
+          "errorType=" + error.getClass().getSimpleName()
+              + (message.isBlank() ? "" : "; errorMessage=" + message));
     }
 
     HttpRequest observedRequest() {
@@ -312,6 +320,7 @@ public void request(long count) {
 
                   @Override
                   public void cancel() {
+                    cancelledBySubscriber.set(true);
                     body.end(true);
                     subscription.cancel();
                   }
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteA2AErrorException.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteA2AErrorException.java
new file mode 100644
index 0000000..b820f26
--- /dev/null
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteA2AErrorException.java
@@ -0,0 +1,243 @@
+/*
+ * Copyright (c) 2026 Huawei Technologies Co., Ltd.
+ * All Rights Reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package dev.openan.workflow.engine.client;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import dev.openan.workflow.engine.util.SensitiveDataRedactor;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import org.a2aproject.sdk.spec.A2AClientHTTPError;
+import org.a2aproject.sdk.spec.A2AError;
+import org.a2aproject.sdk.spec.A2AErrorCodes;
+
+/** A standard A2A HTTP error projected from the {@code google.rpc.Status} JSON envelope. */
+public final class RemoteA2AErrorException extends RuntimeException {
+  private static final ObjectMapper JSON =
+      new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
+  private static final TypeReference<List<Map<String, Object>>> DETAIL_TYPE =
+      new TypeReference<>() {};
+
+  private final int httpStatus;
+  private final int code;
+  private final String status;
+  private final String reason;
+  private final String domain;
+  private final List<Map<String, Object>> details;
+  private final Map<String, List<String>> responseHeaders;
+
+  private RemoteA2AErrorException(
+      int httpStatus,
+      int code,
+      String status,
+      String message,
+      String reason,
+      String domain,
+      List<Map<String, Object>> details,
+      Map<String, List<String>> responseHeaders) {
+    super(message);
+    this.httpStatus = httpStatus;
+    this.code = code;
+    this.status = status;
+    this.reason = reason;
+    this.domain = domain;
+    this.details = List.copyOf(details);
+    this.responseHeaders = SensitiveDataRedactor.safeHeaders(responseHeaders);
+  }
+
+  /** Parses a top-level standard A2A error envelope whose embedded code supplies HTTP status. */
+  public static RemoteA2AErrorException fromPayload(String payload) {
+    return fromResponse(0, payload, Map.of());
+  }
+
+  /** Parses a standard A2A error response, preferring an observed non-2xx HTTP status. */
+  public static RemoteA2AErrorException fromResponse(
+      int observedHttpStatus, String payload, Map<String, List<String>> responseHeaders) {
+    if (payload == null || payload.isBlank()) return null;
+    JsonNode root;
+    try {
+      root = JSON.readTree(payload);
+    } catch (java.io.IOException ignored) {
+      return null;
+    }
+    if (root == null || !root.isObject() || !root.path("error").isObject()) return null;
+    JsonNode error = root.path("error");
+    if (!error.path("code").isIntegralNumber() || !error.path("code").canConvertToInt()) {
+      return null;
+    }
+    int code = error.path("code").asInt();
+    int httpStatus = isErrorStatus(observedHttpStatus) ? observedHttpStatus : code;
+    if (!isErrorStatus(code) || !isErrorStatus(httpStatus)) return null;
+    String message = safeText(error, "message");
+    if (message.isBlank()) return null;
+    String status = safeText(error, "status");
+    List<Map<String, Object>> details = safeDetails(error.path("details"));
+    String reason = "";
+    String domain = "";
+    if (error.path("details").isArray()) {
+      for (JsonNode detail : error.path("details")) {
+        if (!detail.isObject() || !detail.path("reason").isTextual()) continue;
+        reason = safeText(detail, "reason");
+        domain = safeText(detail, "domain");
+        break;
+      }
+    }
+    return new RemoteA2AErrorException(
+        httpStatus,
+        code,
+        status,
+        message,
+        reason,
+        domain,
+        details,
+        responseHeaders == null ? Map.of() : responseHeaders);
+  }
+
+  /** Finds or projects a standard A2A error in a wrapped SDK/transport exception chain. */
+  public static RemoteA2AErrorException findIn(Throwable error) {
+    Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<>());
+    for (Throwable current = error; current != null && seen.add(current); current = current.getCause()) {
+      if (current instanceof RemoteA2AErrorException remote) return remote;
+      if (current instanceof A2AClientHTTPError http) {
+        RemoteA2AErrorException parsed =
+            fromResponse(http.getCode(), http.getResponseBody(), http.getResponseHeaders());
+        if (parsed != null) return parsed;
+        return transportOnly(http);
+      }
+      if (current instanceof A2AError a2a) return fromSdkError(a2a);
+    }
+    return null;
+  }
+
+  private static RemoteA2AErrorException transportOnly(A2AClientHTTPError error) {
+    String message = SensitiveDataRedactor.redact(error.getMessage());
+    return new RemoteA2AErrorException(
+        error.getCode(),
+        error.getCode(),
+        "",
+        message.isBlank() ? "A2A request failed with HTTP " + error.getCode() : message,
+        "",
+        "",
+        List.of(),
+        error.getResponseHeaders());
+  }
+
+  private static RemoteA2AErrorException fromSdkError(A2AError error) {
+    A2AErrorCodes errorCode = A2AErrorCodes.fromCode(error.getCode());
+    int httpStatus = errorCode == null ? 500 : errorCode.httpCode();
+    String reason = errorCode == null ? "" : errorCode.name();
+    String status = errorCode == null ? "INTERNAL" : errorCode.grpcStatus();
+    Map<String, Object> detail = new LinkedHashMap<>();
+    detail.put("@type", "type.googleapis.com/google.rpc.ErrorInfo");
+    if (!reason.isBlank()) detail.put("reason", reason);
+    detail.put("domain", "a2a-protocol.org");
+    if (!error.getDetails().isEmpty()) detail.put("metadata", error.getDetails());
+    List<Map<String, Object>> details = safeDetails(JSON.valueToTree(List.of(detail)));
+    return new RemoteA2AErrorException(
+        httpStatus,
+        httpStatus,
+        status,
+        SensitiveDataRedactor.redact(error.getMessage()),
+        reason,
+        "a2a-protocol.org",
+        details,
+        Map.of());
+  }
+
+  private static List<Map<String, Object>> safeDetails(JsonNode node) {
+    if (!node.isArray()) return List.of();
+    try {
+      String redacted = SensitiveDataRedactor.redact(JSON.writeValueAsString(node));
+      List<Map<String, Object>> converted = JSON.readValue(redacted, DETAIL_TYPE);
+      List<Map<String, Object>> result = new ArrayList<>();
+      for (Map<String, Object> detail : converted) {
+        result.add(Collections.unmodifiableMap(new LinkedHashMap<>(detail)));
+      }
+      return List.copyOf(result);
+    } catch (java.io.IOException ignored) {
+      return List.of();
+    }
+  }
+
+  private static boolean isErrorStatus(int value) {
+    return value >= 400 && value <= 599;
+  }
+
+  private static String safeText(JsonNode node, String key) {
+    return node.path(key).isTextual()
+        ? SensitiveDataRedactor.redact(node.path(key).textValue())
+        : "";
+  }
+
+  /** Stable workflow-facing error code that does not expose SDK exception class names. */
+  public String workflowErrorCode() {
+    String normalizedReason =
+        reason
+            .toLowerCase(Locale.ROOT)
+            .replaceAll("[^a-z0-9]+", "_")
+            .replaceAll("^_+|_+$", "");
+    if (!normalizedReason.isBlank()) return "a2a." + normalizedReason;
+    return "a2a.http." + httpStatus;
+  }
+
+  public int getHttpStatus() {
+    return httpStatus;
+  }
+
+  public int getCode() {
+    return code;
+  }
+
+  public String getStatus() {
+    return status;
+  }
+
+  public String getReason() {
+    return reason;
+  }
+
+  public String getDomain() {
+    return domain;
+  }
+
+  public List<Map<String, Object>> getDetails() {
+    return details;
+  }
+
+  public Map<String, List<String>> getResponseHeaders() {
+    return responseHeaders;
+  }
+
+  public String getRetryAfter() {
+    for (Map.Entry<String, List<String>> header : responseHeaders.entrySet()) {
+      if ("retry-after".equalsIgnoreCase(header.getKey()) && !header.getValue().isEmpty()) {
+        return header.getValue().get(0);
+      }
+    }
+    return "";
+  }
+}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteProblemException.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteProblemException.java
deleted file mode 100644
index c5e10e1..0000000
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteProblemException.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- *    Licensed under the Apache License, Version 2.0 (the "License"); you may
- *    not use this file except in compliance with the License. You may obtain
- *    a copy of the License at
- *
- *         http://www.apache.org/licenses/LICENSE-2.0
- *
- *    Unless required by applicable law or agreed to in writing, software
- *    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- *    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- *    License for the specific language governing permissions and limitations
- *    under the License.
- */
-package dev.openan.workflow.engine.client;
-
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-/**
- * A remote problem response, including errors sent as SSE data after HTTP 200 headers.
- * Known credential fields are redacted using the same rules as protocol logging.
- */
-public final class RemoteProblemException extends RuntimeException {
-  private static final ObjectMapper JSON =
-      new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
-  private final int status;
-  private final String title;
-  private final String detail;
-  private final String type;
-  private final String timestamp;
-
-  private RemoteProblemException(JsonNode problem) {
-    super("Remote agent error (status=" + problem.path("status").asInt() + "): "
-        + text(problem, "title")
-        + (text(problem, "title").isEmpty() ? "" : " — ") + text(problem, "detail"));
-    status = problem.path("status").asInt();
-    title = text(problem, "title");
-    detail = text(problem, "detail");
-    type = text(problem, "type");
-    timestamp = text(problem, "timestamp");
-  }
-
-  /** Returns a top-level problem error, or null for ordinary A2A events and incomplete JSON. */
-  public static RemoteProblemException fromPayload(String payload) {
-    if (payload == null || payload.isBlank()) return null;
-    JsonNode node;
-    try {
-      node = JSON.readTree(payload);
-    } catch (java.io.IOException ignored) {
-      return null;
-    }
-    if (node == null || !node.isObject() || !node.path("status").isIntegralNumber()
-        || !node.path("status").canConvertToInt()) return null;
-    int status = node.path("status").asInt();
-    if (status < 400 || status > 599
-        || (text(node, "title").isBlank() && text(node, "detail").isBlank())) return null;
-    for (String envelope : new String[] {"task", "message", "statusUpdate", "artifactUpdate", "result", "error"}) {
-      if (node.has(envelope)) return null;
-    }
-    return new RemoteProblemException(node);
-  }
-
-  private static String text(JsonNode node, String key) {
-    return node.path(key).isTextual() ? WireLog.redact(node.path(key).textValue()) : "";
-  }
-
-  /** Finds a problem preserved in a transport exception's cause chain, without looping on cycles. */
-  public static RemoteProblemException findIn(Throwable error) {
-    var seen = java.util.Collections.newSetFromMap(new java.util.IdentityHashMap<Throwable, Boolean>());
-    for (Throwable current = error; current != null && seen.add(current); current = current.getCause()) {
-      if (current instanceof RemoteProblemException problem) return problem;
-    }
-    return null;
-  }
-
-  /** Problem status, which can differ from the HTTP envelope status. */
-  public int getStatus() { return status; }
-  /** Remote error title, or an empty string. */
-  public String getTitle() { return title; }
-  /** Remote business failure detail. */
-  public String getDetail() { return detail; }
-  /** Remote problem type, preserved even when empty. */
-  public String getType() { return type; }
-  /** Remote timestamp when supplied. */
-  public String getTimestamp() { return timestamp; }
-}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java
index a646569..c57de42 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java
@@ -19,12 +19,12 @@
 
 package dev.openan.workflow.engine.core;
 
-import dev.openan.workflow.engine.client.RemoteProblemException;
+import dev.openan.workflow.engine.client.RemoteA2AErrorException;
 import dev.openan.workflow.engine.model.BusinessFailure;
 import dev.openan.workflow.engine.model.TaskResult;
 import java.util.*;
 
-/** Maps generic failures, remote problem facts, or explicitly safe host-selected business facts. */
+/** Maps generic failures, standard A2A errors, or explicitly safe host-selected business facts. */
 final class FailureMapping {
   static TaskResult from(Throwable error) {
     Throwable root = error;
@@ -32,7 +32,7 @@ static TaskResult from(Throwable error) {
     while (root.getCause() != null
         && seen.add(root)
         && !(root instanceof BusinessFailure)
-        && !(root instanceof RemoteProblemException)) {
+        && !(root instanceof RemoteA2AErrorException)) {
       root = root.getCause();
     }
     if (root instanceof BusinessFailure business) {
@@ -43,18 +43,23 @@ static TaskResult from(Throwable error) {
           .errorDetails(business.details())
           .build();
     }
-    if (root instanceof RemoteProblemException problem) {
+    RemoteA2AErrorException remoteError = RemoteA2AErrorException.findIn(root);
+    if (remoteError != null) {
+      Map<String, Object> details = new LinkedHashMap<>();
+      details.put("httpStatus", remoteError.getHttpStatus());
+      details.put("code", remoteError.getCode());
+      if (!remoteError.getStatus().isBlank()) details.put("status", remoteError.getStatus());
+      if (!remoteError.getReason().isBlank()) details.put("reason", remoteError.getReason());
+      if (!remoteError.getDomain().isBlank()) details.put("domain", remoteError.getDomain());
+      if (!remoteError.getDetails().isEmpty()) details.put("details", remoteError.getDetails());
+      if (!remoteError.getRetryAfter().isBlank()) {
+        details.put("retryAfter", remoteError.getRetryAfter());
+      }
       return TaskResult.builder()
           .success(false)
-          .errorCode("remote.problem." + problem.getStatus())
-          .error(problem.getMessage())
-          .errorDetails(
-              Map.of(
-                  "status", problem.getStatus(),
-                  "title", problem.getTitle(),
-                  "detail", problem.getDetail(),
-                  "type", problem.getType(),
-                  "timestamp", problem.getTimestamp()))
+          .errorCode(remoteError.workflowErrorCode())
+          .error(remoteError.getMessage())
+          .errorDetails(details)
           .build();
     }
     String code =
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClientTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java
similarity index 67%
rename from workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClientTest.java
rename to workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java
index c17a1ed..c4ae363 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClientTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java
@@ -5,16 +5,16 @@
  * SPDX-License-Identifier: Apache-2.0
  *
  *    Licensed under the Apache License, Version 2.0 (the "License"); you may
- *    not use this file except in compliance with the License. You may obtain
- *    a copy of the License at
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
  *
  *         http://www.apache.org/licenses/LICENSE-2.0
  *
- *    Unless required by applicable law or agreed to in writing, software
- *    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- *    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- *    License for the specific language governing permissions and limitations
- *    under the License.
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
  */
 package dev.openan.workflow.engine.client;
 
@@ -31,15 +31,15 @@
 import org.a2aproject.sdk.client.http.ServerSentEvent;
 import org.junit.jupiter.api.Test;
 
-class ProblemDetectingHttpClientTest {
+class A2AErrorDetectingHttpClientTest {
   @Test
-  void synchronousProblemCancelsUpstreamAndSurvivesAThrowingErrorObserver() throws Exception {
+  void standardErrorCancelsUpstreamAndSurvivesAThrowingErrorObserver() throws Exception {
     var upstream = new CompletableFuture<Void>();
     var failures = new AtomicInteger();
     var events = new AtomicInteger();
     var completions = new AtomicInteger();
     var client = client((messages, errors, complete) -> {
-      messages.accept(new ServerSentEvent("{\"status\":429,\"detail\":\"capacity reached\"}"));
+      messages.accept(new ServerSentEvent(RemoteA2AErrorResponseTest.error(429)));
       messages.accept(new ServerSentEvent("{\"message\":{}}"));
       complete.run();
       errors.accept(new IllegalStateException("late close"));
@@ -50,10 +50,11 @@ void synchronousProblemCancelsUpstreamAndSurvivesAThrowingErrorObserver() throws
           failures.incrementAndGet();
           throw new IllegalStateException("observer failed");
         }, completions::incrementAndGet);
-    var problem = RemoteProblemException.findIn(assertThrows(CompletionException.class, result::join));
-    assertNotNull(problem);
-    assertEquals(429, problem.getStatus());
-    assertEquals(1, problem.getSuppressed().length);
+    var remoteError = RemoteA2AErrorException.findIn(
+        assertThrows(CompletionException.class, result::join));
+    assertNotNull(remoteError);
+    assertEquals(429, remoteError.getHttpStatus());
+    assertEquals(1, remoteError.getSuppressed().length);
     assertTrue(upstream.isCancelled());
     assertEquals(1, failures.get());
     assertEquals(0, events.get());
@@ -80,8 +81,9 @@ void cancellationPropagatesAndLateEventsCannotCompleteTheCall() throws Exception
   }
 
   @Test
-  void successfulDataIsNotRewrittenOrMistakenForANestedProblem() throws Exception {
-    String data = "{\"task\":{\"status\":{\"state\":\"TASK_STATE_COMPLETED\"}},\"metadata\":{\"status\":400}}";
+  void successfulDataIsNotRewrittenOrMistakenForANestedError() throws Exception {
+    String data = "{\"task\":{\"status\":{\"state\":\"TASK_STATE_COMPLETED\"}},"
+        + "\"metadata\":{\"error\":{\"code\":400,\"message\":\"business value\"}}}";
     var seen = new AtomicReference<String>();
     var completions = new AtomicInteger();
     var result = client((messages, errors, complete) -> {
@@ -95,8 +97,25 @@ void successfulDataIsNotRewrittenOrMistakenForANestedProblem() throws Exception
     assertEquals(1, completions.get());
   }
 
+  @Test
+  void streamThatCompletesWithoutAnA2AEventFailsImmediately() throws Exception {
+    var failures = new AtomicReference<Throwable>();
+    var completions = new AtomicInteger();
+    var result = client((messages, errors, complete) -> {
+      complete.run();
+      return CompletableFuture.completedFuture(null);
+    }).createPost().postAsyncSSE(event -> fail("unexpected event"), failures::set,
+        completions::incrementAndGet);
+
+    var failure = assertThrows(CompletionException.class, result::join);
+    assertInstanceOf(java.io.IOException.class, failure.getCause());
+    assertEquals("A2A streaming response closed without an event", failure.getCause().getMessage());
+    assertSame(failure.getCause(), failures.get());
+    assertEquals(0, completions.get());
+  }
+
   private static A2AHttpClient client(Start start) {
-    return new ProblemDetectingHttpClient(new A2AHttpClient() {
+    return new A2AErrorDetectingHttpClient(new A2AHttpClient() {
       @Override public GetBuilder createGet() { throw new UnsupportedOperationException(); }
       @Override public DeleteBuilder createDelete() { throw new UnsupportedOperationException(); }
       @Override public PostBuilder createPost() {
@@ -117,6 +136,7 @@ private static A2AHttpClient client(Start start) {
 
   @FunctionalInterface
   private interface Start {
-    CompletableFuture<Void> run(Consumer<ServerSentEvent> messages, Consumer<Throwable> errors, Runnable complete);
+    CompletableFuture<Void> run(Consumer<ServerSentEvent> messages, Consumer<Throwable> errors,
+        Runnable complete);
   }
 }
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java
index b8e94cc..5991209 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java
@@ -22,10 +22,12 @@
 import static org.junit.jupiter.api.Assertions.*;
 
 import com.sun.net.httpserver.HttpServer;
+import java.io.IOException;
 import java.net.*;
 import java.net.http.*;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.*;
 import org.junit.jupiter.api.AfterEach;
@@ -177,4 +179,116 @@ void loggingFailureCannotChangeBodyDeliveryAndBodySwitchIsExplicit() throws Exce
                 "",
                 Map.of()));
   }
+
+  @Test
+  void subscriberCancellationIsNotLoggedAsATransportFailure() {
+    List<WireLog.Entry> records = new CopyOnWriteArrayList<>();
+    HttpRequest request =
+        HttpRequest.newBuilder(URI.create("https://agent.example/message:stream")).GET().build();
+    HttpResponse.BodyHandler<Void> cancellingHandler =
+        ignored ->
+            new HttpResponse.BodySubscriber<>() {
+              private final CompletableFuture<Void> body = new CompletableFuture<>();
+
+              @Override
+              public CompletionStage<Void> getBody() {
+                return body;
+              }
+
+              @Override
+              public void onSubscribe(Flow.Subscription subscription) {
+                subscription.cancel();
+              }
+
+              @Override
+              public void onNext(List<ByteBuffer> item) {}
+
+              @Override
+              public void onError(Throwable error) {
+                body.completeExceptionally(error);
+              }
+
+              @Override
+              public void onComplete() {
+                body.complete(null);
+              }
+            };
+
+    CompletableFuture<HttpResponse<Void>> result =
+        new ObservedHttpClient(new CancellationReportingHttpClient(), records::add)
+            .sendAsync(request, cancellingHandler);
+
+    assertThrows(CompletionException.class, result::join);
+    assertTrue(records.stream().noneMatch(entry -> entry.direction().equals("FAILURE")));
+  }
+
+  private static final class CancellationReportingHttpClient extends HttpClient {
+    @Override
+    public <T> HttpResponse<T> send(HttpRequest request, HttpResponse.BodyHandler<T> handler) {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public <T> CompletableFuture<HttpResponse<T>> sendAsync(
+        HttpRequest request, HttpResponse.BodyHandler<T> handler) {
+      CompletableFuture<HttpResponse<T>> result = new CompletableFuture<>();
+      HttpResponse.ResponseInfo info =
+          new HttpResponse.ResponseInfo() {
+            @Override
+            public int statusCode() {
+              return 200;
+            }
+
+            @Override
+            public HttpHeaders headers() {
+              return HttpHeaders.of(
+                  Map.of("Content-Type", List.of("text/event-stream")), (name, value) -> true);
+            }
+
+            @Override
+            public Version version() {
+              return Version.HTTP_1_1;
+            }
+          };
+      handler
+          .apply(info)
+          .onSubscribe(
+              new Flow.Subscription() {
+                @Override
+                public void request(long count) {}
+
+                @Override
+                public void cancel() {
+                  result.completeExceptionally(new IOException("subscriber cancelled"));
+                }
+              });
+      return result;
+    }
+
+    @Override
+    public <T> CompletableFuture<HttpResponse<T>> sendAsync(
+        HttpRequest request,
+        HttpResponse.BodyHandler<T> handler,
+        HttpResponse.PushPromiseHandler<T> pushPromiseHandler) {
+      return sendAsync(request, handler);
+    }
+
+    @Override public Optional<CookieHandler> cookieHandler() { return Optional.empty(); }
+    @Override public Optional<Duration> connectTimeout() { return Optional.empty(); }
+    @Override public Redirect followRedirects() { return Redirect.NEVER; }
+    @Override public Optional<ProxySelector> proxy() { return Optional.empty(); }
+    @Override public javax.net.ssl.SSLContext sslContext() { return defaultSslContext(); }
+    @Override public javax.net.ssl.SSLParameters sslParameters() { return new javax.net.ssl.SSLParameters(); }
+    @Override public Optional<Authenticator> authenticator() { return Optional.empty(); }
+    @Override public Version version() { return Version.HTTP_1_1; }
+    @Override public Optional<Executor> executor() { return Optional.empty(); }
+
+    private static javax.net.ssl.SSLContext defaultSslContext() {
+      try {
+        return javax.net.ssl.SSLContext.getDefault();
+      } catch (java.security.NoSuchAlgorithmException error) {
+        throw new IllegalStateException(error);
+      }
+    }
+  }
 }
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteA2AErrorResponseTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteA2AErrorResponseTest.java
new file mode 100644
index 0000000..c157baf
--- /dev/null
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteA2AErrorResponseTest.java
@@ -0,0 +1,194 @@
+/*
+ * Copyright (c) 2026 Huawei Technologies Co., Ltd.
+ * All Rights Reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package dev.openan.workflow.engine.client;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.sun.net.httpserver.HttpServer;
+import java.net.InetSocketAddress;
+import java.net.http.HttpClient;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.a2aproject.sdk.client.http.JdkA2AHttpClient;
+import org.a2aproject.sdk.spec.*;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class RemoteA2AErrorResponseTest {
+  static String error(int code) {
+    String status = code == 429 ? "RESOURCE_EXHAUSTED" : "INVALID_ARGUMENT";
+    String reason = code == 429 ? "ACTIVE_TASK_LIMIT_EXCEEDED" : "INVALID_PARAMS";
+    String domain = code == 429 ? "fixture.invalid" : "a2a-protocol.org";
+    return "{\"error\":{\"code\":" + code + ",\"status\":\"" + status
+        + "\",\"message\":\"该端口下未发现业务信息\",\"details\":["
+        + "{\"@type\":\"type.googleapis.com/google.rpc.BadRequest\","
+        + "\"fieldViolations\":[{\"field\":\"port\",\"description\":\"required\"}]},"
+        + "{\"@type\":\"type.googleapis.com/google.rpc.ErrorInfo\",\"reason\":\""
+        + reason + "\",\"domain\":\"" + domain + "\","
+        + "\"metadata\":{\"taskId\":\"task-123\"}}]}}";
+  }
+
+  @Test
+  void preservesStandardFieldsAndFindsErrorInfoAmongTypedDetails() {
+    var error = RemoteA2AErrorException.fromResponse(
+        400, error(400), Map.of("Retry-After", List.of("5")));
+    assertNotNull(error);
+    assertEquals(400, error.getHttpStatus());
+    assertEquals(400, error.getCode());
+    assertEquals("INVALID_ARGUMENT", error.getStatus());
+    assertEquals("INVALID_PARAMS", error.getReason());
+    assertEquals("a2a-protocol.org", error.getDomain());
+    assertEquals("该端口下未发现业务信息", error.getMessage());
+    assertEquals(2, error.getDetails().size());
+    assertEquals("5", error.getRetryAfter());
+    assertEquals("a2a.invalid_params", error.workflowErrorCode());
+  }
+
+  @Test
+  void ignoresOldProblemShapeAndNestedBusinessErrors() {
+    assertNull(RemoteA2AErrorException.fromPayload(
+        "{\"status\":400,\"title\":\"Invalid Params\",\"detail\":\"missing\"}"));
+    assertNull(RemoteA2AErrorException.fromPayload(
+        "{\"artifactUpdate\":{\"artifact\":{\"parts\":[{\"data\":" + error(400) + "}]}}}"));
+    assertNull(RemoteA2AErrorException.fromPayload("{\"error\":{\"code\":200,\"message\":\"ok\"}}"));
+    assertNull(RemoteA2AErrorException.fromPayload("{\"error\":{\"code\":400}}"));
+    assertNull(RemoteA2AErrorException.fromPayload("{\"error\":{"));
+  }
+
+  @Test
+  void redactsTypedDetailsAndResponseHeaders() {
+    String payload = error(400).replace("\"taskId\":\"task-123\"",
+        "\"accessSession\":\"secret-value\"");
+    var error = RemoteA2AErrorException.fromResponse(
+        400, payload, Map.of("Authorization", List.of("Bearer secret")));
+    assertEquals("***", ((Map<?, ?>) error.getDetails().get(1).get("metadata")).get("accessSession"));
+    assertEquals(List.of("***"), error.getResponseHeaders().get("Authorization"));
+  }
+
+  @Test
+  void normalizesPeerSuppliedReasonBeforeUsingItAsAWorkflowCode() {
+    String payload =
+        error(400).replace("INVALID_PARAMS", "Invalid Params / Port#1");
+
+    var error = RemoteA2AErrorException.fromPayload(payload);
+
+    assertNotNull(error);
+    assertEquals("a2a.invalid_params_port_1", error.workflowErrorCode());
+  }
+
+  @ParameterizedTest
+  @ValueSource(ints = {400, 429, 503})
+  void directRuntimeFailsPromptlyWhenSdkExposesAnErrorBodyAsSseData(int status) throws Exception {
+    var release = new CountDownLatch(1);
+    var disconnected = new CountDownLatch(1);
+    var requests = new AtomicInteger();
+    HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+    server.createContext("/", exchange -> {
+      requests.incrementAndGet();
+      exchange.getRequestBody().readAllBytes();
+      exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
+      exchange.sendResponseHeaders(200, 0);
+      try {
+        byte[] body = ("data: " + error(status) + "\n\n").getBytes(StandardCharsets.UTF_8);
+        exchange.getResponseBody().write(body, 0, 12);
+        exchange.getResponseBody().flush();
+        exchange.getResponseBody().write(body, 12, body.length - 12);
+        exchange.getResponseBody().flush();
+        for (int i = 0; i < 80 && !release.await(100, TimeUnit.MILLISECONDS); i++) {
+          exchange.getResponseBody().write(": keep-alive\n\n".getBytes(StandardCharsets.UTF_8));
+          exchange.getResponseBody().flush();
+        }
+      } catch (java.io.IOException clientClosed) {
+        disconnected.countDown();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      } finally {
+        exchange.close();
+      }
+    });
+    server.start();
+    var runtime = new DefaultA2AJavaClientRuntime(true, null, 20, "HTTP+JSON");
+    try {
+      AgentCard card = AgentCard.builder().name("remote-test").description("test").version("1")
+          .capabilities(AgentCapabilities.builder().streaming(true).build())
+          .defaultInputModes(List.of("text/plain")).defaultOutputModes(List.of("text/plain"))
+          .skills(List.of()).supportedInterfaces(List.of(new AgentInterface("HTTP+JSON",
+              "http://127.0.0.1:" + server.getAddress().getPort() + "/a2a/json"))).build();
+      MessageSendParams params = MessageSendParams.builder().message(Message.builder()
+          .role(Message.Role.ROLE_USER).messageId("test").parts(new TextPart("diagnose")).build()).build();
+      var events = new AtomicInteger();
+      var response = CompletableFuture.supplyAsync(
+          () -> runtime.sendMessage(card, params, null, event -> events.incrementAndGet(), null));
+      var thrown = assertThrows(java.util.concurrent.ExecutionException.class,
+          () -> response.get(3, TimeUnit.SECONDS));
+      var remoteError = RemoteA2AErrorException.findIn(thrown);
+      assertNotNull(remoteError);
+      assertEquals(status, remoteError.getHttpStatus());
+      assertEquals(0, events.get());
+      assertEquals(1, requests.get());
+      assertTrue(disconnected.await(3, TimeUnit.SECONDS), "failed SSE must close its HTTP connection");
+    } finally {
+      release.countDown();
+      runtime.close();
+      server.stop(0);
+    }
+  }
+
+  @Test
+  void projectsTypedSdkErrorsAndHandlesCauseCycles() {
+    var projected = RemoteA2AErrorException.findIn(
+        new IllegalStateException("wrapper", new InvalidParamsError("missing port")));
+    assertNotNull(projected);
+    assertEquals("INVALID_PARAMS", projected.getReason());
+    assertEquals("a2a.invalid_params", projected.workflowErrorCode());
+    assertNull(RemoteA2AErrorException.findIn(null));
+    var one = new IllegalStateException("one");
+    var two = new IllegalStateException("two", one);
+    one.initCause(two);
+    assertNull(RemoteA2AErrorException.findIn(one));
+  }
+
+  @Test
+  void nonStreamingResponsesPreserveTheStandardA2AError() throws Exception {
+    HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+    server.createContext("/", exchange -> {
+      byte[] body = error(400).getBytes(StandardCharsets.UTF_8);
+      exchange.getResponseHeaders().set("Content-Type", "application/a2a+json");
+      exchange.sendResponseHeaders(400, body.length);
+      exchange.getResponseBody().write(body);
+      exchange.close();
+    });
+    server.start();
+    try {
+      var client = new A2AErrorDetectingHttpClient(new JdkA2AHttpClient(HttpClient.newHttpClient()));
+      var error = assertThrows(RemoteA2AErrorException.class, () -> client.createPost()
+          .url("http://127.0.0.1:" + server.getAddress().getPort()).body("{}").post());
+      assertEquals(400, error.getHttpStatus());
+      assertEquals("INVALID_PARAMS", error.getReason());
+    } finally {
+      server.stop(0);
+    }
+  }
+}
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteProblemResponseTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteProblemResponseTest.java
deleted file mode 100644
index 4d1caba..0000000
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteProblemResponseTest.java
+++ /dev/null
@@ -1,152 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- *    Licensed under the Apache License, Version 2.0 (the "License"); you may
- *    not use this file except in compliance with the License. You may obtain
- *    a copy of the License at
- *
- *         http://www.apache.org/licenses/LICENSE-2.0
- *
- *    Unless required by applicable law or agreed to in writing, software
- *    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- *    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- *    License for the specific language governing permissions and limitations
- *    under the License.
- */
-package dev.openan.workflow.engine.client;
-
-import static org.junit.jupiter.api.Assertions.*;
-
-import com.sun.net.httpserver.HttpServer;
-import java.net.InetSocketAddress;
-import java.net.http.HttpClient;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.a2aproject.sdk.client.http.JdkA2AHttpClient;
-import org.a2aproject.sdk.spec.*;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
-
-class RemoteProblemResponseTest {
-  static String problem(int status) {
-    return "{\"status\":" + status + ",\"detail\":\"该端口下未发现业务信息\","
-        + "\"title\":\"Invalid Params\",\"type\":\"\",\"timestamp\":\"2026-08-31T09:07:36Z\"}";
-  }
-
-  @Test
-  void preservesProblemFieldsWithoutInspectingNestedBusinessData() {
-    var error = RemoteProblemException.fromPayload(problem(400));
-    assertNotNull(error);
-    assertEquals(400, error.getStatus());
-    assertEquals("Invalid Params", error.getTitle());
-    assertEquals("该端口下未发现业务信息", error.getDetail());
-    assertEquals("", error.getType());
-    assertEquals("2026-08-31T09:07:36Z", error.getTimestamp());
-    assertNull(RemoteProblemException.fromPayload("{\"artifactUpdate\":{\"data\":" + problem(400) + "}}"));
-    assertNull(RemoteProblemException.fromPayload("{\"status\":{\"state\":\"TASK_STATE_WORKING\"}}"));
-    assertNull(RemoteProblemException.fromPayload("{\"status\":400}"));
-    assertNull(RemoteProblemException.fromPayload(problem(200)));
-    assertNull(RemoteProblemException.fromPayload("{\"status\":429,"));
-  }
-
-  @ParameterizedTest
-  @ValueSource(ints = {400, 429, 503})
-  void directRuntimeFailsPromptlyWithoutWaitingForSseClosureOrRetrying(int status) throws Exception {
-    var release = new CountDownLatch(1);
-    var disconnected = new CountDownLatch(1);
-    var requests = new AtomicInteger();
-    HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
-    server.createContext("/", exchange -> {
-      requests.incrementAndGet();
-      exchange.getRequestBody().readAllBytes();
-      exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
-      exchange.sendResponseHeaders(200, 0);
-      try {
-        String payload = status == 503
-            ? problem(status).replace("该端口下未发现业务信息", "connection closed locally; request rejected")
-            : problem(status);
-        byte[] body = ("data: " + payload + "\n\n").getBytes(StandardCharsets.UTF_8);
-        exchange.getResponseBody().write(body, 0, 12);
-        exchange.getResponseBody().flush();
-        exchange.getResponseBody().write(body, 12, body.length - 12);
-        exchange.getResponseBody().flush();
-        for (int i = 0; i < 80 && !release.await(100, TimeUnit.MILLISECONDS); i++) {
-          exchange.getResponseBody().write(": keep-alive\n\n".getBytes(StandardCharsets.UTF_8));
-          exchange.getResponseBody().flush();
-        }
-      } catch (java.io.IOException clientClosed) {
-        disconnected.countDown();
-      } catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-      } finally {
-        exchange.close();
-      }
-    });
-    server.start();
-    var runtime = new DefaultA2AJavaClientRuntime(true, null, 20, "HTTP+JSON");
-    try {
-      AgentCard card = AgentCard.builder().name("remote-test").description("test").version("1")
-          .capabilities(AgentCapabilities.builder().streaming(true).build())
-          .defaultInputModes(List.of("text/plain")).defaultOutputModes(List.of("text/plain"))
-          .skills(List.of()).supportedInterfaces(List.of(new AgentInterface("HTTP+JSON",
-              "http://127.0.0.1:" + server.getAddress().getPort() + "/a2a/json"))).build();
-      MessageSendParams params = MessageSendParams.builder().message(Message.builder()
-          .role(Message.Role.ROLE_USER).messageId("test").parts(new TextPart("diagnose")).build()).build();
-      var events = new AtomicInteger();
-      var response = CompletableFuture.supplyAsync(
-          () -> runtime.sendMessage(card, params, null, event -> events.incrementAndGet(), null));
-      var thrown = assertThrows(java.util.concurrent.ExecutionException.class,
-          () -> response.get(3, TimeUnit.SECONDS));
-      Throwable cause = thrown;
-      while (cause.getCause() != null && !(cause instanceof RemoteProblemException)) cause = cause.getCause();
-      var problem = assertInstanceOf(RemoteProblemException.class, cause);
-      assertEquals(status, problem.getStatus());
-      assertEquals(0, events.get());
-      assertEquals(1, requests.get());
-      assertTrue(disconnected.await(3, TimeUnit.SECONDS), "failed SSE must close its HTTP connection");
-    } finally {
-      release.countDown();
-      runtime.close();
-      server.stop(0);
-    }
-  }
-
-  @Test
-  void causeLookupHandlesNullCyclesAndWrappedProblems() {
-    var problem = RemoteProblemException.fromPayload(problem(429));
-    assertSame(problem, RemoteProblemException.findIn(new IllegalStateException("wrapper", problem)));
-    assertNull(RemoteProblemException.findIn(null));
-    var one = new IllegalStateException("one");
-    var two = new IllegalStateException("two", one);
-    one.initCause(two);
-    assertNull(RemoteProblemException.findIn(one));
-  }
-
-  @Test
-  void nonStreamingResponsesPreserveTheRemoteBusinessFailure() throws Exception {
-    HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
-    server.createContext("/", exchange -> {
-      byte[] body = problem(400).getBytes(StandardCharsets.UTF_8);
-      exchange.sendResponseHeaders(400, body.length);
-      exchange.getResponseBody().write(body);
-      exchange.close();
-    });
-    server.start();
-    try {
-      var client = new ProblemDetectingHttpClient(new JdkA2AHttpClient(HttpClient.newHttpClient()));
-      var error = assertThrows(RemoteProblemException.class, () -> client.createPost()
-          .url("http://127.0.0.1:" + server.getAddress().getPort()).body("{}").post());
-      assertEquals(400, error.getStatus());
-    } finally {
-      server.stop(0);
-    }
-  }
-}
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteProblemFailureTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java
similarity index 61%
rename from workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteProblemFailureTest.java
rename to workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java
index ac07d5a..78ac0e6 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteProblemFailureTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java
@@ -5,23 +5,23 @@
  * SPDX-License-Identifier: Apache-2.0
  *
  *    Licensed under the Apache License, Version 2.0 (the "License"); you may
- *    not use this file except in compliance with the License. You may obtain
- *    a copy of the License at
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
  *
  *         http://www.apache.org/licenses/LICENSE-2.0
  *
- *    Unless required by applicable law or agreed to in writing, software
- *    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- *    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- *    License for the specific language governing permissions and limitations
- *    under the License.
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
  */
 package dev.openan.workflow.engine.core;
 
 import static org.junit.jupiter.api.Assertions.*;
 
 import dev.openan.workflow.engine.StubWorkflowEngineClient;
-import dev.openan.workflow.engine.client.RemoteProblemException;
+import dev.openan.workflow.engine.client.RemoteA2AErrorException;
 import dev.openan.workflow.engine.control.ControlPoint;
 import dev.openan.workflow.engine.model.*;
 import java.util.List;
@@ -33,31 +33,41 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
-class RemoteProblemFailureTest {
+class RemoteA2AErrorFailureTest {
+  private static String error(int status) {
+    String canonical = status == 429 ? "RESOURCE_EXHAUSTED" : "INVALID_ARGUMENT";
+    String reason = status == 429 ? "ACTIVE_TASK_LIMIT_EXCEEDED" : "INVALID_PARAMS";
+    return "{\"error\":{\"code\":" + status + ",\"status\":\"" + canonical
+        + "\",\"message\":\"Remote agent rejected the request\",\"details\":[{"
+        + "\"@type\":\"type.googleapis.com/google.rpc.ErrorInfo\",\"reason\":\""
+        + reason + "\",\"domain\":\"example.invalid\"}]}}";
+  }
+
   @ParameterizedTest
   @ValueSource(ints = {400, 429})
-  void waitsForTheOtherCityAndSkipsMergeWhileReportingCorrelatedFailure(int status) throws Exception {
+  void waitsForTheOtherAgentAndSkipsMergeWhileReportingCorrelatedFailure(int status)
+      throws Exception {
     var started = new java.util.concurrent.CountDownLatch(2);
     var reported = new java.util.concurrent.CountDownLatch(1);
-    var city1 = new CompletableFuture<SendMessageResult>();
-    var city2 = new CompletableFuture<SendMessageResult>();
+    var agent1 = new CompletableFuture<SendMessageResult>();
+    var agent2 = new CompletableFuture<SendMessageResult>();
     var observed = new java.util.concurrent.CopyOnWriteArrayList<Map<String, Object>>();
     var client = new StubWorkflowEngineClient() {
       @Override public CompletableFuture<SendMessageResult> dispatch(
           TaskRequest request, MessageContent content, ControlPoint callbacks) {
         started.countDown();
-        return request.getAgentName().equals("city1") ? city1 : city2;
+        return request.getAgentName().equals("agent1") ? agent1 : agent2;
       }
     };
     var merges = new AtomicInteger();
     var callbacks = ControlPoint.builder()
-        .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("diagnose")))
+        .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("run task")))
         .onSelfTask(request -> {
           merges.incrementAndGet();
           return CompletableFuture.completedFuture(TaskResult.success(List.of("must not merge")));
         }).build();
     var workflow = Workflow.builder().name("parallel-failure").steps(List.of(
-        city("city1"), city("city2"),
+        remote("agent1"), remote("agent2"),
         WorkflowStep.builder().name("merge").stepType(StepType.SELF_LOOP)
             .subtasks(List.of(Task.builder().agent("host").description("aggregate").build()))
             .next(List.of()).build())).build();
@@ -72,17 +82,16 @@ void waitsForTheOtherCityAndSkipsMergeWhileReportingCorrelatedFailure(int status
         }
       }
     };
-    var execution = new WorkflowExecutor(workflow, callbacks, client, events, "complaint", "zh").run();
+    var execution = new WorkflowExecutor(workflow, callbacks, client, events, "request", "en").run();
     try {
       assertTrue(started.await(5, java.util.concurrent.TimeUnit.SECONDS));
-      city1.completeExceptionally(RemoteProblemException.fromPayload(
-          "{\"status\":" + status + ",\"detail\":\"OMC拒绝请求\"}"));
+      agent1.completeExceptionally(RemoteA2AErrorException.fromPayload(error(status)));
       assertTrue(reported.await(5, java.util.concurrent.TimeUnit.SECONDS));
-      assertFalse(execution.isDone(), "The other in-flight diagnosis must be collected");
-      assertFalse(city2.isCancelled());
-      city2.complete(SendMessageResult.builder().taskState("TASK_STATE_COMPLETED")
+      assertFalse(execution.isDone(), "The other in-flight task must be collected");
+      assertFalse(agent2.isCancelled());
+      agent2.complete(SendMessageResult.builder().taskState("TASK_STATE_COMPLETED")
           .receivedMessages(List.of(new ReceivedMessage(
-              MessageContent.text("city2 diagnosis"), Map.of(), List.of()))).build());
+              MessageContent.text("agent2 result"), Map.of(), List.of()))).build());
       var result = execution.get(5, java.util.concurrent.TimeUnit.SECONDS);
       assertFalse(result.isSuccess());
       assertEquals(2, result.getHistory().size());
@@ -95,67 +104,68 @@ void waitsForTheOtherCityAndSkipsMergeWhileReportingCorrelatedFailure(int status
       }
       assertEquals(observed.get(0).get("executionId"), observed.get(1).get("executionId"));
       assertNotEquals(observed.get(0).get("taskId"), observed.get(1).get("taskId"));
-      var success = result.getHistory().stream().filter(h -> h.get("agent").equals("city2")).findFirst().orElseThrow();
+      var success = result.getHistory().stream()
+          .filter(h -> h.get("agent").equals("agent2")).findFirst().orElseThrow();
       assertEquals("success", success.get("status"));
-      assertEquals(List.of("city2 diagnosis"), success.get("outputs"));
+      assertEquals(List.of("agent2 result"), success.get("outputs"));
     } finally {
-      city1.cancel(true);
-      city2.cancel(true);
+      agent1.cancel(true);
+      agent2.cancel(true);
       execution.cancel(true);
     }
   }
 
-  private static WorkflowStep city(String name) {
+  private static WorkflowStep remote(String name) {
     return WorkflowStep.builder().name(name).stepType(StepType.ALL_SUCCESS)
-        .subtasks(List.of(Task.builder().agent(name).description("diagnose").build()))
+        .subtasks(List.of(Task.builder().agent(name).description("execute").build()))
         .next(List.of(new JumpCondition("merge", ""))).build();
   }
+
   @ParameterizedTest
   @ValueSource(ints = {400, 429})
-  void remoteFailureReachesWorkflowHistoryWithoutNegotiationOrSuccessfulOutputs(int status) {
-    var problem = RemoteProblemException.fromPayload(
-        "{\"status\":" + status + ",\"detail\":\"OMC拒绝请求\",\"type\":\"\"}");
+  void protocolFailureReachesWorkflowHistoryWithoutNegotiationOrSuccessfulOutputs(int status) {
+    var remoteError = RemoteA2AErrorException.fromPayload(error(status));
     var calls = new AtomicInteger();
     var negotiations = new AtomicInteger();
-    var client = new StubWorkflowEngineClient("omc") {
+    var client = new StubWorkflowEngineClient("remote-agent") {
       @Override
       public CompletableFuture<SendMessageResult> dispatch(
           TaskRequest request, MessageContent content, ControlPoint callbacks) {
         calls.incrementAndGet();
-        return CompletableFuture.failedFuture(new CompletionException(problem));
+        return CompletableFuture.failedFuture(new CompletionException(remoteError));
       }
     };
-    var step = WorkflowStep.builder().name("diagnose").stepType(StepType.ALL_SUCCESS)
-        .subtasks(List.of(Task.builder().agent("omc").description("diagnose").build()))
+    var step = WorkflowStep.builder().name("remote").stepType(StepType.ALL_SUCCESS)
+        .subtasks(List.of(Task.builder().agent("remote-agent").description("execute").build()))
         .next(List.of()).build();
     var workflow = Workflow.builder().name("remote-failure").steps(List.of(step)).build();
     var callbacks = ControlPoint.builder()
-        .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("diagnose")))
+        .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("execute")))
         .onNegotiation(request -> {
           negotiations.incrementAndGet();
           return CompletableFuture.failedFuture(new AssertionError("must not negotiate"));
         }).build();
-    var result = new WorkflowExecutor(workflow, callbacks, client, null, "complaint", "zh").run().join();
+    var result = new WorkflowExecutor(workflow, callbacks, client, null, "request", "en").run().join();
     assertFalse(result.isSuccess());
     assertEquals(1, calls.get());
     assertEquals(0, negotiations.get());
     var history = result.getHistory().get(0);
     assertEquals("failed", history.get("status"));
-    assertEquals("remote.problem." + status, history.get("errorCode"));
+    assertEquals(status == 429 ? "a2a.active_task_limit_exceeded" : "a2a.invalid_params",
+        history.get("errorCode"));
     assertEquals(List.of(), history.get("outputs"));
-    assertTrue(history.get("error").toString().contains("OMC拒绝请求"));
-    assertEquals(status, ((Map<?, ?>) history.get("errorDetails")).get("status"));
-    assertEquals("", ((Map<?, ?>) history.get("errorDetails")).get("type"));
+    assertTrue(history.get("error").toString().contains("rejected"));
+    assertEquals(status, ((Map<?, ?>) history.get("errorDetails")).get("httpStatus"));
+    assertNotNull(((Map<?, ?>) history.get("errorDetails")).get("details"));
   }
 
   @Test
   void knownSecretsAreRedactedAndUnrecognizedProviderErrorsRemainPrivate() {
-    var problem = RemoteProblemException.fromPayload(
-        "{\"status\":400,\"detail\":\"bad request password=private-value accessSession=private-session\"}");
-    var result = FailureMapping.from(problem);
+    var remoteError = RemoteA2AErrorException.fromPayload(error(400).replace(
+        "Remote agent rejected the request", "bad request password=private-value"));
+    var result = FailureMapping.from(remoteError);
     assertFalse(result.toString().contains("private-value"));
-    assertFalse(result.toString().contains("private-session"));
-    assertFalse(problem.getMessage().contains("private-value"));
+    assertFalse(remoteError.getMessage().contains("private-value"));
     assertEquals("IllegalStateException",
         FailureMapping.from(new IllegalStateException("raw private provider body")).getError());
   }