apache
diff --git a/‎docs/a2a-protocol/ARCHITECTURE.md‎
Lines changed: 164 additions & 3 deletions b/‎docs/a2a-protocol/ARCHITECTURE.md‎
Lines changed: 164 additions & 3 deletions
diff --git a/‎docs/a2a-protocol/IMPLEMENTATION_SUMMARY.md‎
Lines changed: 74 additions & 8 deletions b/‎docs/a2a-protocol/IMPLEMENTATION_SUMMARY.md‎
Lines changed: 74 additions & 8 deletions
@@ -64,16 +64,80 @@ graph TD
 
 ### 3.2 Component Design (`eventmesh-protocol-a2a`)
 
-The core logic resides in the `eventmesh-protocol-plugin` module.
+The core protocol logic resides in the `eventmesh-protocol-plugin` module.
 
 *   **`EnhancedA2AProtocolAdaptor`**: The central brain of the protocol.
     *   **Intelligent Parsing**: Automatically detects message format (MCP vs. Raw CloudEvent).
     *   **Protocol Delegation**: Delegates to `CloudEvents` or `HTTP` adaptors when necessary.
     *   **Semantic Mapping**: Transforms JSON-RPC methods and IDs into CloudEvent attributes.
 *   **`A2AProtocolConstants`**: Defines standard operations like `task/get`, `message/sendStream`.
 *   **`JsonRpc*` Models**: Strictly typed POJOs for JSON-RPC 2.0 compliance.
+*   **`AgentCard` / `AgentSkill` / `AgentInterface`**: Agent capability discovery models.
+*   **`A2ATopicFactory`**: Topic naming and parsing utility for request/response/status topics.
+*   **`A2AClient`**: Java SDK for agent developers — AgentCard registration, task submission (sync/async), task status query, heartbeat, and transport-based request handling. Returns typed `TaskResult` objects.
+*   **`A2AMessageTransport`**: Transport-agnostic pub/sub interface (InMemory implementation for dev/testing).
 
-### 3.3 Asynchronous RPC Mapping ( The "Async Bridge" )
+### 3.3 Gateway Runtime Architecture (`eventmesh-runtime`)
+
+The Gateway runtime provides a standalone Netty HTTP server bridging external clients to the A2A event bus.
+
+```mermaid
+graph TD
+    Client["Client / A2AClient SDK"] -- "HTTP REST" --> Server["A2AGatewayServer<br/>(Netty HTTP)"]
+    Server --> Handler["A2AGatewayHttpHandler"]
+    Handler --> GwService["A2AGatewayService"]
+    GwService --> Registry["TaskRegistry<br/>(state machine + TTL)"]
+    GwService --> Transport["InMemoryA2AMessageTransport"]
+    GwService --> PubSub["A2APublishSubscribeService<br/>(AgentCard discovery)"]
+    Transport -- "publish/subscribe" --> Agent["Target Agent"]
+    Agent -- "response event" --> Transport
+    Transport --> GwService
+
+    style Server fill:#f9f,stroke:#333,stroke-width:2px
+    style Registry fill:#cfc,stroke:#333
+    style Transport fill:#ccf,stroke:#333
+```
+
+#### Core Components
+
+| Component | Module | Responsibility |
+| :--- | :--- | :--- |
+| `A2AGatewayServer` | runtime | Standalone Netty HTTP server entry point. Pre-registers mock agents, wires all components. |
+| `A2AGatewayHttpHandler` | runtime | HTTP request router. Maps REST endpoints to service calls. Supports SSE streaming. |
+| `A2AGatewayService` | runtime | Core orchestration: task submission, response handling, status subscription, SSE push. |
+| `TaskRegistry` | runtime | In-memory task lifecycle state machine with TTL auto-cleanup. |
+| `A2APublishSubscribeService` | runtime | AgentCard registration, discovery, and heartbeat management. |
+| `InMemoryA2AMessageTransport` | runtime | In-memory pub/sub (replaceable by EventMesh broker). |
+| `A2ACardHttpHandler` | runtime | AgentCard CRUD REST endpoints (`/a2a/cards/*`). |
+| `A2AClient` | protocol-a2a | Java SDK for agent developers (HTTP + transport). |
+
+#### Task Lifecycle State Machine
+
+```
+SUBMITTED → WORKING → COMPLETED
+                    ↘ FAILED
+                    ↘ CANCELLED
+```
+
+*   **TTL Auto-Cleanup**: Terminal-state tasks are automatically removed after a configurable TTL (default: 5 minutes). A daemon thread runs cleanup every 60 seconds.
+*   **Race Condition Prevention**: `pendingTasks.put(taskId, future)` is called **before** `transport.publish()` to ensure the future is registered before any synchronous delivery could trigger `handleResponse()`.
+
+#### REST API
+
+| Method | Path | Description |
+| :--- | :--- | :--- |
+| `POST` | `/a2a/tasks?mode=sync` | Submit task synchronously (wait for result) |
+| `POST` | `/a2a/tasks?mode=async` | Submit task asynchronously (return taskId immediately) |
+| `GET` | `/a2a/tasks/{taskId}` | Get task status and result |
+| `DELETE` | `/a2a/tasks/{taskId}` | Cancel a task |
+| `GET` | `/a2a/tasks/{taskId}/wait` | Long-poll wait for task result |
+| `GET` | `/a2a/tasks/{taskId}/stream` | **SSE** stream of task status updates |
+| `GET` | `/a2a/agents` | List all registered agents |
+| `POST` | `/a2a/heartbeat` | Agent heartbeat |
+| `GET` | `/a2a/cards/list` | List all AgentCards |
+| `POST` | `/a2a/cards/card/{org}/{unit}/{agent}` | Register an AgentCard |
+
+### 3.4 Asynchronous RPC Mapping ( The "Async Bridge" )
 
 To support MCP on an Event Bus, synchronous RPC concepts are mapped to asynchronous events:
 
@@ -113,6 +177,18 @@ To support MCP on an Event Bus, synchronous RPC concepts are mapped to asynchron
 *   **Operation**: `message/sendStream`
 *   **Mechanism**: Maps to `.stream` event type and preserves sequence order via `seq` extension attribute.
 
+#### D. SSE Task Streaming (Gateway)
+*   **Endpoint**: `GET /a2a/tasks/{taskId}/stream`
+*   **Mechanism**: Server-Sent Events (`text/event-stream`) pushes real-time task state transitions.
+*   **Flow**: Initial state → WORKING updates → terminal state → connection close.
+*   **Implementation**: Handler writes `DefaultHttpContent` chunks directly to the Netty channel, returning `null` to skip the standard `FullHttpResponse` path.
+
+#### E. Task TTL Auto-Cleanup (Gateway)
+*   Terminal-state tasks are automatically removed by a daemon scheduler after a configurable TTL (default: 5 minutes, cleanup interval: 60 seconds), preventing memory leaks.
+
+#### F. AgentCard Discovery & Heartbeat (Gateway)
+*   AgentCards expire after 60 seconds without heartbeat. `POST /a2a/heartbeat` refreshes the last-seen timestamp.
+
 ## 5. Usage Examples
 
 ### 5.1 Sending a Tool Call (Request)
@@ -149,7 +225,92 @@ To support MCP on an Event Bus, synchronous RPC concepts are mapped to asynchron
 *   `subject`: `market.crypto.btc`
 *   `targetagent`: (Empty)
 
+### 5.3 Gateway REST API (HTTP)
+
+The A2A Gateway provides a REST API for external clients and non-Java agents.
+
+#### 5.3.1 Submit Task (Sync)
+
+```bash
+curl -X POST 'http://localhost:10105/a2a/tasks?mode=sync' \
+  -H 'Content-Type: application/json' \
+  -d '{"targetAgent":"weather-agent","message":"Beijing"}'
+```
+
+Response:
+```json
+{
+  "taskId": "task-a1b2c3d4",
+  "state": "COMPLETED",
+  "data": "The weather in Beijing is sunny, 25°C"
+}
+```
+
+#### 5.3.2 Submit Task (Async)
+
+```bash
+curl -X POST 'http://localhost:10105/a2a/tasks?mode=async' \
+  -H 'Content-Type: application/json' \
+  -d '{"targetAgent":"weather-agent","message":"Shanghai"}'
+```
+
+#### 5.3.3 SSE Stream
+
+```bash
+curl -N http://localhost:10105/a2a/tasks/{taskId}/stream
+```
+
+Response (`text/event-stream`):
+```
+data: {"taskId":"task-a1b2c3d4","state":"SUBMITTED"}
+
+data: {"taskId":"task-a1b2c3d4","state":"WORKING","data":"processing..."}
+
+data: {"taskId":"task-a1b2c3d4","state":"completed","data":"result..."}
+```
+
+#### 5.3.4 List Agents
+
+```bash
+curl http://localhost:10105/a2a/agents
+```
+
+### 5.4 A2AClient SDK (Java)
+
+```java
+A2AClient client = A2AClient.builder()
+    .gatewayUrl("http://localhost:10105")
+    .namespace("global")
+    .agentName("my-agent")
+    .agentCard(card)
+    .heartbeatInterval(30_000)
+    .build();
+
+client.start();
+
+// Synchronous task (returns typed TaskResult)
+TaskResult result = client.sendTaskSync("weather-agent", "Beijing", null);
+
+// Asynchronous task (returns taskId immediately)
+String taskId = client.sendTaskAsync("weather-agent", "Shanghai", null);
+
+// Poll status
+TaskResult status = client.getTaskStatus(taskId);
+
+// Cancel
+boolean cancelled = client.cancelTask(taskId);
+
+// List registered agents (typed List<String>)
+List<String> agents = client.listAgents();
+
+client.shutdown();
+```
+
 ## 6. Future Roadmap
 
+*   **EventMesh Broker Integration**: Replace `InMemoryA2AMessageTransport` with the real EventMesh broker for production deployment.
 *   **Schema Registry**: Implement dynamic discovery of Agent capabilities via `methods/list`.
-*   **Sidecar Injection**: Fully integrate the adaptor into the EventMesh Sidecar.
+*   **Sidecar Injection**: Fully integrate the adaptor into the EventMesh Sidecar.
+*   **WebSocket Streaming**: Extend SSE to bidirectional WebSocket for real-time agent dialogue.
+*   **Task Persistence**: Persist `TaskRegistry` state to a durable store for crash recovery.
+*   **Authentication**: Add API key / JWT authentication to the Gateway REST API.
@@ -13,7 +13,7 @@ A2A 协议已成功重构为采用 **MCP (Model Context Protocol)** 架构，将
     - **请求 (Requests)** 映射为 `*.req` 事件，属性 `mcptype=request`。
     - **响应 (Responses)** 映射为 `*.resp` 事件，属性 `mcptype=response`。
     - **关联 (Correlation)** 通过将 JSON-RPC `id` 映射到 CloudEvent `collaborationid` 来处理。
-- **路由优化**: 实现了“深度内容路由提取”：
+- **路由优化**: 实现了"深度内容路由提取"：
     - `params._agentId` -> CloudEvent 扩展属性 `targetagent` (P2P)。
     - `params._topic` -> CloudEvent Subject (Pub/Sub)。
 
@@ -24,14 +24,80 @@ A2A 协议已成功重构为采用 **MCP (Model Context Protocol)** 架构，将
 ### 3. 标准化与兼容性
 - **数据模型**: 定义了符合 JSON-RPC 2.0 规范的 `JsonRpcRequest`、`JsonRpcResponse`、`JsonRpcError` POJO 对象。
 - **方法定义**: 引入了 `McpMethods` 常量，支持标准操作如 `tools/call`、`resources/read`。
+- **AgentCard 模型**: 实现了 `AgentCard`、`AgentSkill`、`AgentInterface`、`AgentCapabilities` 等完整的 Agent 能力描述模型。
 
-### 4. 测试与质量
-- **单元测试**: 在 `EnhancedA2AProtocolAdaptorTest` 中实现了对请求/响应循环、错误处理、通知和批处理的全面覆盖。
-- **集成演示**: `McpIntegrationDemoTest` 模拟了 P2P RPC 闭环。
-- **模式测试**: `McpPatternsIntegrationTest` 模拟了 Pub/Sub 和 Streaming 流程。
+### 4. Gateway 运行时架构 (`eventmesh-runtime`)
+
+完整的独立 HTTP Gateway 服务，桥接外部客户端到 A2A 事件总线。
+
+#### 核心组件
+
+| 组件 | 职责 |
+| :--- | :--- |
+| `A2AGatewayServer` | Netty HTTP 服务器入口，预注册 mock agent，组装所有组件 |
+| `A2AGatewayHttpHandler` | HTTP 请求路由，支持 SSE 流式响应 |
+| `A2AGatewayService` | 核心编排：任务提交、响应处理、状态订阅、SSE 推送 |
+| `TaskRegistry` | 内存任务状态机 + TTL 自动清理 |
+| `A2APublishSubscribeService` | AgentCard 注册、发现、心跳管理 |
+| `InMemoryA2AMessageTransport` | 内存 pub/sub 实现（可替换为 EventMesh broker） |
+| `A2ACardHttpHandler` | AgentCard CRUD REST 端点 |
+| `A2AClient` | Java SDK，提供类型化 API |
+
+#### REST API
+
+| 方法 | 路径 | 说明 |
+| :--- | :--- | :--- |
+| `POST` | `/a2a/tasks?mode=sync` | 同步提交任务 |
+| `POST` | `/a2a/tasks?mode=async` | 异步提交任务 |
+| `GET` | `/a2a/tasks/{taskId}` | 查询任务状态 |
+| `DELETE` | `/a2a/tasks/{taskId}` | 取消任务 |
+| `GET` | `/a2a/tasks/{taskId}/wait` | 长轮询等待结果 |
+| `GET` | `/a2a/tasks/{taskId}/stream` | SSE 流式推送状态更新 |
+| `GET` | `/a2a/agents` | 列出已注册 agents |
+| `POST` | `/a2a/heartbeat` | Agent 心跳 |
+| `GET` | `/a2a/cards/list` | 列出所有 AgentCard |
+| `POST` | `/a2a/cards/card/{org}/{unit}/{agent}` | 注册 AgentCard |
+
+### 5. 关键改进
+
+#### 5.1 TaskRegistry TTL 自动清理
+- **问题**: 终态任务（COMPLETED/FAILED/CANCELLED）无限累积导致内存泄漏。
+- **方案**: 守护线程 `ScheduledExecutorService` 每 60 秒扫描一次，清理超过 TTL（默认 5 分钟）的终态任务。
+- **配置**: `TaskRegistry(taskTtlMs, cleanupIntervalMs)` 构造函数支持自定义调优。
+
+#### 5.2 竞态条件修复
+- **问题**: `InMemoryTransport` 同步投递消息，若 `transport.publish()` 在 `pendingTasks.put()` 之前执行，`handleResponse()` 会先于 `put()` 运行，导致 future 永不完成。
+- **方案**: 严格保证 `pendingTasks.put(taskId, future)` 在 `transport.publish()` 之前执行，并添加注释说明顺序重要性。
+
+#### 5.3 A2AClient 类型化返回
+- **改进**: `getTaskStatus()` 返回 `TaskResult` 对象（而非原始 JSON 字符串），`listAgents()` 返回 `List<String>`（而非原始 JSON）。
+- **兼容**: `TaskResult.data` 字段使用 `@JsonAlias("result")` 注解，兼容服务端 `result` 字段名。
+
+#### 5.4 SSE 流式响应
+- **端点**: `GET /a2a/tasks/{taskId}/stream`
+- **实现**: Handler 直接写入 Netty channel（`DefaultHttpContent` chunks），返回 `null` 跳过标准 `FullHttpResponse` 路径。通过 `StatusSubscriber` 回调实时推送状态变更。
+
+#### 5.5 使用文档
+- 新建 `eventmesh-examples/.../demo/README.md`，包含架构图、API 表、curl 示例、SDK 用法、运行方式。
+
+### 6. 测试与质量
+- **协议层单元测试**: `EnhancedA2AProtocolAdaptorTest` 覆盖请求/响应循环、错误处理、通知和批处理。
+- **Topic 工具测试**: `A2ATopicFactoryTest` 覆盖 topic 生成与解析。
+- **Gateway 运行时测试**:
+    - `TaskRegistryTest` — 任务状态机 + TTL 清理验证
+    - `InMemoryA2AMessageTransportTest` — 内存传输投递
+    - `A2AGatewayServiceTest` — Gateway 服务层
+    - `A2AGatewayEndToEndTest` — 进程内全链路
+    - `A2AClientServerIntegrationTest` — 真实 HTTP 客户端-服务端集成测试
+- **集成演示**: `McpIntegrationDemoTest`、`McpPatternsIntegrationTest`、`McpComprehensiveDemoTest`、`CloudEventsComprehensiveDemoTest`
+- **总计**: 73 个测试场景，全部通过。
 
 ## 下一步计划
 
-1. **路由集成**: 更新 EventMesh Runtime Router，利用新的 `targetagent` 和 `a2amethod` 扩展属性实现高级路由规则。
-2. **Schema 注册中心**: 实现一个“注册中心智能体 (Registry Agent)”，允许智能体动态发布其 MCP 能力 (`methods/list`)。
-3. **Sidecar 支持**: 将 A2A 适配器逻辑暴露在 Sidecar 代理中，允许非 Java 智能体 (Python, Node.js) 通过简单的 HTTP/JSON 进行交互。
+1. **EventMesh Broker 集成**: 用真实 EventMesh broker 替换 `InMemoryA2AMessageTransport`，实现生产级部署。
+2. **路由集成**: 更新 EventMesh Runtime Router，利用 `targetagent` 和 `a2amethod` 扩展属性实现高级路由规则。
+3. **Schema 注册中心**: 实现"注册中心智能体 (Registry Agent)"，允许智能体动态发布 MCP 能力 (`methods/list`)。
+4. **Sidecar 支持**: 将 A2A 适配器逻辑暴露在 Sidecar 代理中，允许非 Java 智能体通过 HTTP/JSON 交互。
+5. **WebSocket 流式**: 将 SSE 扩展为双向 WebSocket，支持实时 agent 对话。
+6. **任务持久化**: 将 `TaskRegistry` 状态持久化到 Redis/DB，支持崩溃恢复。
+7. **认证授权**: 为 Gateway REST API 添加 API Key / JWT 认证。