docs/a2a-protocol/IMPLEMENTATION_SUMMARY_EN.md

# Implementation Summary: EventMesh A2A Protocol v2.0 (MCP Edition)

## Key Achievements

The A2A protocol has been successfully refactored to adopt the **MCP (Model Context Protocol)** architecture, positioning EventMesh as a modern **Agent Collaboration Bus**.

### 1. Core Protocol Refactoring (`EnhancedA2AProtocolAdaptor`)
- **Hybrid Engine (JSON-RPC & CloudEvents)**: Implemented a smart parsing engine that supports:
    - **MCP/JSON-RPC 2.0**: For LLM-friendly, low-code integration.
    - **Native CloudEvents**: For advanced, protocol-compliant integration.
    - The adaptor automatically delegates processing based on the payload content (`jsonrpc` detection).
- **Async RPC Mapping**: Established a bridge between synchronous RPC semantics and asynchronous Event-Driven Architecture (EDA).
    - **Requests** map to `*.req` events with `mcptype=request`.
    - **Responses** map to `*.resp` events with `mcptype=response`.
    - **Correlation** is handled by mapping JSON-RPC `id` to CloudEvent `collaborationid`.
- **Routing Optimization**: Implemented "Deep Body Routing" extraction:
    - `params._agentId` -> CloudEvent Extension `targetagent` (P2P).
    - `params._topic` -> CloudEvent Subject (Pub/Sub).

### 2. Native Pub/Sub & Streaming
- **Pub/Sub**: Added support for O(1) broadcast complexity by mapping `_topic` to CloudEvent Subject.
- **Streaming**: Added support for `message/sendStream` operation, mapping to `.stream` event type and preserving sequence via `_seq` -> `seq` extension.

### 3. Standardization & Compatibility
- **Models**: Defined `JsonRpcRequest`, `JsonRpcResponse`, `JsonRpcError` POJOs compliant with JSON-RPC 2.0 spec.
- **Methods**: Introduced `McpMethods` constants for standard operations like `tools/call`, `resources/read`.
- **AgentCard Models**: Implemented `AgentCard`, `AgentSkill`, `AgentInterface`, `AgentCapabilities` for complete agent capability description.

### 4. Gateway Runtime Architecture (`eventmesh-runtime`)

A standalone HTTP Gateway service bridging external clients to the A2A event bus.

#### Core Components

| Component | Responsibility |
| :--- | :--- |
| `A2AGatewayServer` | Netty HTTP server entry point, pre-registers mock agents, wires all components |
| `A2AGatewayHttpHandler` | HTTP request router, supports SSE streaming responses |
| `A2AGatewayService` | Core orchestration: task submission, response handling, status subscription, SSE push |
| `TaskRegistry` | In-memory task state machine + TTL auto-cleanup |
| `A2APublishSubscribeService` | AgentCard registration, discovery, heartbeat management |
| `InMemoryA2AMessageTransport` | In-memory pub/sub (replaceable by EventMesh broker) |
| `A2ACardHttpHandler` | AgentCard CRUD REST endpoints |
| `A2AClient` | Java SDK with typed API |

#### REST API

| Method | Path | Description |
| :--- | :--- | :--- |
| `POST` | `/a2a/tasks?mode=sync` | Submit task synchronously |
| `POST` | `/a2a/tasks?mode=async` | Submit task asynchronously |
| `GET` | `/a2a/tasks/{taskId}` | Get task status |
| `DELETE` | `/a2a/tasks/{taskId}` | Cancel a task |
| `GET` | `/a2a/tasks/{taskId}/wait` | Long-poll wait for result |
| `GET` | `/a2a/tasks/{taskId}/stream` | SSE stream of task status updates |
| `GET` | `/a2a/agents` | List registered agents |
| `POST` | `/a2a/heartbeat` | Agent heartbeat |
| `GET` | `/a2a/cards/list` | List all AgentCards |
| `POST` | `/a2a/cards/card/{org}/{unit}/{agent}` | Register an AgentCard |

### 5. Key Improvements

#### 5.1 TaskRegistry TTL Auto-Cleanup
- **Problem**: Terminal-state tasks (COMPLETED/FAILED/CANCELLED) accumulate indefinitely, causing memory leaks.
- **Solution**: A daemon `ScheduledExecutorService` runs every 60 seconds, removing terminal-state tasks older than the TTL (default: 5 minutes).
- **Configuration**: `TaskRegistry(taskTtlMs, cleanupIntervalMs)` constructor allows custom tuning.

#### 5.2 Race Condition Fix
- **Problem**: `InMemoryTransport` delivers messages synchronously. If `transport.publish()` executes before `pendingTasks.put()`, `handleResponse()` runs before `put()` and the future never completes.
- **Solution**: Ensure `pendingTasks.put(taskId, future)` is called **before** `transport.publish()`, with comments documenting the ordering importance.

#### 5.3 A2AClient Typed Returns
- **Improvement**: `getTaskStatus()` returns `TaskResult` object (instead of raw JSON string), `listAgents()` returns `List<String>` (instead of raw JSON).
- **Compatibility**: `TaskResult.data` field uses `@JsonAlias("result")` annotation to handle the server's `result` field name.

#### 5.4 SSE Streaming Response
- **Endpoint**: `GET /a2a/tasks/{taskId}/stream`
- **Implementation**: Handler writes `DefaultHttpContent` chunks directly to the Netty channel, returning `null` to skip the standard `FullHttpResponse` path. Uses `StatusSubscriber` callbacks for real-time state push.

#### 5.5 Usage Documentation
- Created `eventmesh-examples/.../demo/README.md` with architecture diagram, API table, curl examples, SDK usage, and run instructions.

### 6. Testing & Quality
- **Protocol Unit Tests**: `EnhancedA2AProtocolAdaptorTest` covers Request/Response cycles, Error handling, Notifications, and Batching.
- **Topic Utility Tests**: `A2ATopicFactoryTest` covers topic generation and parsing.
- **Gateway Runtime Tests**:
    - `TaskRegistryTest` — Task state machine + TTL cleanup verification
    - `InMemoryA2AMessageTransportTest` — In-memory transport delivery
    - `A2AGatewayServiceTest` — Gateway service layer
    - `A2AGatewayEndToEndTest` — In-process end-to-end flow
    - `A2AClientServerIntegrationTest` — Real HTTP client-server integration test
- **Integration Demos**: `McpIntegrationDemoTest`, `McpPatternsIntegrationTest`, `McpComprehensiveDemoTest`, `CloudEventsComprehensiveDemoTest`
- **Total**: 73 test scenarios, all passing.

## Next Steps

1. **EventMesh Broker Integration**: Replace `InMemoryA2AMessageTransport` with the real EventMesh broker for production deployment.
2. **Router Integration**: Update EventMesh Runtime Router to leverage `targetagent` and `a2amethod` extension attributes for advanced routing rules.
3. **Schema Registry**: Implement a "Registry Agent" that allows agents to publish their MCP capabilities (`methods/list`) dynamically.
4. **Sidecar Support**: Expose the A2A adaptor logic in the Sidecar proxy to allow non-Java agents (Python, Node.js) to interact via simple HTTP/JSON.
5. **WebSocket Streaming**: Extend SSE to bidirectional WebSocket for real-time agent dialogue.
6. **Task Persistence**: Persist `TaskRegistry` state to a durable store (Redis/DB) for crash recovery.
7. **Authentication**: Add API key / JWT authentication to the Gateway REST API.