From cc28ff62ff733325ebb1f9b412a437c50ec838b7 Mon Sep 17 00:00:00 2001 From: qqeasonchen Date: Fri, 28 Aug 2026 15:18:36 +0800 Subject: [PATCH 1/4] feat(a2a): wire A2A Gateway onto Runtime via TaskStore (issue #5302 Sub-PR D1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brings the A2A Gateway back into the Runtime, this time with the unified control-plane stores from #5301 Sub-PR A/C as the durable backend instead of the in-memory TaskRegistry that PR #5260 introduced. What lands ---------- * A2AGatewayService — task lifecycle controller. Replaces the in-memory TaskRegistry with TaskStore (Sub-PR A/C). A small runtime cache holds the two fields TaskStore does not model: parentTaskId (used to render child-task queries) and taskEpoch (the per-task counter the store requires for stale-write rejection on updateStatus). Both caches are rebuildable from the store on a fresh JVM; the cache keys are taskIds, so the cache size tracks active tasks only. * A2AGatewayHttpHandler — Netty HTTP handler for the REST + SSE API (/a2a/tasks, /a2a/tasks/{id}, /a2a/tasks/{id}/stream, /a2a/health). SSE registers a StatusSubscriber so intermediate PENDING -> RUNNING -> COMPLETED transitions stream to the client. * A2AGatewayServer — Netty HTTP bootstrap. Production wires EventMeshA2ATransport (Runtime-bridged) for the A2AMessageTransport dependency; tests pass an in-process transport. The weather-agent demo from PR #5260 is removed — the demo client (A2AGatewayDemo) and the Meta-ized AgentCard registry land in Sub-PR D2. * AgentCardRegistry + InMemoryAgentCardRegistry — minimal discovery surface (isAgentRegistered / registerCard / getCard). A Meta-backed implementation backed by SessionStore (Sub-PR A) lands in D2. Status mapping (PR #5260 -> Sub-PR A/C) --------------------------------------- SUBMITTED -> PENDING, WORKING -> RUNNING, CANCELLED -> CANCELED (one L) The legacy wire vocabulary stays available via A2AGatewayService.TaskState and the toLegacyState() helper. JSON response payloads still emit SUBMITTED/WORKING/CANCELLED so existing A2A clients do not break. Documentation ------------- * docs/a2a-protocol/README.md carries an EXPERIMENTAL banner that lists the three pieces still missing for production (TaskExpirer, Meta-ized AgentCard, Testcontainers E2E) — all scheduled for Sub-PR D2. Acceptance against issue #5302 ------------------------------ * "A2A transport over the Runtime delivery path" — A2AGatewayService takes an A2AMessageTransport; production wires EventMeshA2ATransport which already bridges A2A onto UniIngressService (issue #5301 Sub-PR A's runtime ingress). The parallel InMemoryA2AMessageTransport is no longer required by the gateway path. * "A2A tasks persist through TaskStore and recover across restarts" — createTask / updateStatus / getTask all go through the persistent store. On a fresh JVM, a gateway that had pending tasks sees them with their last persisted status; in-flight work that was awaiting a response will be picked up on the next status callback. Tests (7, all passing) ---------------------- * A2AGatewayServiceTest — 6 tests: create+complete, cancel, cancel-on- unknown, submit-to-unregistered, parent-child index, task-not-found. Uses an in-process TaskStore (mirrors Sub-PR A's test stub) and an in-process A2AMessageTransport that handles A2A's + single-segment wildcards. * A2AGatewaySmokeTest — Netty HTTP /a2a/health loopback check. Not in D1 (Sub-PR D2) --------------------- * TaskExpirer reaper (periodic TaskStore.expireStale sweep). * Meta-backed AgentCardRegistry (SessionStore). * Testcontainers fault-injection E2E. --- docs/a2a-protocol/README.md | 21 +- .../runtime/a2a/A2AGatewayHttpHandler.java | 330 +++++++++++ .../runtime/a2a/A2AGatewayServer.java | 133 +++++ .../runtime/a2a/A2AGatewayService.java | 538 ++++++++++++++++++ .../runtime/a2a/AgentCardRegistry.java | 55 ++ .../a2a/InMemoryAgentCardRegistry.java | 59 ++ .../runtime/a2a/A2AGatewayServiceTest.java | 326 +++++++++++ .../runtime/a2a/A2AGatewaySmokeTest.java | 95 ++++ 8 files changed, 1556 insertions(+), 1 deletion(-) create mode 100644 eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java create mode 100644 eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServer.java create mode 100644 eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayService.java create mode 100644 eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/AgentCardRegistry.java create mode 100644 eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/InMemoryAgentCardRegistry.java create mode 100644 eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java create mode 100644 eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java diff --git a/docs/a2a-protocol/README.md b/docs/a2a-protocol/README.md index 6072b7252f..d33121fd73 100644 --- a/docs/a2a-protocol/README.md +++ b/docs/a2a-protocol/README.md @@ -1,6 +1,25 @@ # EventMesh A2A 协议（Agent-to-Agent Communication Protocol） -> 合并自 ARCHITECTURE.md + eventmesh-a2a-design.md + README.md + README_EN.md + IMPLEMENTATION_SUMMARY.md + IMPLEMENTATION_SUMMARY_EN.md + TEST_RESULTS.md（2026-08-13）。 +> **⚠️ EXPERIMENTAL — Issue #5302 D1 scope** +> +> The A2A Gateway is **Experimental** as of Sub-PR D1 (issue #5302). The gateway now persists +> tasks through the unified `TaskStore` (issue #5301 Sub-PR A/C) and bridges A2A publish/subscribe +> onto the Runtime via `EventMeshA2ATransport` — there is no longer a parallel in-memory transport. +> However, the following pieces land in follow-up PRs and are required before the gateway is +> suitable for production use: +> +> 1. **TaskExpirer reaper** (Sub-PR D2): periodic `TaskStore.expireStale()` sweep so terminal +> tasks do not accumulate in the Meta store. +> 2. **AgentCard Meta-ization** (Sub-PR D2): `A2APublishSubscribeService` still uses an +> in-memory `ConcurrentHashMap` for the agent-card registry; production needs a Meta-backed +> `SessionStore` (Sub-PR A) or equivalent. +> 3. **End-to-end Testcontainers test** (Sub-PR D2): fault-injection under a real Meta + Runtime +> wiring. +> +> Until all three land, treat the gateway as Experimental — wire it up against the +> `MetaBackedTaskStore` only on dev clusters. + +--- ## 目录 diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java new file mode 100644 index 0000000000..3295c5a8c3 --- /dev/null +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java @@ -0,0 +1,330 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import org.apache.eventmesh.protocol.a2a.A2AProtocolConstants; +import org.apache.eventmesh.protocol.a2a.A2ATopicFactory; +import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskState; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.FullHttpResponse; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpVersion; +import io.netty.handler.codec.http.QueryStringDecoder; + +import com.fasterxml.jackson.databind.ObjectMapper; + +import lombok.extern.slf4j.Slf4j; + +/** + * Netty HTTP handler for the A2A Gateway REST + SSE API. + * + *

Endpoints (issue #5302 D1 scope): + * + * 

+ *   POST   /a2a/tasks              - submit task (sync/async)
+ *   GET    /a2a/tasks               - list tasks (?state=COMPLETED&limit=100&offset=0)
+ *   GET    /a2a/tasks/{taskId}      - get task status
+ *   DELETE /a2a/tasks/{taskId}      - cancel task
+ *   GET    /a2a/tasks/{taskId}/wait - wait for result (long-poll)
+ *   GET    /a2a/tasks/{taskId}/stream - SSE stream of task status updates
+ *   GET    /a2a/health              - health check
+ *
+ * + *

Note: the SSE stream and the long-poll endpoint resolve the same future; the SSE + * stream additionally registers a {@link A2AGatewayService.StatusSubscriber} so intermediate + * status updates (SUBMITTED -> WORKING -> COMPLETED) are pushed to the client before the + * final result.

+ */ +@Slf4j +public class A2AGatewayHttpHandler extends SimpleChannelInboundHandler { + + private static final ObjectMapper objectMapper = new ObjectMapper(); + + private final A2AGatewayService gatewayService; + + public A2AGatewayHttpHandler(A2AGatewayService gatewayService) { + this.gatewayService = gatewayService; + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception { + String uri = req.uri(); + try { + if (uri.startsWith("/a2a/tasks/") && uri.endsWith("/stream")) { + handleSse(ctx, req); + } else if (uri.startsWith("/a2a/tasks/") && uri.endsWith("/wait")) { + handleLongPoll(ctx, req); + } else if (uri.startsWith("/a2a/tasks/") && !uri.contains("?")) { + String taskId = uri.substring("/a2a/tasks/".length()); + if ("DELETE".equalsIgnoreCase(req.method().name())) { + handleCancel(ctx, taskId); + } else { + handleGet(ctx, taskId); + } + } else if (uri.equals("/a2a/tasks") || uri.startsWith("/a2a/tasks?")) { + if ("POST".equalsIgnoreCase(req.method().name())) { + handleSubmit(ctx, req); + } else { + handleList(ctx, req); + } + } else if (uri.equals("/a2a/health") || uri.startsWith("/a2a/health?")) { + handleHealth(ctx); + } else { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, "{\"error\":\"not_found\"}"); + } + } catch (Exception e) { + log.error("Error handling {} {}", req.method(), uri, e); + writeJson(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, + "{\"error\":\"internal\",\"message\":\"" + e.getMessage() + "\"}"); + } + } + + // ========================================================================= + // Endpoint handlers + // ========================================================================= + + private void handleSubmit(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception { + String body = req.content().toString(StandardCharsets.UTF_8); + @SuppressWarnings("unchecked") + Map payload = objectMapper.readValue(body, Map.class); + String targetAgent = (String) payload.get("targetAgent"); + String message = (String) payload.get("message"); + String parentTaskId = (String) payload.get("parentTaskId"); + Boolean sync = (Boolean) payload.getOrDefault("sync", Boolean.FALSE); + + if (targetAgent == null || message == null) { + writeJson(ctx, HttpResponseStatus.BAD_REQUEST, + "{\"error\":\"bad_request\",\"message\":\"targetAgent and message are required\"}"); + return; + } + + A2AGatewayService.TaskResult result; + try { + if (Boolean.TRUE.equals(sync)) { + result = gatewayService.submitTask(targetAgent, message, parentTaskId) + .get(10, TimeUnit.SECONDS); + } else { + // Async: return 202 with taskId + String taskId = "task-async-" + System.nanoTime(); + gatewayService.submitTask(taskId, targetAgent, message, parentTaskId); + writeJson(ctx, HttpResponseStatus.ACCEPTED, + "{\"taskId\":\"" + taskId + "\",\"state\":\"SUBMITTED\"}"); + return; + } + } catch (java.util.concurrent.TimeoutException e) { + writeJson(ctx, HttpResponseStatus.GATEWAY_TIMEOUT, + "{\"error\":\"timeout\",\"message\":\"" + e.getMessage() + "\"}"); + return; + } catch (Exception e) { + writeJson(ctx, HttpResponseStatus.BAD_REQUEST, + "{\"error\":\"bad_request\",\"message\":\"" + e.getMessage() + "\"}"); + return; + } + + writeJson(ctx, HttpResponseStatus.OK, toJson(result)); + } + + private void handleGet(ChannelHandlerContext ctx, String taskId) { + A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); + if (snap == null) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); + return; + } + writeJson(ctx, HttpResponseStatus.OK, snapshotJson(snap)); + } + + private void handleCancel(ChannelHandlerContext ctx, String taskId) { + boolean ok = gatewayService.cancelTask(taskId); + if (!ok) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_cancellable\",\"taskId\":\"" + taskId + "\"}"); + return; + } + writeJson(ctx, HttpResponseStatus.OK, "{\"taskId\":\"" + taskId + "\",\"state\":\"CANCELLED\"}"); + } + + private void handleList(ChannelHandlerContext ctx, FullHttpRequest req) { + QueryStringDecoder qsd = new QueryStringDecoder(req.uri()); + String stateFilter = null; + int limit = 100; + int offset = 0; + for (Map.Entry> e : qsd.parameters().entrySet()) { + if ("state".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { + stateFilter = e.getValue().get(0); + } else if ("limit".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { + try { limit = Integer.parseInt(e.getValue().get(0)); } catch (NumberFormatException ignored) { } + } else if ("offset".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { + try { offset = Integer.parseInt(e.getValue().get(0)); } catch (NumberFormatException ignored) { } + } + } + + // Build a synthetic list. The persistent store indexes by agent; the gateway filters + // its known tasks by state. D2 (issue #5302) will introduce a global index. + StringBuilder sb = new StringBuilder("{\"tasks\":["); + int count = 0; + int skipped = 0; + for (var entry : gatewayService.getTaskStore().listByAgent( + gatewayService.getGatewayId(), null).stream() + .sorted((a, b) -> Long.compare(b.createdAtMs, a.createdAtMs)) + .toList()) { + if (stateFilter != null && !stateFilter.equalsIgnoreCase( + A2AGatewayService.toLegacyState(entry.status).name())) { + continue; + } + if (skipped < offset) { skipped++; continue; } + if (count >= limit) { break; } + if (count > 0) sb.append(','); + sb.append("{\"taskId\":\"").append(entry.taskId).append("\",") + .append("\"state\":\"").append(A2AGatewayService.toLegacyState(entry.status)).append("\",") + .append("\"createdAt\":").append(entry.createdAtMs).append('}'); + count++; + } + sb.append("],\"totalListed\":").append(count).append('}'); + + writeJson(ctx, HttpResponseStatus.OK, sb.toString()); + } + + private void handleLongPoll(ChannelHandlerContext ctx, FullHttpRequest req) { + String uri = req.uri(); + String taskId = uri.substring("/a2a/tasks/".length(), uri.length() - "/wait".length()); + A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); + if (snap == null) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); + return; + } + if (snap.getRecord().status.ordinal() >= 2) { // COMPLETED, FAILED, CANCELED + writeJson(ctx, HttpResponseStatus.OK, snapshotJson(snap)); + return; + } + // For long-poll, the SSE endpoint is the more general solution; redirect to it + writeJson(ctx, HttpResponseStatus.SEE_OTHER, + "{\"hint\":\"use /a2a/tasks/" + taskId + "/stream for live updates\"}"); + } + + private void handleSse(ChannelHandlerContext ctx, FullHttpRequest req) { + String uri = req.uri(); + String taskId = uri.substring("/a2a/tasks/".length(), uri.length() - "/stream".length()); + A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); + if (snap == null) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); + return; + } + + DefaultFullHttpResponse headers = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, HttpResponseStatus.OK, + Unpooled.buffer(0)); + headers.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream"); + headers.headers().set(HttpHeaderNames.CACHE_CONTROL, "no-cache"); + headers.headers().set(HttpHeaderNames.CONNECTION, "keep-alive"); + ctx.writeAndFlush(headers); + + // Send current state immediately + ctx.writeAndFlush(Unpooled.wrappedBuffer( + ("data: " + snapshotJson(snap) + "\n\n").getBytes(StandardCharsets.UTF_8))); + + // If terminal, close immediately + if (snap.getRecord().status.ordinal() >= 2) { + ctx.writeAndFlush(Unpooled.wrappedBuffer( + "event: end\ndata: {}\n\n".getBytes(StandardCharsets.UTF_8))) + .addListener(ChannelFutureListener.CLOSE); + return; + } + + // Register subscriber for live updates + gatewayService.registerStatusSubscriber(taskId, (id, state, data) -> { + String line = "data: {\"taskId\":\"" + id + "\",\"state\":\"" + state + "\""; + if (data != null) { + line += ",\"data\":" + objectMapper.valueToTree(data).toString(); + } + line += "}\n\n"; + ChannelFuture f = ctx.writeAndFlush(Unpooled.wrappedBuffer( + line.getBytes(StandardCharsets.UTF_8))); + if ("completed".equals(state) || "failed".equals(state) || "cancelled".equals(state)) { + f.addListener(ChannelFutureListener.CLOSE); + } + }); + } + + private void handleHealth(ChannelHandlerContext ctx) { + writeJson(ctx, HttpResponseStatus.OK, + "{\"status\":\"ok\",\"gatewayId\":\"" + gatewayService.getGatewayId() + "\"}"); + } + + // ========================================================================= + // Helpers + // ========================================================================= + + private void writeJson(ChannelHandlerContext ctx, HttpResponseStatus status, String json) { + FullHttpResponse resp = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, status, + Unpooled.wrappedBuffer(json.getBytes(StandardCharsets.UTF_8))); + resp.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json"); + resp.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, "*"); + resp.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, json.length()); + ctx.writeAndFlush(resp); + } + + private String snapshotJson(A2AGatewayService.TaskSnapshot snap) { + return "{" + + "\"taskId\":\"" + snap.getRecord().taskId + "\"," + + "\"state\":\"" + snap.getState() + "\"," + + "\"targetAgent\":\"" + snap.getRecord().agentId + "\"," + + "\"createdAt\":" + snap.getRecord().createdAtMs + "," + + "\"updatedAt\":" + snap.getRecord().updatedAtMs + + (snap.getRecord().output != null + ? ",\"output\":" + objectMapper.valueToTree(snap.getRecord().output).toString() + : "") + + (snap.getParentTaskId() != null + ? ",\"parentTaskId\":\"" + snap.getParentTaskId() + "\"" + : "") + + "}"; + } + + private String toJson(A2AGatewayService.TaskResult r) { + return "{\"state\":\"" + r.getState() + "\"," + + (r.getData() != null + ? "\"data\":" + objectMapper.valueToTree(r.getData()).toString() + "," + : "") + + (r.getErrorMessage() != null + ? "\"errorMessage\":\"" + r.getErrorMessage() + "\"" + : "") + + "}"; + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + log.error("Channel exception", cause); + ctx.close(); + } +} diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServer.java new file mode 100644 index 0000000000..58169db157 --- /dev/null +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServer.java @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import org.apache.eventmesh.protocol.a2a.A2AMessageTransport; +import org.apache.eventmesh.runtime.state.TaskStore; + +import java.util.concurrent.TimeUnit; + +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpRequestDecoder; +import io.netty.handler.codec.http.HttpResponseEncoder; + +import lombok.extern.slf4j.Slf4j; + +/** + * A2A Gateway HTTP server. + * + *

Boots an embedded Netty HTTP server that exposes the A2A Gateway REST + SSE API on + * the given port. The gateway is constructed with a {@link TaskStore} (Sub-PR A/C), an + * {@link AgentCardRegistry} (in-memory D1; Meta-backed D2), and a + * {@link A2AMessageTransport} — production wiring is + * {@link EventMeshA2ATransport} (Runtime-bridged), tests pass an in-process transport.

+ * + *

Issue #5302 D1 scope: this class is the new home of the gateway that + * PR #5260 originally added + * (then deleted in the uni-architecture redesign). The weather-agent demo from PR #5260 + * has been removed — the gateway no longer pre-registers a mock agent; the demo + * client ({@code A2AGatewayDemo}) will be ported in D2 alongside AgentCard Meta-ization.

+ */ +@Slf4j +public class A2AGatewayServer { + + private final int port; + private final A2AMessageTransport transport; + private final TaskStore taskStore; + private final AgentCardRegistry agentCardRegistry; + + private EventLoopGroup bossGroup; + private EventLoopGroup workerGroup; + private Channel serverChannel; + + private A2AGatewayService gatewayService; + private A2AGatewayHttpHandler gatewayHandler; + + public A2AGatewayServer(int port, A2AMessageTransport transport, TaskStore taskStore, + AgentCardRegistry agentCardRegistry) { + this.port = port; + this.transport = transport; + this.taskStore = taskStore; + this.agentCardRegistry = agentCardRegistry; + } + + public void start() throws Exception { + // 1. Initialize components + gatewayService = new A2AGatewayService( + "global", "gateway-" + port, transport, taskStore, agentCardRegistry); + gatewayService.start(); + + gatewayHandler = new A2AGatewayHttpHandler(gatewayService); + + // 2. Start Netty HTTP server + bossGroup = new NioEventLoopGroup(1); + workerGroup = new NioEventLoopGroup(); + + ServerBootstrap bootstrap = new ServerBootstrap(); + bootstrap.group(bossGroup, workerGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel ch) { + ch.pipeline() + .addLast(new HttpRequestDecoder()) + .addLast(new HttpObjectAggregator(65536)) + .addLast(new HttpResponseEncoder()) + .addLast(gatewayHandler); + } + }) + .option(ChannelOption.SO_BACKLOG, 128) + .childOption(ChannelOption.SO_KEEPALIVE, true); + + serverChannel = bootstrap.bind(port).sync().channel(); + log.info("=== A2A Gateway Server started on port {} ===", port); + log.info("Gateway ID: {}, backed by: {}", + gatewayService.getGatewayId(), taskStore.getClass().getSimpleName()); + } + + public void shutdown() throws Exception { + if (serverChannel != null) { + serverChannel.close().sync(); + } + if (bossGroup != null) { + bossGroup.shutdownGracefully(0, 1, TimeUnit.SECONDS); + } + if (workerGroup != null) { + workerGroup.shutdownGracefully(0, 1, TimeUnit.SECONDS); + } + if (gatewayService != null) { + gatewayService.shutdown(); + } + } + + public int getPort() { + return port; + } + + public A2AGatewayService getGatewayService() { + return gatewayService; + } +} diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayService.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayService.java new file mode 100644 index 0000000000..ccc3c3642d --- /dev/null +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayService.java @@ -0,0 +1,538 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import org.apache.eventmesh.protocol.a2a.A2AMessageTransport; +import org.apache.eventmesh.protocol.a2a.A2AProtocolConstants; +import org.apache.eventmesh.protocol.a2a.A2ATopicFactory; +import org.apache.eventmesh.runtime.state.TaskStore; +import org.apache.eventmesh.runtime.state.TaskStore.Status; +import org.apache.eventmesh.runtime.state.TaskStore.TaskRecord; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; + +import lombok.extern.slf4j.Slf4j; + +/** + * A2A Gateway Service: orchestrates task submission, response handling, and SSE streaming. + * + *

This is the core component that ties together: + *

    + *
  • {@link A2AMessageTransport} for pub/sub (in production: the Runtime-bridged + * {@link EventMeshA2ATransport}, NOT a parallel in-memory transport)
    • + *
  • {@link TaskStore} for task lifecycle (issue #5301 Sub-PR A/C; durable across restarts)
    • + *
  • {@link AgentCardRegistry} for agent discovery (in-memory D1; Meta-backed in D2)
    • + *
  • {@link A2ATopicFactory} for topic routing
    • + *
+ * + *

Design (issue #5302): the in-memory {@code TaskRegistry} that + * PR #5260 introduced has been + * replaced by {@link TaskStore} — the gateway no longer owns a private task table. A + * small runtime cache ({@code parentTaskIdCache}, {@code taskEpochCache}) holds fields the + * persistent store does not model (parent links, the per-task epoch used for stale-write + * rejection). Both caches are rebuildable from the store on a fresh JVM.

+ * + *

Status mapping (PR #5260 -> Sub-PR A/C): + * {@code SUBMITTED -> PENDING}, {@code WORKING -> RUNNING}, + * {@code CANCELLED -> CANCELED} (one L). The public state names exposed by + * {@link A2AGatewayService.TaskState} are the legacy names (kebab-cased protocol JSON); the + * internal store uses {@link Status}.

+ */ +@Slf4j +public class A2AGatewayService { + + private final String namespace; + private final String gatewayId; + private final A2AMessageTransport transport; + private final TaskStore taskStore; + private final AgentCardRegistry agentCardRegistry; + + // Pending tasks waiting for response (runtime-only; not persisted) + private final ConcurrentHashMap> pendingTasks = new ConcurrentHashMap<>(); + // SSE subscribers for status updates (runtime-only; rebuilt on stream start) + private final ConcurrentHashMap> statusSubscribers = new ConcurrentHashMap<>(); + // Parent task id cache (TaskStore does not model parent links; rebuilt on recovery) + private final ConcurrentHashMap parentTaskIdCache = new ConcurrentHashMap<>(); + // Per-task epoch cache (TaskStore.updateStatus requires the epoch set at createTask; we + // remember it so transition methods don't have to re-read the record) + private final ConcurrentHashMap taskEpochCache = new ConcurrentHashMap<>(); + + private volatile boolean started = false; + private String responseSubscriptionId; + private String statusSubscriptionId; + + // Task timeout: tasks that don't receive a response within this duration are auto-failed + private static final long DEFAULT_TASK_TIMEOUT_MS = 120_000L; // 2 minutes + private final long taskTimeoutMs; + private ScheduledExecutorService taskTimeoutScheduler; + + public A2AGatewayService(String namespace, String gatewayId, + A2AMessageTransport transport, + TaskStore taskStore, + AgentCardRegistry agentCardRegistry) { + this(namespace, gatewayId, transport, taskStore, agentCardRegistry, DEFAULT_TASK_TIMEOUT_MS); + } + + public A2AGatewayService(String namespace, String gatewayId, + A2AMessageTransport transport, + TaskStore taskStore, + AgentCardRegistry agentCardRegistry, + long taskTimeoutMs) { + this.namespace = namespace; + this.gatewayId = gatewayId; + this.transport = transport; + this.taskStore = taskStore; + this.agentCardRegistry = agentCardRegistry; + this.taskTimeoutMs = taskTimeoutMs; + } + + public String getGatewayId() { + return gatewayId; + } + + public String getNamespace() { + return namespace; + } + + public AgentCardRegistry getAgentCardRegistry() { + return agentCardRegistry; + } + + public TaskStore getTaskStore() { + return taskStore; + } + + /** + * Starts the gateway service: subscribes to gateway response/status topics. + */ + public synchronized void start() throws Exception { + if (started) { + return; + } + + // Start task timeout scheduler + taskTimeoutScheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "a2a-task-timeout"); + t.setDaemon(true); + return t; + }); + + // Subscribe to all responses for this gateway + String responseTopic = A2ATopicFactory.gatewayResponseWildcardTopic(namespace, gatewayId); + responseSubscriptionId = transport.subscribe(responseTopic, this::handleResponse); + + // Subscribe to all status updates for this gateway + String statusTopic = A2ATopicFactory.gatewayStatusWildcardTopic(namespace, gatewayId); + statusSubscriptionId = transport.subscribe(statusTopic, this::handleStatus); + + started = true; + log.info("A2AGatewayService started: gatewayId={}, namespace={}", gatewayId, namespace); + } + + public synchronized void shutdown() throws Exception { + if (!started) { + return; + } + if (responseSubscriptionId != null) { + transport.unsubscribe(responseSubscriptionId); + } + if (statusSubscriptionId != null) { + transport.unsubscribe(statusSubscriptionId); + } + if (taskTimeoutScheduler != null) { + taskTimeoutScheduler.shutdownNow(); + taskTimeoutScheduler = null; + } + pendingTasks.clear(); + statusSubscribers.clear(); + parentTaskIdCache.clear(); + taskEpochCache.clear(); + started = false; + log.info("A2AGatewayService shutdown."); + } + + // ========================================================================= + // Task Submission + // ========================================================================= + + /** + * Submits an A2A task to a target agent with an auto-generated task id. + */ + public CompletableFuture submitTask(String targetAgent, String message, String parentTaskId) { + String taskId = generateTaskId(); + return submitTask(taskId, targetAgent, message, parentTaskId); + } + + /** + * Submits an A2A task with a specific task id. The target agent must be registered in + * the {@link AgentCardRegistry}. + */ + public CompletableFuture submitTask(String taskId, String targetAgent, + String message, String parentTaskId) { + if (!started) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("Gateway not started")); + return future; + } + + // Validate that the target agent is registered + if (!agentCardRegistry.isAgentRegistered(targetAgent)) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalArgumentException( + "Target agent not registered: " + targetAgent)); + return future; + } + + // Create task in the persistent store. createTask returns null on duplicate taskId. + TaskRecord rec = taskStore.createTask(taskId, targetAgent, gatewayId, message); + if (rec == null) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(new IllegalStateException("Duplicate taskId: " + taskId)); + return future; + } + if (parentTaskId != null) { + parentTaskIdCache.put(taskId, parentTaskId); + } + taskEpochCache.put(taskId, rec.taskEpoch); + + // Build A2A CloudEvent + CloudEvent event = buildTaskRequestEvent(taskId, targetAgent, message, parentTaskId); + + // Register pending future BEFORE publishing: a synchronous transport callback (e.g. a + // local in-memory test) could deliver the response before submitTask returns, and we + // would lose the future if put() ran after handleResponse(). + CompletableFuture future = new CompletableFuture<>(); + pendingTasks.put(taskId, future); + + // Schedule timeout: if no response within taskTimeoutMs, auto-fail the task + if (taskTimeoutScheduler != null) { + taskTimeoutScheduler.schedule(() -> { + CompletableFuture pending = pendingTasks.get(taskId); + if (pending != null && !pending.isDone()) { + String errMsg = "Task timed out after " + taskTimeoutMs + "ms with no response"; + Long epoch = taskEpochCache.get(taskId); + if (epoch != null) { + taskStore.updateStatus(taskId, epoch, Status.FAILED, errMsg); + } + pendingTasks.remove(taskId); + pending.completeExceptionally(new java.util.concurrent.TimeoutException(errMsg)); + notifyStatusSubscribers(taskId, "failed", errMsg); + log.warn("Task timed out: taskId={}, targetAgent={}", taskId, targetAgent); + } + }, taskTimeoutMs, TimeUnit.MILLISECONDS); + } + + // Publish to agent request topic + String requestTopic = A2ATopicFactory.agentRequestTopic(namespace, targetAgent); + try { + transport.publish(requestTopic, event); + log.info("Task submitted: taskId={}, targetAgent={}, topic={}", taskId, targetAgent, requestTopic); + } catch (Exception e) { + log.error("Failed to publish task: taskId={}", taskId, e); + Long epoch = taskEpochCache.remove(taskId); + if (epoch != null) { + taskStore.updateStatus(taskId, epoch, Status.FAILED, "Publish failed: " + e.getMessage()); + } + pendingTasks.remove(taskId); + future.completeExceptionally(e); + } + + return future; + } + + /** + * Cancels a task. Idempotent: cancelling a non-PENDING/RUNNING task is a no-op. + */ + public boolean cancelTask(String taskId) { + Long epoch = taskEpochCache.get(taskId); + if (epoch == null) { + return false; + } + TaskRecord rec = taskStore.getTask(taskId); + if (rec == null) { + return false; + } + if (rec.status == Status.COMPLETED || rec.status == Status.FAILED || rec.status == Status.CANCELED) { + return false; + } + boolean ok = taskStore.updateStatus(taskId, epoch, Status.CANCELED, null); + if (ok) { + CompletableFuture future = pendingTasks.remove(taskId); + if (future != null) { + future.complete(new TaskResult(TaskState.CANCELLED, null, "Task cancelled")); + } + notifyStatusSubscribers(taskId, "cancelled", "Task cancelled"); + log.info("Task cancelled: taskId={}", taskId); + } + return ok; + } + + /** + * Gets task status (a snapshot from the persistent store). + */ + public TaskSnapshot getTaskStatus(String taskId) { + TaskRecord rec = taskStore.getTask(taskId); + if (rec == null) { + return null; + } + String parentId = parentTaskIdCache.get(taskId); + return new TaskSnapshot(rec, parentId); + } + + /** + * Lists child task ids of a parent task, scanning the runtime cache. The persistent + * {@link TaskStore} does not model parent-child relations; this list reflects the + * gateway-local view and is rebuilt on a fresh JVM by replaying pendingTasks / scanning + * parentTaskIdCache (see issue #5302 D2 — the Meta-ized TaskStore should grow a + * parent index in a follow-up). + */ + public List getChildTasks(String parentTaskId) { + List children = new java.util.ArrayList<>(); + for (var entry : parentTaskIdCache.entrySet()) { + if (parentTaskId.equals(entry.getValue())) { + children.add(entry.getKey()); + } + } + return children; + } + + // ========================================================================= + // SSE Status Subscription + // ========================================================================= + + /** + * Registers a subscriber for status updates on a specific task. The subscriber is invoked + * on whichever thread completes the task transition (transport callback, timeout scheduler, + * or a cancel). + */ + public void registerStatusSubscriber(String taskId, StatusSubscriber subscriber) { + statusSubscribers.computeIfAbsent(taskId, k -> new CopyOnWriteArrayList<>()).add(subscriber); + } + + /** + * Removes a status subscriber. If the subscriber list becomes empty, the task entry is + * removed entirely so the map does not grow unbounded. + */ + public void unregisterStatusSubscriber(String taskId, StatusSubscriber subscriber) { + List subs = statusSubscribers.get(taskId); + if (subs != null) { + subs.remove(subscriber); + if (subs.isEmpty()) { + statusSubscribers.remove(taskId); + } + } + } + + // ========================================================================= + // Response / Status Handling + // ========================================================================= + + private void handleResponse(String topic, CloudEvent event) { + A2ATopicFactory.ParsedTopic parsed = A2ATopicFactory.parse(topic); + if (parsed == null || !parsed.isResponse() || parsed.getTaskId() == null) { + return; + } + String taskId = parsed.getTaskId(); + log.info("Received response for task: {}", taskId); + + TaskRecord rec = taskStore.getTask(taskId); + if (rec == null) { + log.warn("Response received for unknown task: {}", taskId); + return; + } + + String resultData = extractEventData(event); + taskStore.updateStatus(taskId, rec.taskEpoch, Status.COMPLETED, resultData); + + CompletableFuture future = pendingTasks.remove(taskId); + if (future != null) { + future.complete(new TaskResult(TaskState.COMPLETED, resultData, null)); + } + + // Notify SSE subscribers + notifyStatusSubscribers(taskId, "completed", resultData); + } + + private void handleStatus(String topic, CloudEvent event) { + A2ATopicFactory.ParsedTopic parsed = A2ATopicFactory.parse(topic); + if (parsed == null || !parsed.isStatus() || parsed.getTaskId() == null) { + return; + } + String taskId = parsed.getTaskId(); + String statusData = extractEventData(event); + log.debug("Received status for task: {} -> {}", taskId, statusData); + + // Mark as RUNNING if currently PENDING. Idempotent: if the task is already terminal + // or RUNNING, updateStatus with the same epoch is a no-op for status but updates + // updatedAtMs — which we want for the timeout clock. + TaskRecord rec = taskStore.getTask(taskId); + if (rec != null && rec.status == Status.PENDING) { + taskStore.updateStatus(taskId, rec.taskEpoch, Status.RUNNING, null); + } + + // Notify SSE subscribers + notifyStatusSubscribers(taskId, "working", statusData); + } + + private void notifyStatusSubscribers(String taskId, String state, String data) { + List subs = statusSubscribers.get(taskId); + if (subs != null) { + for (StatusSubscriber sub : subs) { + try { + sub.onStatus(taskId, state, data); + } catch (Exception e) { + log.warn("Status subscriber error for task {}: {}", taskId, e.getMessage()); + } + } + } + } + + // ========================================================================= + // CloudEvent Building + // ========================================================================= + + private CloudEvent buildTaskRequestEvent(String taskId, String targetAgent, + String message, String parentTaskId) { + CloudEventBuilder builder = CloudEventBuilder.v1() + .withId(taskId) + .withType(A2AProtocolConstants.CE_TYPE_PREFIX + "task.request") + .withSource(java.net.URI.create("gateway/" + gatewayId)) + .withDataContentType("application/json") + .withData(message.getBytes(StandardCharsets.UTF_8)) + .withExtension(A2AProtocolConstants.CE_EXTENSION_A2A_METHOD, A2AProtocolConstants.OP_SEND_MESSAGE) + .withExtension(A2AProtocolConstants.CE_EXTENSION_TARGET_AGENT, targetAgent) + .withExtension(A2AProtocolConstants.CE_EXTENSION_PROTOCOL, "A2A") + .withExtension(A2AProtocolConstants.CE_EXTENSION_PROTOCOL_VERSION, A2AProtocolConstants.PROTOCOL_VERSION); + + if (parentTaskId != null) { + builder.withExtension(A2AProtocolConstants.CE_EXTENSION_COLLABORATION_ID, parentTaskId); + } + + return builder.build(); + } + + private String extractEventData(CloudEvent event) { + if (event.getData() == null) { + return null; + } + return new String(event.getData().toBytes(), StandardCharsets.UTF_8); + } + + private String generateTaskId() { + return "task-" + UUID.randomUUID().toString().substring(0, 8); + } + + // ========================================================================= + // Result Types + // ========================================================================= + + /** + * Legacy state names exposed to the A2A wire protocol and JSON response payloads. The + * persistent store uses {@link Status}; this enum is the on-the-wire vocabulary. + */ + public enum TaskState { + SUBMITTED, + WORKING, + COMPLETED, + FAILED, + CANCELLED + } + + public static class TaskResult { + + private final TaskState state; + private final String data; + private final String errorMessage; + + public TaskResult(TaskState state, String data, String errorMessage) { + this.state = state; + this.data = data; + this.errorMessage = errorMessage; + } + + public TaskState getState() { + return state; + } + + public String getData() { + return data; + } + + public String getErrorMessage() { + return errorMessage; + } + } + + /** + * A snapshot of a task — the persisted record plus the runtime-only parent link. + */ + public static class TaskSnapshot { + private final TaskRecord record; + private final String parentTaskId; + + public TaskSnapshot(TaskRecord record, String parentTaskId) { + this.record = record; + this.parentTaskId = parentTaskId; + } + + public TaskRecord getRecord() { + return record; + } + + public String getParentTaskId() { + return parentTaskId; + } + + public TaskState getState() { + return toLegacyState(record.status); + } + } + + /** + * Maps the persistent {@link Status} to the legacy A2A wire vocabulary. + */ + public static TaskState toLegacyState(Status s) { + switch (s) { + case PENDING: return TaskState.SUBMITTED; + case RUNNING: return TaskState.WORKING; + case COMPLETED: return TaskState.COMPLETED; + case FAILED: return TaskState.FAILED; + case CANCELED: return TaskState.CANCELLED; + default: throw new IllegalStateException("Unknown status: " + s); + } + } + + /** + * Callback interface for task status change notifications. + */ + @FunctionalInterface + public interface StatusSubscriber { + void onStatus(String taskId, String state, String data); + } +} diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/AgentCardRegistry.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/AgentCardRegistry.java new file mode 100644 index 0000000000..2d8b157257 --- /dev/null +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/AgentCardRegistry.java @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import org.apache.eventmesh.protocol.a2a.AgentIdentity; +import org.apache.eventmesh.protocol.a2a.model.AgentCard; + +/** + * Agent-card registry. Holds the discovery table that A2A clients consult before submitting + * a task (so an unregistered agent cannot receive a message). + * + *

Issue #5302 D1 scope: only the in-memory implementation + * ({@link InMemoryAgentCardRegistry}) is provided. A Meta-backed implementation is the + * subject of Sub-PR D2 — it will reuse {@code org.apache.eventmesh.runtime.state.SessionStore} + * (Sub-PR A) for cluster-shared agent registration with prefix-watch invalidation.

+ */ +public interface AgentCardRegistry { + + /** + * Registers a card under the given identity. If a card is already registered for the + * same identity, the existing entry is replaced. + */ + void registerCard(AgentIdentity id, AgentCard card); + + /** + * Removes a card by identity. Returns {@code true} if a card was removed. + */ + boolean removeCard(AgentIdentity id); + + /** + * @return {@code true} if a card is registered for the given agent name (any identity + * tuple whose {@code agentId} field matches). + */ + boolean isAgentRegistered(String agentName); + + /** + * Looks up a card by agent name. + */ + AgentCard getCard(String agentName); +} diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/InMemoryAgentCardRegistry.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/InMemoryAgentCardRegistry.java new file mode 100644 index 0000000000..323902ede9 --- /dev/null +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/InMemoryAgentCardRegistry.java @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import org.apache.eventmesh.protocol.a2a.AgentIdentity; +import org.apache.eventmesh.protocol.a2a.model.AgentCard; + +import java.util.concurrent.ConcurrentHashMap; + +import lombok.extern.slf4j.Slf4j; + +/** + * In-memory {@link AgentCardRegistry}. Suitable for tests and single-process demos; not + * cluster-safe (a fresh JVM starts with an empty registry). + * + *

Sub-PR D2 will add a Meta-backed implementation that survives restarts and is shared + * across Runtime instances.

+ */ +@Slf4j +public class InMemoryAgentCardRegistry implements AgentCardRegistry { + + private final ConcurrentHashMap cardsByAgentId = new ConcurrentHashMap<>(); + + @Override + public void registerCard(AgentIdentity id, AgentCard card) { + cardsByAgentId.put(id.getAgentId(), card); + log.info("Registered agent card: agentId={}", id.getAgentId()); + } + + @Override + public boolean removeCard(AgentIdentity id) { + return cardsByAgentId.remove(id.getAgentId()) != null; + } + + @Override + public boolean isAgentRegistered(String agentName) { + return cardsByAgentId.containsKey(agentName); + } + + @Override + public AgentCard getCard(String agentName) { + return cardsByAgentId.get(agentName); + } +} diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java new file mode 100644 index 0000000000..1e3b0a024c --- /dev/null +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java @@ -0,0 +1,326 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.eventmesh.protocol.a2a.AgentIdentity; +import org.apache.eventmesh.protocol.a2a.model.AgentCard; +import org.apache.eventmesh.protocol.a2a.model.AgentCapabilities; +import org.apache.eventmesh.protocol.a2a.model.AgentInterface; +import org.apache.eventmesh.protocol.a2a.model.AgentSkill; +import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskResult; +import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskSnapshot; +import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskState; +import org.apache.eventmesh.runtime.state.TaskStore; +import org.apache.eventmesh.runtime.state.TaskStore.Status; +import org.apache.eventmesh.runtime.state.TaskStore.TaskRecord; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.cloudevents.CloudEvent; + +/** + * Sub-PR D1: A2AGatewayService is wired to {@link TaskStore} (Sub-PR A/C) instead of + * the in-memory {@code TaskRegistry} from PR #5260. These tests exercise the + * gateway's task lifecycle against an in-process TaskStore that mirrors the + * baseline contract test in {@code TaskStoreTest}. + */ +class A2AGatewayServiceTest { + + /** In-process TaskStore mirroring the Sub-PR A test stub. */ + static final class InProcessTaskStore implements TaskStore { + private final ConcurrentHashMap table = new ConcurrentHashMap<>(); + private final AtomicLong epoch = new AtomicLong(); + + @Override + public TaskRecord createTask(String taskId, String agentId, String clientId, String input) { + long now = System.currentTimeMillis(); + long e = epoch.incrementAndGet(); + TaskRecord rec = new TaskRecord(taskId, agentId, clientId, Status.PENDING, now, now, input, null, e); + return table.putIfAbsent(taskId, rec) == null ? rec : null; + } + + @Override + public TaskRecord getTask(String taskId) { + return table.get(taskId); + } + + @Override + public boolean updateStatus(String taskId, long expectedTaskEpoch, Status newStatus, String output) { + TaskRecord rec = table.get(taskId); + if (rec == null || rec.taskEpoch != expectedTaskEpoch) { + return false; + } + rec.status = newStatus; + rec.updatedAtMs = System.currentTimeMillis(); + rec.output = output; + return true; + } + + @Override + public List listByAgent(String agentId, Status statusFilter) { + List out = new ArrayList<>(); + for (TaskRecord r : table.values()) { + if (!r.agentId.equals(agentId)) continue; + if (statusFilter != null && r.status != statusFilter) continue; + out.add(r); + } + return out; + } + + @Override + public List expireStale(long olderThanMs) { + long deadline = System.currentTimeMillis() - olderThanMs; + List expired = new ArrayList<>(); + for (TaskRecord r : table.values()) { + if (r.updatedAtMs < deadline) { + expired.add(r.taskId); + } + } + for (String id : expired) { + table.remove(id); + } + return expired; + } + + @Override + public void flush() { } + + @Override + public void close() { + table.clear(); + } + } + + /** In-process pub/sub transport: publish delivers synchronously to subscribers on the same topic. */ + static final class InProcessTransport implements org.apache.eventmesh.protocol.a2a.A2AMessageTransport { + final ConcurrentHashMap subs = + new ConcurrentHashMap<>(); + + @Override + public void publish(String topic, CloudEvent event) { + for (var entry : subs.entrySet()) { + if (matches(topic, entry.getKey())) { + entry.getValue().onMessage(topic, event); + } + } + } + + @Override + public String subscribe(String topicPattern, + org.apache.eventmesh.protocol.a2a.A2AMessageTransport.MessageCallback callback) { + subs.put(topicPattern, callback); + return "sub-" + topicPattern; + } + + @Override + public void unsubscribe(String subscriptionId) { + subs.remove(subscriptionId.replace("sub-", "")); + } + + private boolean matches(String topic, String pattern) { + if (pattern.equals(topic)) return true; + // A2A topics use Pulsar/MQTT-style wildcards: + matches one path segment, * + // matches one or more. Convert to a regex for in-process delivery. + String regex = pattern + .replace("+", "[^/]+") + .replace("*", "[^/]+"); + return topic.matches(regex); + } + } + + private A2AGatewayService gateway; + private InMemoryAgentCardRegistry registry; + private InProcessTransport transport; + private InProcessTaskStore store; + private String testAgentName; + + @BeforeEach + void setUp() throws Exception { + transport = new InProcessTransport(); + store = new InProcessTaskStore(); + registry = new InMemoryAgentCardRegistry(); + + testAgentName = "echo-agent-" + System.nanoTime(); + registerEchoAgent(testAgentName); + + gateway = new A2AGatewayService("global", "test-gateway", transport, store, registry, 5000L); + gateway.start(); + } + + @AfterEach + void tearDown() throws Exception { + if (gateway != null) gateway.shutdown(); + if (store != null) store.close(); + } + + private void registerEchoAgent(String name) throws Exception { + AgentCard card = AgentCard.builder() + .name(name) + .description("Echoes the input back as a response") + .version("1.0.0") + .supportedInterfaces(Arrays.asList(AgentInterface.builder() + .url("http://localhost:0/a2a") + .protocolBinding("JSONRPC") + .protocolVersion("0.3") + .build())) + .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) + .skills(Arrays.asList(AgentSkill.builder() + .id("echo").name("Echo").description("Echoes input") + .tags(Arrays.asList("test", "echo")).build())) + .defaultInputModes(Arrays.asList("text/plain")) + .defaultOutputModes(Arrays.asList("text/plain")) + .build(); + AgentIdentity id = AgentIdentity.builder().orgId("default").unitId("default").agentId(name).build(); + registry.registerCard(id, card); + + // Subscribe to the agent's request topic and echo back as a response. + String requestTopic = org.apache.eventmesh.protocol.a2a.A2ATopicFactory + .agentRequestTopic("global", name); + transport.subscribe(requestTopic, (topic, event) -> { + String taskId = event.getId(); + try { + String respTopic = org.apache.eventmesh.protocol.a2a.A2ATopicFactory + .gatewayResponseTopic("global", "test-gateway", taskId); + io.cloudevents.CloudEvent resp = io.cloudevents.core.builder.CloudEventBuilder.v1() + .withId(taskId) + .withType("org.apache.eventmesh.protocol.a2a.task.response") + .withSource(java.net.URI.create("agent/" + name)) + .withDataContentType("application/json") + .withData(("{\"echo\":\"" + new String(event.getData().toBytes(), + java.nio.charset.StandardCharsets.UTF_8) + "\"}") + .getBytes(java.nio.charset.StandardCharsets.UTF_8)) + .build(); + transport.publish(respTopic, resp); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Test + void submitTaskCreatesPendingRecord() throws Exception { + CompletableFuture f = gateway.submitTask(testAgentName, "{\"q\":\"hi\"}", null); + // The synchronous echo agent delivers a response almost immediately; wait for it. + TaskResult r = f.get(2, TimeUnit.SECONDS); + assertEquals(TaskState.COMPLETED, r.getState()); + + // The task record should now be COMPLETED in the persistent store. + TaskSnapshot snap = gateway.getTaskStatus("never-existed-task"); + assertNull(snap); + + // Verify at least one task was created and is COMPLETED. + List tasks = store.listByAgent(testAgentName, null); + assertEquals(1, tasks.size()); + assertEquals(Status.COMPLETED, tasks.get(0).status); + assertNotNull(tasks.get(0).output); + } + + @Test + void cancelMarksTaskCanceled() throws Exception { + // Submit to a non-echo agent (no auto-responder) to keep the task in PENDING. + String silentAgent = "silent-" + System.nanoTime(); + AgentCard card = AgentCard.builder() + .name(silentAgent).description("Silent").version("1.0.0") + .supportedInterfaces(Arrays.asList(AgentInterface.builder() + .url("http://localhost:0/a2a").protocolBinding("JSONRPC").protocolVersion("0.3").build())) + .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) + .skills(new ArrayList<>()) + .defaultInputModes(Arrays.asList("text/plain")) + .defaultOutputModes(Arrays.asList("text/plain")) + .build(); + registry.registerCard( + AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); + + // submit with a custom taskId so we can cancel before the timeout fires + String taskId = "task-cancel-" + System.nanoTime(); + CompletableFuture f = gateway.submitTask(taskId, silentAgent, "{}", null); + // Cancel before the response + boolean ok = gateway.cancelTask(taskId); + assertTrue(ok, "cancel should succeed on a PENDING task"); + + // The store should now reflect CANCELED + TaskRecord rec = store.getTask(taskId); + assertNotNull(rec); + assertEquals(Status.CANCELED, rec.status); + + // The future should be completed with a CANCELLED result + TaskResult r = f.get(1, TimeUnit.SECONDS); + assertEquals(TaskState.CANCELLED, r.getState()); + } + + @Test + void cancelOnUnknownTaskIsNoop() { + assertFalse(gateway.cancelTask("task-never-existed")); + } + + @Test + void submitToUnregisteredAgentFails() { + CompletableFuture f = gateway.submitTask("ghost-agent", "{}", null); + org.junit.jupiter.api.Assertions.assertThrows(java.util.concurrent.ExecutionException.class, + () -> f.get(1, TimeUnit.SECONDS)); + } + + @Test + void parentChildIndexIsTrackedInRuntimeCache() throws Exception { + String silentAgent = "silent-pc-" + System.nanoTime(); + AgentCard card = AgentCard.builder() + .name(silentAgent).description("Silent pc").version("1.0.0") + .supportedInterfaces(Arrays.asList(AgentInterface.builder() + .url("http://localhost:0/a2a").protocolBinding("JSONRPC").protocolVersion("0.3").build())) + .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) + .skills(new ArrayList<>()) + .defaultInputModes(Arrays.asList("text/plain")) + .defaultOutputModes(Arrays.asList("text/plain")) + .build(); + registry.registerCard( + AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); + + String parent = "parent-" + System.nanoTime(); + String child1 = "child1-" + System.nanoTime(); + String child2 = "child2-" + System.nanoTime(); + gateway.submitTask(parent, silentAgent, "{}", null); + gateway.submitTask(child1, silentAgent, "{}", parent); + gateway.submitTask(child2, silentAgent, "{}", parent); + + List children = gateway.getChildTasks(parent); + assertEquals(2, children.size()); + assertTrue(children.contains(child1)); + assertTrue(children.contains(child2)); + } + + @Test + void taskNotFoundInStore() { + assertNull(gateway.getTaskStatus("never-existed")); + } +} diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java new file mode 100644 index 0000000000..1c7df222c9 --- /dev/null +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import org.apache.eventmesh.runtime.state.TaskStore; + +import java.net.HttpURLConnection; +import java.net.URI; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Collections; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.cloudevents.CloudEvent; + +/** + * Sub-PR D1: smoke-test the A2A Gateway HTTP server boot path. Boots the Netty + * server on a fixed port, hits the /a2a/health endpoint over loopback, + * asserts 200 OK, then shuts the server down. + */ +class A2AGatewaySmokeTest { + + static final class StubTaskStore implements TaskStore { + @Override public org.apache.eventmesh.runtime.state.TaskStore.TaskRecord createTask( + String taskId, String agentId, String clientId, String input) { return null; } + @Override public org.apache.eventmesh.runtime.state.TaskStore.TaskRecord getTask(String taskId) { return null; } + @Override public boolean updateStatus(String taskId, long expectedTaskEpoch, + org.apache.eventmesh.runtime.state.TaskStore.Status newStatus, String output) { return false; } + @Override public java.util.List listByAgent( + String agentId, org.apache.eventmesh.runtime.state.TaskStore.Status statusFilter) { return Collections.emptyList(); } + @Override public java.util.List expireStale(long olderThanMs) { return Collections.emptyList(); } + @Override public void flush() { } + @Override public void close() { } + } + + static final class NoopTransport implements org.apache.eventmesh.protocol.a2a.A2AMessageTransport { + @Override public void publish(String topic, CloudEvent event) { } + @Override public String subscribe(String topicPattern, + org.apache.eventmesh.protocol.a2a.A2AMessageTransport.MessageCallback callback) { return "noop"; } + @Override public void unsubscribe(String subscriptionId) { } + } + + private A2AGatewayServer server; + private int port; + + @BeforeEach + void setUp() throws Exception { + // Use a high port to avoid collisions with system services. + port = 18080; + server = new A2AGatewayServer(port, new NoopTransport(), new StubTaskStore(), + new InMemoryAgentCardRegistry()); + server.start(); + } + + @AfterEach + void tearDown() throws Exception { + if (server != null) server.shutdown(); + } + + @Test + void healthEndpointRespondsOk() throws Exception { + URL url = URI.create("http://127.0.0.1:" + port + "/a2a/health").toURL(); + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + conn.setRequestMethod("GET"); + conn.setConnectTimeout(2000); + conn.setReadTimeout(2000); + int code = conn.getResponseCode(); + assertEquals(200, code, "health endpoint must return 200"); + byte[] body = conn.getInputStream().readAllBytes(); + String s = new String(body, StandardCharsets.UTF_8); + assertNotNull(s); + assertEquals(true, s.contains("\"status\":\"ok\"")); + } +} From 56d1897062c38afd800ba4f37af599943789e5ac Mon Sep 17 00:00:00 2001 From: qqeasonchen Date: Fri, 28 Aug 2026 15:41:22 +0800 Subject: [PATCH 2/4] fix(a2a): address checkstyle violations in A2A Gateway D1 (PR #5313) A2AGatewayHttpHandler.java: - Remove unused imports (A2AProtocolConstants, A2ATopicFactory, TaskState) - Expand try-catch blocks for limit/offset parsing (EmptyCatchBlock, NeedBraces) - Expand handleList loop control flow (NeedBraces) - Collapse duplicate blank line between package and imports (EmptyLineSeparator) A2AGatewayServiceTest.java: - Expand single-line if-blocks in InProcessTaskStore.listByAgent - Expand if-block in InProcessTransport.matches - Expand if-blocks in tearDown - Move taskId declaration closer to its use (VariableDeclarationUsageDistance) A2AGatewaySmokeTest.java: - Rewrite StubTaskStore / NoopTransport with multi-line method bodies (LeftCurly, RightCurlyAlone, NeedBraces, EmptyLineSeparator) - Add missing static imports (assertEquals, assertNotNull) Build PR #5313 fix. --- .../runtime/a2a/A2AGatewayHttpHandler.java | 671 +++++++++--------- .../runtime/a2a/A2AGatewayServiceTest.java | 662 ++++++++--------- .../runtime/a2a/A2AGatewaySmokeTest.java | 238 ++++--- 3 files changed, 820 insertions(+), 751 deletions(-) diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java index 3295c5a8c3..859ad4e46d 100644 --- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java +++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/a2a/A2AGatewayHttpHandler.java @@ -1,330 +1,341 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; - -import org.apache.eventmesh.protocol.a2a.A2AProtocolConstants; -import org.apache.eventmesh.protocol.a2a.A2ATopicFactory; -import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskState; - -import java.nio.charset.StandardCharsets; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; - -import io.netty.buffer.Unpooled; -import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.SimpleChannelInboundHandler; -import io.netty.handler.codec.http.DefaultFullHttpResponse; -import io.netty.handler.codec.http.FullHttpRequest; -import io.netty.handler.codec.http.FullHttpResponse; -import io.netty.handler.codec.http.HttpHeaderNames; -import io.netty.handler.codec.http.HttpResponseStatus; -import io.netty.handler.codec.http.HttpVersion; -import io.netty.handler.codec.http.QueryStringDecoder; - -import com.fasterxml.jackson.databind.ObjectMapper; - -import lombok.extern.slf4j.Slf4j; - -/** - * Netty HTTP handler for the A2A Gateway REST + SSE API. - * - *

Endpoints (issue #5302 D1 scope): - * - * 

- *   POST   /a2a/tasks              - submit task (sync/async)
- *   GET    /a2a/tasks               - list tasks (?state=COMPLETED&limit=100&offset=0)
- *   GET    /a2a/tasks/{taskId}      - get task status
- *   DELETE /a2a/tasks/{taskId}      - cancel task
- *   GET    /a2a/tasks/{taskId}/wait - wait for result (long-poll)
- *   GET    /a2a/tasks/{taskId}/stream - SSE stream of task status updates
- *   GET    /a2a/health              - health check
- *
- * - *

Note: the SSE stream and the long-poll endpoint resolve the same future; the SSE - * stream additionally registers a {@link A2AGatewayService.StatusSubscriber} so intermediate - * status updates (SUBMITTED -> WORKING -> COMPLETED) are pushed to the client before the - * final result.

- */ -@Slf4j -public class A2AGatewayHttpHandler extends SimpleChannelInboundHandler { - - private static final ObjectMapper objectMapper = new ObjectMapper(); - - private final A2AGatewayService gatewayService; - - public A2AGatewayHttpHandler(A2AGatewayService gatewayService) { - this.gatewayService = gatewayService; - } - - @Override - protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception { - String uri = req.uri(); - try { - if (uri.startsWith("/a2a/tasks/") && uri.endsWith("/stream")) { - handleSse(ctx, req); - } else if (uri.startsWith("/a2a/tasks/") && uri.endsWith("/wait")) { - handleLongPoll(ctx, req); - } else if (uri.startsWith("/a2a/tasks/") && !uri.contains("?")) { - String taskId = uri.substring("/a2a/tasks/".length()); - if ("DELETE".equalsIgnoreCase(req.method().name())) { - handleCancel(ctx, taskId); - } else { - handleGet(ctx, taskId); - } - } else if (uri.equals("/a2a/tasks") || uri.startsWith("/a2a/tasks?")) { - if ("POST".equalsIgnoreCase(req.method().name())) { - handleSubmit(ctx, req); - } else { - handleList(ctx, req); - } - } else if (uri.equals("/a2a/health") || uri.startsWith("/a2a/health?")) { - handleHealth(ctx); - } else { - writeJson(ctx, HttpResponseStatus.NOT_FOUND, "{\"error\":\"not_found\"}"); - } - } catch (Exception e) { - log.error("Error handling {} {}", req.method(), uri, e); - writeJson(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, - "{\"error\":\"internal\",\"message\":\"" + e.getMessage() + "\"}"); - } - } - - // ========================================================================= - // Endpoint handlers - // ========================================================================= - - private void handleSubmit(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception { - String body = req.content().toString(StandardCharsets.UTF_8); - @SuppressWarnings("unchecked") - Map payload = objectMapper.readValue(body, Map.class); - String targetAgent = (String) payload.get("targetAgent"); - String message = (String) payload.get("message"); - String parentTaskId = (String) payload.get("parentTaskId"); - Boolean sync = (Boolean) payload.getOrDefault("sync", Boolean.FALSE); - - if (targetAgent == null || message == null) { - writeJson(ctx, HttpResponseStatus.BAD_REQUEST, - "{\"error\":\"bad_request\",\"message\":\"targetAgent and message are required\"}"); - return; - } - - A2AGatewayService.TaskResult result; - try { - if (Boolean.TRUE.equals(sync)) { - result = gatewayService.submitTask(targetAgent, message, parentTaskId) - .get(10, TimeUnit.SECONDS); - } else { - // Async: return 202 with taskId - String taskId = "task-async-" + System.nanoTime(); - gatewayService.submitTask(taskId, targetAgent, message, parentTaskId); - writeJson(ctx, HttpResponseStatus.ACCEPTED, - "{\"taskId\":\"" + taskId + "\",\"state\":\"SUBMITTED\"}"); - return; - } - } catch (java.util.concurrent.TimeoutException e) { - writeJson(ctx, HttpResponseStatus.GATEWAY_TIMEOUT, - "{\"error\":\"timeout\",\"message\":\"" + e.getMessage() + "\"}"); - return; - } catch (Exception e) { - writeJson(ctx, HttpResponseStatus.BAD_REQUEST, - "{\"error\":\"bad_request\",\"message\":\"" + e.getMessage() + "\"}"); - return; - } - - writeJson(ctx, HttpResponseStatus.OK, toJson(result)); - } - - private void handleGet(ChannelHandlerContext ctx, String taskId) { - A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); - if (snap == null) { - writeJson(ctx, HttpResponseStatus.NOT_FOUND, - "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); - return; - } - writeJson(ctx, HttpResponseStatus.OK, snapshotJson(snap)); - } - - private void handleCancel(ChannelHandlerContext ctx, String taskId) { - boolean ok = gatewayService.cancelTask(taskId); - if (!ok) { - writeJson(ctx, HttpResponseStatus.NOT_FOUND, - "{\"error\":\"not_cancellable\",\"taskId\":\"" + taskId + "\"}"); - return; - } - writeJson(ctx, HttpResponseStatus.OK, "{\"taskId\":\"" + taskId + "\",\"state\":\"CANCELLED\"}"); - } - - private void handleList(ChannelHandlerContext ctx, FullHttpRequest req) { - QueryStringDecoder qsd = new QueryStringDecoder(req.uri()); - String stateFilter = null; - int limit = 100; - int offset = 0; - for (Map.Entry> e : qsd.parameters().entrySet()) { - if ("state".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { - stateFilter = e.getValue().get(0); - } else if ("limit".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { - try { limit = Integer.parseInt(e.getValue().get(0)); } catch (NumberFormatException ignored) { } - } else if ("offset".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { - try { offset = Integer.parseInt(e.getValue().get(0)); } catch (NumberFormatException ignored) { } - } - } - - // Build a synthetic list. The persistent store indexes by agent; the gateway filters - // its known tasks by state. D2 (issue #5302) will introduce a global index. - StringBuilder sb = new StringBuilder("{\"tasks\":["); - int count = 0; - int skipped = 0; - for (var entry : gatewayService.getTaskStore().listByAgent( - gatewayService.getGatewayId(), null).stream() - .sorted((a, b) -> Long.compare(b.createdAtMs, a.createdAtMs)) - .toList()) { - if (stateFilter != null && !stateFilter.equalsIgnoreCase( - A2AGatewayService.toLegacyState(entry.status).name())) { - continue; - } - if (skipped < offset) { skipped++; continue; } - if (count >= limit) { break; } - if (count > 0) sb.append(','); - sb.append("{\"taskId\":\"").append(entry.taskId).append("\",") - .append("\"state\":\"").append(A2AGatewayService.toLegacyState(entry.status)).append("\",") - .append("\"createdAt\":").append(entry.createdAtMs).append('}'); - count++; - } - sb.append("],\"totalListed\":").append(count).append('}'); - - writeJson(ctx, HttpResponseStatus.OK, sb.toString()); - } - - private void handleLongPoll(ChannelHandlerContext ctx, FullHttpRequest req) { - String uri = req.uri(); - String taskId = uri.substring("/a2a/tasks/".length(), uri.length() - "/wait".length()); - A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); - if (snap == null) { - writeJson(ctx, HttpResponseStatus.NOT_FOUND, - "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); - return; - } - if (snap.getRecord().status.ordinal() >= 2) { // COMPLETED, FAILED, CANCELED - writeJson(ctx, HttpResponseStatus.OK, snapshotJson(snap)); - return; - } - // For long-poll, the SSE endpoint is the more general solution; redirect to it - writeJson(ctx, HttpResponseStatus.SEE_OTHER, - "{\"hint\":\"use /a2a/tasks/" + taskId + "/stream for live updates\"}"); - } - - private void handleSse(ChannelHandlerContext ctx, FullHttpRequest req) { - String uri = req.uri(); - String taskId = uri.substring("/a2a/tasks/".length(), uri.length() - "/stream".length()); - A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); - if (snap == null) { - writeJson(ctx, HttpResponseStatus.NOT_FOUND, - "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); - return; - } - - DefaultFullHttpResponse headers = new DefaultFullHttpResponse( - HttpVersion.HTTP_1_1, HttpResponseStatus.OK, - Unpooled.buffer(0)); - headers.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream"); - headers.headers().set(HttpHeaderNames.CACHE_CONTROL, "no-cache"); - headers.headers().set(HttpHeaderNames.CONNECTION, "keep-alive"); - ctx.writeAndFlush(headers); - - // Send current state immediately - ctx.writeAndFlush(Unpooled.wrappedBuffer( - ("data: " + snapshotJson(snap) + "\n\n").getBytes(StandardCharsets.UTF_8))); - - // If terminal, close immediately - if (snap.getRecord().status.ordinal() >= 2) { - ctx.writeAndFlush(Unpooled.wrappedBuffer( - "event: end\ndata: {}\n\n".getBytes(StandardCharsets.UTF_8))) - .addListener(ChannelFutureListener.CLOSE); - return; - } - - // Register subscriber for live updates - gatewayService.registerStatusSubscriber(taskId, (id, state, data) -> { - String line = "data: {\"taskId\":\"" + id + "\",\"state\":\"" + state + "\""; - if (data != null) { - line += ",\"data\":" + objectMapper.valueToTree(data).toString(); - } - line += "}\n\n"; - ChannelFuture f = ctx.writeAndFlush(Unpooled.wrappedBuffer( - line.getBytes(StandardCharsets.UTF_8))); - if ("completed".equals(state) || "failed".equals(state) || "cancelled".equals(state)) { - f.addListener(ChannelFutureListener.CLOSE); - } - }); - } - - private void handleHealth(ChannelHandlerContext ctx) { - writeJson(ctx, HttpResponseStatus.OK, - "{\"status\":\"ok\",\"gatewayId\":\"" + gatewayService.getGatewayId() + "\"}"); - } - - // ========================================================================= - // Helpers - // ========================================================================= - - private void writeJson(ChannelHandlerContext ctx, HttpResponseStatus status, String json) { - FullHttpResponse resp = new DefaultFullHttpResponse( - HttpVersion.HTTP_1_1, status, - Unpooled.wrappedBuffer(json.getBytes(StandardCharsets.UTF_8))); - resp.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json"); - resp.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, "*"); - resp.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, json.length()); - ctx.writeAndFlush(resp); - } - - private String snapshotJson(A2AGatewayService.TaskSnapshot snap) { - return "{" - + "\"taskId\":\"" + snap.getRecord().taskId + "\"," - + "\"state\":\"" + snap.getState() + "\"," - + "\"targetAgent\":\"" + snap.getRecord().agentId + "\"," - + "\"createdAt\":" + snap.getRecord().createdAtMs + "," - + "\"updatedAt\":" + snap.getRecord().updatedAtMs - + (snap.getRecord().output != null - ? ",\"output\":" + objectMapper.valueToTree(snap.getRecord().output).toString() - : "") - + (snap.getParentTaskId() != null - ? ",\"parentTaskId\":\"" + snap.getParentTaskId() + "\"" - : "") - + "}"; - } - - private String toJson(A2AGatewayService.TaskResult r) { - return "{\"state\":\"" + r.getState() + "\"," - + (r.getData() != null - ? "\"data\":" + objectMapper.valueToTree(r.getData()).toString() + "," - : "") - + (r.getErrorMessage() != null - ? "\"errorMessage\":\"" + r.getErrorMessage() + "\"" - : "") - + "}"; - } - - @Override - public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { - log.error("Channel exception", cause); - ctx.close(); - } -} +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.FullHttpResponse; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpVersion; +import io.netty.handler.codec.http.QueryStringDecoder; + +import com.fasterxml.jackson.databind.ObjectMapper; + +import lombok.extern.slf4j.Slf4j; + +/** + * Netty HTTP handler for the A2A Gateway REST + SSE API. + * + *

Endpoints (issue #5302 D1 scope): + * + * 

+ *   POST   /a2a/tasks              - submit task (sync/async)
+ *   GET    /a2a/tasks               - list tasks (?state=COMPLETED&limit=100&offset=0)
+ *   GET    /a2a/tasks/{taskId}      - get task status
+ *   DELETE /a2a/tasks/{taskId}      - cancel task
+ *   GET    /a2a/tasks/{taskId}/wait - wait for result (long-poll)
+ *   GET    /a2a/tasks/{taskId}/stream - SSE stream of task status updates
+ *   GET    /a2a/health              - health check
+ *
+ * + *

Note: the SSE stream and the long-poll endpoint resolve the same future; the SSE + * stream additionally registers a {@link A2AGatewayService.StatusSubscriber} so intermediate + * status updates (SUBMITTED -> WORKING -> COMPLETED) are pushed to the client before the + * final result.

+ */ +@Slf4j +public class A2AGatewayHttpHandler extends SimpleChannelInboundHandler { + + private static final ObjectMapper objectMapper = new ObjectMapper(); + + private final A2AGatewayService gatewayService; + + public A2AGatewayHttpHandler(A2AGatewayService gatewayService) { + this.gatewayService = gatewayService; + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception { + String uri = req.uri(); + try { + if (uri.startsWith("/a2a/tasks/") && uri.endsWith("/stream")) { + handleSse(ctx, req); + } else if (uri.startsWith("/a2a/tasks/") && uri.endsWith("/wait")) { + handleLongPoll(ctx, req); + } else if (uri.startsWith("/a2a/tasks/") && !uri.contains("?")) { + String taskId = uri.substring("/a2a/tasks/".length()); + if ("DELETE".equalsIgnoreCase(req.method().name())) { + handleCancel(ctx, taskId); + } else { + handleGet(ctx, taskId); + } + } else if (uri.equals("/a2a/tasks") || uri.startsWith("/a2a/tasks?")) { + if ("POST".equalsIgnoreCase(req.method().name())) { + handleSubmit(ctx, req); + } else { + handleList(ctx, req); + } + } else if (uri.equals("/a2a/health") || uri.startsWith("/a2a/health?")) { + handleHealth(ctx); + } else { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, "{\"error\":\"not_found\"}"); + } + } catch (Exception e) { + log.error("Error handling {} {}", req.method(), uri, e); + writeJson(ctx, HttpResponseStatus.INTERNAL_SERVER_ERROR, + "{\"error\":\"internal\",\"message\":\"" + e.getMessage() + "\"}"); + } + } + + // ========================================================================= + // Endpoint handlers + // ========================================================================= + + private void handleSubmit(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception { + String body = req.content().toString(StandardCharsets.UTF_8); + @SuppressWarnings("unchecked") + Map payload = objectMapper.readValue(body, Map.class); + String targetAgent = (String) payload.get("targetAgent"); + String message = (String) payload.get("message"); + String parentTaskId = (String) payload.get("parentTaskId"); + Boolean sync = (Boolean) payload.getOrDefault("sync", Boolean.FALSE); + + if (targetAgent == null || message == null) { + writeJson(ctx, HttpResponseStatus.BAD_REQUEST, + "{\"error\":\"bad_request\",\"message\":\"targetAgent and message are required\"}"); + return; + } + + A2AGatewayService.TaskResult result; + try { + if (Boolean.TRUE.equals(sync)) { + result = gatewayService.submitTask(targetAgent, message, parentTaskId) + .get(10, TimeUnit.SECONDS); + } else { + // Async: return 202 with taskId + String taskId = "task-async-" + System.nanoTime(); + gatewayService.submitTask(taskId, targetAgent, message, parentTaskId); + writeJson(ctx, HttpResponseStatus.ACCEPTED, + "{\"taskId\":\"" + taskId + "\",\"state\":\"SUBMITTED\"}"); + return; + } + } catch (java.util.concurrent.TimeoutException e) { + writeJson(ctx, HttpResponseStatus.GATEWAY_TIMEOUT, + "{\"error\":\"timeout\",\"message\":\"" + e.getMessage() + "\"}"); + return; + } catch (Exception e) { + writeJson(ctx, HttpResponseStatus.BAD_REQUEST, + "{\"error\":\"bad_request\",\"message\":\"" + e.getMessage() + "\"}"); + return; + } + + writeJson(ctx, HttpResponseStatus.OK, toJson(result)); + } + + private void handleGet(ChannelHandlerContext ctx, String taskId) { + A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); + if (snap == null) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); + return; + } + writeJson(ctx, HttpResponseStatus.OK, snapshotJson(snap)); + } + + private void handleCancel(ChannelHandlerContext ctx, String taskId) { + boolean ok = gatewayService.cancelTask(taskId); + if (!ok) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_cancellable\",\"taskId\":\"" + taskId + "\"}"); + return; + } + writeJson(ctx, HttpResponseStatus.OK, "{\"taskId\":\"" + taskId + "\",\"state\":\"CANCELLED\"}"); + } + + private void handleList(ChannelHandlerContext ctx, FullHttpRequest req) { + QueryStringDecoder qsd = new QueryStringDecoder(req.uri()); + String stateFilter = null; + int limit = 100; + int offset = 0; + for (Map.Entry> e : qsd.parameters().entrySet()) { + if ("state".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { + stateFilter = e.getValue().get(0); + } else if ("limit".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { + try { + limit = Integer.parseInt(e.getValue().get(0)); + } catch (NumberFormatException ignored) { + // keep default limit on invalid input + } + } else if ("offset".equalsIgnoreCase(e.getKey()) && !e.getValue().isEmpty()) { + try { + offset = Integer.parseInt(e.getValue().get(0)); + } catch (NumberFormatException ignored) { + // keep default offset on invalid input + } + } + } + + // Build a synthetic list. The persistent store indexes by agent; the gateway filters + // its known tasks by state. D2 (issue #5302) will introduce a global index. + StringBuilder sb = new StringBuilder("{\"tasks\":["); + int count = 0; + int skipped = 0; + for (var entry : gatewayService.getTaskStore().listByAgent( + gatewayService.getGatewayId(), null).stream() + .sorted((a, b) -> Long.compare(b.createdAtMs, a.createdAtMs)) + .toList()) { + if (stateFilter != null && !stateFilter.equalsIgnoreCase( + A2AGatewayService.toLegacyState(entry.status).name())) { + continue; + } + if (skipped < offset) { + skipped++; + continue; + } + if (count >= limit) { + break; + } + if (count > 0) { + sb.append(','); + } + sb.append("{\"taskId\":\"").append(entry.taskId).append("\",") + .append("\"state\":\"").append(A2AGatewayService.toLegacyState(entry.status)).append("\",") + .append("\"createdAt\":").append(entry.createdAtMs).append('}'); + count++; + } + sb.append("],\"totalListed\":").append(count).append('}'); + + writeJson(ctx, HttpResponseStatus.OK, sb.toString()); + } + + private void handleLongPoll(ChannelHandlerContext ctx, FullHttpRequest req) { + String uri = req.uri(); + String taskId = uri.substring("/a2a/tasks/".length(), uri.length() - "/wait".length()); + A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); + if (snap == null) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); + return; + } + if (snap.getRecord().status.ordinal() >= 2) { // COMPLETED, FAILED, CANCELED + writeJson(ctx, HttpResponseStatus.OK, snapshotJson(snap)); + return; + } + // For long-poll, the SSE endpoint is the more general solution; redirect to it + writeJson(ctx, HttpResponseStatus.SEE_OTHER, + "{\"hint\":\"use /a2a/tasks/" + taskId + "/stream for live updates\"}"); + } + + private void handleSse(ChannelHandlerContext ctx, FullHttpRequest req) { + String uri = req.uri(); + String taskId = uri.substring("/a2a/tasks/".length(), uri.length() - "/stream".length()); + A2AGatewayService.TaskSnapshot snap = gatewayService.getTaskStatus(taskId); + if (snap == null) { + writeJson(ctx, HttpResponseStatus.NOT_FOUND, + "{\"error\":\"not_found\",\"taskId\":\"" + taskId + "\"}"); + return; + } + + DefaultFullHttpResponse headers = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, HttpResponseStatus.OK, + Unpooled.buffer(0)); + headers.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/event-stream"); + headers.headers().set(HttpHeaderNames.CACHE_CONTROL, "no-cache"); + headers.headers().set(HttpHeaderNames.CONNECTION, "keep-alive"); + ctx.writeAndFlush(headers); + + // Send current state immediately + ctx.writeAndFlush(Unpooled.wrappedBuffer( + ("data: " + snapshotJson(snap) + "\n\n").getBytes(StandardCharsets.UTF_8))); + + // If terminal, close immediately + if (snap.getRecord().status.ordinal() >= 2) { + ctx.writeAndFlush(Unpooled.wrappedBuffer( + "event: end\ndata: {}\n\n".getBytes(StandardCharsets.UTF_8))) + .addListener(ChannelFutureListener.CLOSE); + return; + } + + // Register subscriber for live updates + gatewayService.registerStatusSubscriber(taskId, (id, state, data) -> { + String line = "data: {\"taskId\":\"" + id + "\",\"state\":\"" + state + "\""; + if (data != null) { + line += ",\"data\":" + objectMapper.valueToTree(data).toString(); + } + line += "}\n\n"; + ChannelFuture f = ctx.writeAndFlush(Unpooled.wrappedBuffer( + line.getBytes(StandardCharsets.UTF_8))); + if ("completed".equals(state) || "failed".equals(state) || "cancelled".equals(state)) { + f.addListener(ChannelFutureListener.CLOSE); + } + }); + } + + private void handleHealth(ChannelHandlerContext ctx) { + writeJson(ctx, HttpResponseStatus.OK, + "{\"status\":\"ok\",\"gatewayId\":\"" + gatewayService.getGatewayId() + "\"}"); + } + + // ========================================================================= + // Helpers + // ========================================================================= + + private void writeJson(ChannelHandlerContext ctx, HttpResponseStatus status, String json) { + FullHttpResponse resp = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, status, + Unpooled.wrappedBuffer(json.getBytes(StandardCharsets.UTF_8))); + resp.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json"); + resp.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, "*"); + resp.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, json.length()); + ctx.writeAndFlush(resp); + } + + private String snapshotJson(A2AGatewayService.TaskSnapshot snap) { + return "{" + + "\"taskId\":\"" + snap.getRecord().taskId + "\"," + + "\"state\":\"" + snap.getState() + "\"," + + "\"targetAgent\":\"" + snap.getRecord().agentId + "\"," + + "\"createdAt\":" + snap.getRecord().createdAtMs + "," + + "\"updatedAt\":" + snap.getRecord().updatedAtMs + + (snap.getRecord().output != null + ? ",\"output\":" + objectMapper.valueToTree(snap.getRecord().output).toString() + : "") + + (snap.getParentTaskId() != null + ? ",\"parentTaskId\":\"" + snap.getParentTaskId() + "\"" + : "") + + "}"; + } + + private String toJson(A2AGatewayService.TaskResult r) { + return "{\"state\":\"" + r.getState() + "\"," + + (r.getData() != null + ? "\"data\":" + objectMapper.valueToTree(r.getData()).toString() + "," + : "") + + (r.getErrorMessage() != null + ? "\"errorMessage\":\"" + r.getErrorMessage() + "\"" + : "") + + "}"; + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + log.error("Channel exception", cause); + ctx.close(); + } +} diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java index 1e3b0a024c..78e3ab0f47 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java @@ -1,326 +1,336 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import org.apache.eventmesh.protocol.a2a.AgentIdentity; -import org.apache.eventmesh.protocol.a2a.model.AgentCard; -import org.apache.eventmesh.protocol.a2a.model.AgentCapabilities; -import org.apache.eventmesh.protocol.a2a.model.AgentInterface; -import org.apache.eventmesh.protocol.a2a.model.AgentSkill; -import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskResult; -import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskSnapshot; -import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskState; -import org.apache.eventmesh.runtime.state.TaskStore; -import org.apache.eventmesh.runtime.state.TaskStore.Status; -import org.apache.eventmesh.runtime.state.TaskStore.TaskRecord; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; - -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import io.cloudevents.CloudEvent; - -/** - * Sub-PR D1: A2AGatewayService is wired to {@link TaskStore} (Sub-PR A/C) instead of - * the in-memory {@code TaskRegistry} from PR #5260. These tests exercise the - * gateway's task lifecycle against an in-process TaskStore that mirrors the - * baseline contract test in {@code TaskStoreTest}. - */ -class A2AGatewayServiceTest { - - /** In-process TaskStore mirroring the Sub-PR A test stub. */ - static final class InProcessTaskStore implements TaskStore { - private final ConcurrentHashMap table = new ConcurrentHashMap<>(); - private final AtomicLong epoch = new AtomicLong(); - - @Override - public TaskRecord createTask(String taskId, String agentId, String clientId, String input) { - long now = System.currentTimeMillis(); - long e = epoch.incrementAndGet(); - TaskRecord rec = new TaskRecord(taskId, agentId, clientId, Status.PENDING, now, now, input, null, e); - return table.putIfAbsent(taskId, rec) == null ? rec : null; - } - - @Override - public TaskRecord getTask(String taskId) { - return table.get(taskId); - } - - @Override - public boolean updateStatus(String taskId, long expectedTaskEpoch, Status newStatus, String output) { - TaskRecord rec = table.get(taskId); - if (rec == null || rec.taskEpoch != expectedTaskEpoch) { - return false; - } - rec.status = newStatus; - rec.updatedAtMs = System.currentTimeMillis(); - rec.output = output; - return true; - } - - @Override - public List listByAgent(String agentId, Status statusFilter) { - List out = new ArrayList<>(); - for (TaskRecord r : table.values()) { - if (!r.agentId.equals(agentId)) continue; - if (statusFilter != null && r.status != statusFilter) continue; - out.add(r); - } - return out; - } - - @Override - public List expireStale(long olderThanMs) { - long deadline = System.currentTimeMillis() - olderThanMs; - List expired = new ArrayList<>(); - for (TaskRecord r : table.values()) { - if (r.updatedAtMs < deadline) { - expired.add(r.taskId); - } - } - for (String id : expired) { - table.remove(id); - } - return expired; - } - - @Override - public void flush() { } - - @Override - public void close() { - table.clear(); - } - } - - /** In-process pub/sub transport: publish delivers synchronously to subscribers on the same topic. */ - static final class InProcessTransport implements org.apache.eventmesh.protocol.a2a.A2AMessageTransport { - final ConcurrentHashMap subs = - new ConcurrentHashMap<>(); - - @Override - public void publish(String topic, CloudEvent event) { - for (var entry : subs.entrySet()) { - if (matches(topic, entry.getKey())) { - entry.getValue().onMessage(topic, event); - } - } - } - - @Override - public String subscribe(String topicPattern, - org.apache.eventmesh.protocol.a2a.A2AMessageTransport.MessageCallback callback) { - subs.put(topicPattern, callback); - return "sub-" + topicPattern; - } - - @Override - public void unsubscribe(String subscriptionId) { - subs.remove(subscriptionId.replace("sub-", "")); - } - - private boolean matches(String topic, String pattern) { - if (pattern.equals(topic)) return true; - // A2A topics use Pulsar/MQTT-style wildcards: + matches one path segment, * - // matches one or more. Convert to a regex for in-process delivery. - String regex = pattern - .replace("+", "[^/]+") - .replace("*", "[^/]+"); - return topic.matches(regex); - } - } - - private A2AGatewayService gateway; - private InMemoryAgentCardRegistry registry; - private InProcessTransport transport; - private InProcessTaskStore store; - private String testAgentName; - - @BeforeEach - void setUp() throws Exception { - transport = new InProcessTransport(); - store = new InProcessTaskStore(); - registry = new InMemoryAgentCardRegistry(); - - testAgentName = "echo-agent-" + System.nanoTime(); - registerEchoAgent(testAgentName); - - gateway = new A2AGatewayService("global", "test-gateway", transport, store, registry, 5000L); - gateway.start(); - } - - @AfterEach - void tearDown() throws Exception { - if (gateway != null) gateway.shutdown(); - if (store != null) store.close(); - } - - private void registerEchoAgent(String name) throws Exception { - AgentCard card = AgentCard.builder() - .name(name) - .description("Echoes the input back as a response") - .version("1.0.0") - .supportedInterfaces(Arrays.asList(AgentInterface.builder() - .url("http://localhost:0/a2a") - .protocolBinding("JSONRPC") - .protocolVersion("0.3") - .build())) - .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) - .skills(Arrays.asList(AgentSkill.builder() - .id("echo").name("Echo").description("Echoes input") - .tags(Arrays.asList("test", "echo")).build())) - .defaultInputModes(Arrays.asList("text/plain")) - .defaultOutputModes(Arrays.asList("text/plain")) - .build(); - AgentIdentity id = AgentIdentity.builder().orgId("default").unitId("default").agentId(name).build(); - registry.registerCard(id, card); - - // Subscribe to the agent's request topic and echo back as a response. - String requestTopic = org.apache.eventmesh.protocol.a2a.A2ATopicFactory - .agentRequestTopic("global", name); - transport.subscribe(requestTopic, (topic, event) -> { - String taskId = event.getId(); - try { - String respTopic = org.apache.eventmesh.protocol.a2a.A2ATopicFactory - .gatewayResponseTopic("global", "test-gateway", taskId); - io.cloudevents.CloudEvent resp = io.cloudevents.core.builder.CloudEventBuilder.v1() - .withId(taskId) - .withType("org.apache.eventmesh.protocol.a2a.task.response") - .withSource(java.net.URI.create("agent/" + name)) - .withDataContentType("application/json") - .withData(("{\"echo\":\"" + new String(event.getData().toBytes(), - java.nio.charset.StandardCharsets.UTF_8) + "\"}") - .getBytes(java.nio.charset.StandardCharsets.UTF_8)) - .build(); - transport.publish(respTopic, resp); - } catch (Exception e) { - throw new RuntimeException(e); - } - }); - } - - @Test - void submitTaskCreatesPendingRecord() throws Exception { - CompletableFuture f = gateway.submitTask(testAgentName, "{\"q\":\"hi\"}", null); - // The synchronous echo agent delivers a response almost immediately; wait for it. - TaskResult r = f.get(2, TimeUnit.SECONDS); - assertEquals(TaskState.COMPLETED, r.getState()); - - // The task record should now be COMPLETED in the persistent store. - TaskSnapshot snap = gateway.getTaskStatus("never-existed-task"); - assertNull(snap); - - // Verify at least one task was created and is COMPLETED. - List tasks = store.listByAgent(testAgentName, null); - assertEquals(1, tasks.size()); - assertEquals(Status.COMPLETED, tasks.get(0).status); - assertNotNull(tasks.get(0).output); - } - - @Test - void cancelMarksTaskCanceled() throws Exception { - // Submit to a non-echo agent (no auto-responder) to keep the task in PENDING. - String silentAgent = "silent-" + System.nanoTime(); - AgentCard card = AgentCard.builder() - .name(silentAgent).description("Silent").version("1.0.0") - .supportedInterfaces(Arrays.asList(AgentInterface.builder() - .url("http://localhost:0/a2a").protocolBinding("JSONRPC").protocolVersion("0.3").build())) - .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) - .skills(new ArrayList<>()) - .defaultInputModes(Arrays.asList("text/plain")) - .defaultOutputModes(Arrays.asList("text/plain")) - .build(); - registry.registerCard( - AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); - - // submit with a custom taskId so we can cancel before the timeout fires - String taskId = "task-cancel-" + System.nanoTime(); - CompletableFuture f = gateway.submitTask(taskId, silentAgent, "{}", null); - // Cancel before the response - boolean ok = gateway.cancelTask(taskId); - assertTrue(ok, "cancel should succeed on a PENDING task"); - - // The store should now reflect CANCELED - TaskRecord rec = store.getTask(taskId); - assertNotNull(rec); - assertEquals(Status.CANCELED, rec.status); - - // The future should be completed with a CANCELLED result - TaskResult r = f.get(1, TimeUnit.SECONDS); - assertEquals(TaskState.CANCELLED, r.getState()); - } - - @Test - void cancelOnUnknownTaskIsNoop() { - assertFalse(gateway.cancelTask("task-never-existed")); - } - - @Test - void submitToUnregisteredAgentFails() { - CompletableFuture f = gateway.submitTask("ghost-agent", "{}", null); - org.junit.jupiter.api.Assertions.assertThrows(java.util.concurrent.ExecutionException.class, - () -> f.get(1, TimeUnit.SECONDS)); - } - - @Test - void parentChildIndexIsTrackedInRuntimeCache() throws Exception { - String silentAgent = "silent-pc-" + System.nanoTime(); - AgentCard card = AgentCard.builder() - .name(silentAgent).description("Silent pc").version("1.0.0") - .supportedInterfaces(Arrays.asList(AgentInterface.builder() - .url("http://localhost:0/a2a").protocolBinding("JSONRPC").protocolVersion("0.3").build())) - .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) - .skills(new ArrayList<>()) - .defaultInputModes(Arrays.asList("text/plain")) - .defaultOutputModes(Arrays.asList("text/plain")) - .build(); - registry.registerCard( - AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); - - String parent = "parent-" + System.nanoTime(); - String child1 = "child1-" + System.nanoTime(); - String child2 = "child2-" + System.nanoTime(); - gateway.submitTask(parent, silentAgent, "{}", null); - gateway.submitTask(child1, silentAgent, "{}", parent); - gateway.submitTask(child2, silentAgent, "{}", parent); - - List children = gateway.getChildTasks(parent); - assertEquals(2, children.size()); - assertTrue(children.contains(child1)); - assertTrue(children.contains(child2)); - } - - @Test - void taskNotFoundInStore() { - assertNull(gateway.getTaskStatus("never-existed")); - } -} +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.eventmesh.protocol.a2a.AgentIdentity; +import org.apache.eventmesh.protocol.a2a.model.AgentCard; +import org.apache.eventmesh.protocol.a2a.model.AgentCapabilities; +import org.apache.eventmesh.protocol.a2a.model.AgentInterface; +import org.apache.eventmesh.protocol.a2a.model.AgentSkill; +import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskResult; +import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskSnapshot; +import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskState; +import org.apache.eventmesh.runtime.state.TaskStore; +import org.apache.eventmesh.runtime.state.TaskStore.Status; +import org.apache.eventmesh.runtime.state.TaskStore.TaskRecord; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.cloudevents.CloudEvent; + +/** + * Sub-PR D1: A2AGatewayService is wired to {@link TaskStore} (Sub-PR A/C) instead of + * the in-memory {@code TaskRegistry} from PR #5260. These tests exercise the + * gateway's task lifecycle against an in-process TaskStore that mirrors the + * baseline contract test in {@code TaskStoreTest}. + */ +class A2AGatewayServiceTest { + + /** In-process TaskStore mirroring the Sub-PR A test stub. */ + static final class InProcessTaskStore implements TaskStore { + private final ConcurrentHashMap table = new ConcurrentHashMap<>(); + private final AtomicLong epoch = new AtomicLong(); + + @Override + public TaskRecord createTask(String taskId, String agentId, String clientId, String input) { + long now = System.currentTimeMillis(); + long e = epoch.incrementAndGet(); + TaskRecord rec = new TaskRecord(taskId, agentId, clientId, Status.PENDING, now, now, input, null, e); + return table.putIfAbsent(taskId, rec) == null ? rec : null; + } + + @Override + public TaskRecord getTask(String taskId) { + return table.get(taskId); + } + + @Override + public boolean updateStatus(String taskId, long expectedTaskEpoch, Status newStatus, String output) { + TaskRecord rec = table.get(taskId); + if (rec == null || rec.taskEpoch != expectedTaskEpoch) { + return false; + } + rec.status = newStatus; + rec.updatedAtMs = System.currentTimeMillis(); + rec.output = output; + return true; + } + + @Override + public List listByAgent(String agentId, Status statusFilter) { + List out = new ArrayList<>(); + for (TaskRecord r : table.values()) { + if (!r.agentId.equals(agentId)) { + continue; + } + if (statusFilter != null && r.status != statusFilter) { + continue; + } + out.add(r); + } + return out; + } + + @Override + public List expireStale(long olderThanMs) { + long deadline = System.currentTimeMillis() - olderThanMs; + List expired = new ArrayList<>(); + for (TaskRecord r : table.values()) { + if (r.updatedAtMs < deadline) { + expired.add(r.taskId); + } + } + for (String id : expired) { + table.remove(id); + } + return expired; + } + + @Override + public void flush() { } + + @Override + public void close() { + table.clear(); + } + } + + /** In-process pub/sub transport: publish delivers synchronously to subscribers on the same topic. */ + static final class InProcessTransport implements org.apache.eventmesh.protocol.a2a.A2AMessageTransport { + final ConcurrentHashMap subs = + new ConcurrentHashMap<>(); + + @Override + public void publish(String topic, CloudEvent event) { + for (var entry : subs.entrySet()) { + if (matches(topic, entry.getKey())) { + entry.getValue().onMessage(topic, event); + } + } + } + + @Override + public String subscribe(String topicPattern, + org.apache.eventmesh.protocol.a2a.A2AMessageTransport.MessageCallback callback) { + subs.put(topicPattern, callback); + return "sub-" + topicPattern; + } + + @Override + public void unsubscribe(String subscriptionId) { + subs.remove(subscriptionId.replace("sub-", "")); + } + + private boolean matches(String topic, String pattern) { + if (pattern.equals(topic)) { + return true; + } + // A2A topics use Pulsar/MQTT-style wildcards: + matches one path segment, * + // matches one or more. Convert to a regex for in-process delivery. + String regex = pattern + .replace("+", "[^/]+") + .replace("*", "[^/]+"); + return topic.matches(regex); + } + } + + private A2AGatewayService gateway; + private InMemoryAgentCardRegistry registry; + private InProcessTransport transport; + private InProcessTaskStore store; + private String testAgentName; + + @BeforeEach + void setUp() throws Exception { + transport = new InProcessTransport(); + store = new InProcessTaskStore(); + registry = new InMemoryAgentCardRegistry(); + + testAgentName = "echo-agent-" + System.nanoTime(); + registerEchoAgent(testAgentName); + + gateway = new A2AGatewayService("global", "test-gateway", transport, store, registry, 5000L); + gateway.start(); + } + + @AfterEach + void tearDown() throws Exception { + if (gateway != null) { + gateway.shutdown(); + } + if (store != null) { + store.close(); + } + } + + private void registerEchoAgent(String name) throws Exception { + AgentCard card = AgentCard.builder() + .name(name) + .description("Echoes the input back as a response") + .version("1.0.0") + .supportedInterfaces(Arrays.asList(AgentInterface.builder() + .url("http://localhost:0/a2a") + .protocolBinding("JSONRPC") + .protocolVersion("0.3") + .build())) + .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) + .skills(Arrays.asList(AgentSkill.builder() + .id("echo").name("Echo").description("Echoes input") + .tags(Arrays.asList("test", "echo")).build())) + .defaultInputModes(Arrays.asList("text/plain")) + .defaultOutputModes(Arrays.asList("text/plain")) + .build(); + AgentIdentity id = AgentIdentity.builder().orgId("default").unitId("default").agentId(name).build(); + registry.registerCard(id, card); + + // Subscribe to the agent's request topic and echo back as a response. + String requestTopic = org.apache.eventmesh.protocol.a2a.A2ATopicFactory + .agentRequestTopic("global", name); + transport.subscribe(requestTopic, (topic, event) -> { + String taskId = event.getId(); + try { + String respTopic = org.apache.eventmesh.protocol.a2a.A2ATopicFactory + .gatewayResponseTopic("global", "test-gateway", taskId); + io.cloudevents.CloudEvent resp = io.cloudevents.core.builder.CloudEventBuilder.v1() + .withId(taskId) + .withType("org.apache.eventmesh.protocol.a2a.task.response") + .withSource(java.net.URI.create("agent/" + name)) + .withDataContentType("application/json") + .withData(("{\"echo\":\"" + new String(event.getData().toBytes(), + java.nio.charset.StandardCharsets.UTF_8) + "\"}") + .getBytes(java.nio.charset.StandardCharsets.UTF_8)) + .build(); + transport.publish(respTopic, resp); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Test + void submitTaskCreatesPendingRecord() throws Exception { + CompletableFuture f = gateway.submitTask(testAgentName, "{\"q\":\"hi\"}", null); + // The synchronous echo agent delivers a response almost immediately; wait for it. + TaskResult r = f.get(2, TimeUnit.SECONDS); + assertEquals(TaskState.COMPLETED, r.getState()); + + // The task record should now be COMPLETED in the persistent store. + TaskSnapshot snap = gateway.getTaskStatus("never-existed-task"); + assertNull(snap); + + // Verify at least one task was created and is COMPLETED. + List tasks = store.listByAgent(testAgentName, null); + assertEquals(1, tasks.size()); + assertEquals(Status.COMPLETED, tasks.get(0).status); + assertNotNull(tasks.get(0).output); + } + + @Test + void cancelMarksTaskCanceled() throws Exception { + // Submit to a non-echo agent (no auto-responder) to keep the task in PENDING. + String silentAgent = "silent-" + System.nanoTime(); + AgentCard card = AgentCard.builder() + .name(silentAgent).description("Silent").version("1.0.0") + .supportedInterfaces(Arrays.asList(AgentInterface.builder() + .url("http://localhost:0/a2a").protocolBinding("JSONRPC").protocolVersion("0.3").build())) + .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) + .skills(new ArrayList<>()) + .defaultInputModes(Arrays.asList("text/plain")) + .defaultOutputModes(Arrays.asList("text/plain")) + .build(); + registry.registerCard( + AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); + + String taskId = "task-cancel-" + System.nanoTime(); + // submit and then cancel before the timeout fires + CompletableFuture f = gateway.submitTask(taskId, silentAgent, "{}", null); + // Cancel before the response + boolean ok = gateway.cancelTask(taskId); + assertTrue(ok, "cancel should succeed on a PENDING task"); + + // The store should now reflect CANCELED + TaskRecord rec = store.getTask(taskId); + assertNotNull(rec); + assertEquals(Status.CANCELED, rec.status); + + // The future should be completed with a CANCELLED result + TaskResult r = f.get(1, TimeUnit.SECONDS); + assertEquals(TaskState.CANCELLED, r.getState()); + } + + @Test + void cancelOnUnknownTaskIsNoop() { + assertFalse(gateway.cancelTask("task-never-existed")); + } + + @Test + void submitToUnregisteredAgentFails() { + CompletableFuture f = gateway.submitTask("ghost-agent", "{}", null); + org.junit.jupiter.api.Assertions.assertThrows(java.util.concurrent.ExecutionException.class, + () -> f.get(1, TimeUnit.SECONDS)); + } + + @Test + void parentChildIndexIsTrackedInRuntimeCache() throws Exception { + String silentAgent = "silent-pc-" + System.nanoTime(); + AgentCard card = AgentCard.builder() + .name(silentAgent).description("Silent pc").version("1.0.0") + .supportedInterfaces(Arrays.asList(AgentInterface.builder() + .url("http://localhost:0/a2a").protocolBinding("JSONRPC").protocolVersion("0.3").build())) + .capabilities(AgentCapabilities.builder().streaming(false).pushNotifications(false).build()) + .skills(new ArrayList<>()) + .defaultInputModes(Arrays.asList("text/plain")) + .defaultOutputModes(Arrays.asList("text/plain")) + .build(); + registry.registerCard( + AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); + + String parent = "parent-" + System.nanoTime(); + String child1 = "child1-" + System.nanoTime(); + String child2 = "child2-" + System.nanoTime(); + gateway.submitTask(parent, silentAgent, "{}", null); + gateway.submitTask(child1, silentAgent, "{}", parent); + gateway.submitTask(child2, silentAgent, "{}", parent); + + List children = gateway.getChildTasks(parent); + assertEquals(2, children.size()); + assertTrue(children.contains(child1)); + assertTrue(children.contains(child2)); + } + + @Test + void taskNotFoundInStore() { + assertNull(gateway.getTaskStatus("never-existed")); + } +} diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java index 1c7df222c9..4b12f948c6 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java @@ -1,95 +1,143 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; - -import org.apache.eventmesh.runtime.state.TaskStore; - -import java.net.HttpURLConnection; -import java.net.URI; -import java.net.URL; -import java.nio.charset.StandardCharsets; -import java.util.Collections; - -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import io.cloudevents.CloudEvent; - -/** - * Sub-PR D1: smoke-test the A2A Gateway HTTP server boot path. Boots the Netty - * server on a fixed port, hits the /a2a/health endpoint over loopback, - * asserts 200 OK, then shuts the server down. - */ -class A2AGatewaySmokeTest { - - static final class StubTaskStore implements TaskStore { - @Override public org.apache.eventmesh.runtime.state.TaskStore.TaskRecord createTask( - String taskId, String agentId, String clientId, String input) { return null; } - @Override public org.apache.eventmesh.runtime.state.TaskStore.TaskRecord getTask(String taskId) { return null; } - @Override public boolean updateStatus(String taskId, long expectedTaskEpoch, - org.apache.eventmesh.runtime.state.TaskStore.Status newStatus, String output) { return false; } - @Override public java.util.List listByAgent( - String agentId, org.apache.eventmesh.runtime.state.TaskStore.Status statusFilter) { return Collections.emptyList(); } - @Override public java.util.List expireStale(long olderThanMs) { return Collections.emptyList(); } - @Override public void flush() { } - @Override public void close() { } - } - - static final class NoopTransport implements org.apache.eventmesh.protocol.a2a.A2AMessageTransport { - @Override public void publish(String topic, CloudEvent event) { } - @Override public String subscribe(String topicPattern, - org.apache.eventmesh.protocol.a2a.A2AMessageTransport.MessageCallback callback) { return "noop"; } - @Override public void unsubscribe(String subscriptionId) { } - } - - private A2AGatewayServer server; - private int port; - - @BeforeEach - void setUp() throws Exception { - // Use a high port to avoid collisions with system services. - port = 18080; - server = new A2AGatewayServer(port, new NoopTransport(), new StubTaskStore(), - new InMemoryAgentCardRegistry()); - server.start(); - } - - @AfterEach - void tearDown() throws Exception { - if (server != null) server.shutdown(); - } - - @Test - void healthEndpointRespondsOk() throws Exception { - URL url = URI.create("http://127.0.0.1:" + port + "/a2a/health").toURL(); - HttpURLConnection conn = (HttpURLConnection) url.openConnection(); - conn.setRequestMethod("GET"); - conn.setConnectTimeout(2000); - conn.setReadTimeout(2000); - int code = conn.getResponseCode(); - assertEquals(200, code, "health endpoint must return 200"); - byte[] body = conn.getInputStream().readAllBytes(); - String s = new String(body, StandardCharsets.UTF_8); - assertNotNull(s); - assertEquals(true, s.contains("\"status\":\"ok\"")); - } -} +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.eventmesh.runtime.a2a; + +import org.apache.eventmesh.runtime.state.TaskStore; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.net.HttpURLConnection; +import java.net.URI; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Collections; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.cloudevents.CloudEvent; + +/** + * Sub-PR D1: smoke-test the A2A Gateway HTTP server boot path. Boots the Netty + * server on a fixed port, hits the /a2a/health endpoint over loopback, + * asserts 200 OK, then shuts the server down. + */ +class A2AGatewaySmokeTest { + + private A2AGatewayServer server; + private int port; + + @BeforeEach + void setUp() throws Exception { + // Use a high port to avoid collisions with system services. + port = 18080; + server = new A2AGatewayServer(port, new NoopTransport(), new StubTaskStore(), + new InMemoryAgentCardRegistry()); + server.start(); + } + + @AfterEach + void tearDown() throws Exception { + if (server != null) { + server.shutdown(); + } + } + + @Test + void healthEndpointRespondsOk() throws Exception { + URL url = URI.create("http://127.0.0.1:" + port + "/a2a/health").toURL(); + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + conn.setRequestMethod("GET"); + conn.setConnectTimeout(2000); + conn.setReadTimeout(2000); + int code = conn.getResponseCode(); + assertEquals(200, code, "health endpoint must return 200"); + byte[] body = conn.getInputStream().readAllBytes(); + String s = new String(body, StandardCharsets.UTF_8); + assertNotNull(s); + assertEquals(true, s.contains("\"status\":\"ok\"")); + } + + /** + * Stub {@link TaskStore} returning null / empty for every call. Used only to satisfy the + * gateway's constructor; the smoke test never submits a task, so the stub's behavior is + * never exercised. + */ + static final class StubTaskStore implements TaskStore { + + @Override + public TaskRecord createTask(String taskId, String agentId, String clientId, String input) { + return null; + } + + @Override + public TaskRecord getTask(String taskId) { + return null; + } + + @Override + public boolean updateStatus(String taskId, long expectedTaskEpoch, + TaskStore.Status newStatus, String output) { + return false; + } + + @Override + public java.util.List listByAgent(String agentId, TaskStore.Status statusFilter) { + return Collections.emptyList(); + } + + @Override + public java.util.List expireStale(long olderThanMs) { + return Collections.emptyList(); + } + + @Override + public void flush() { + // no buffered writes + } + + @Override + public void close() { + // nothing to release + } + } + + /** + * No-op A2AMessageTransport: every publish is dropped, every subscribe returns a constant id. + */ + static final class NoopTransport implements org.apache.eventmesh.protocol.a2a.A2AMessageTransport { + + @Override + public void publish(String topic, CloudEvent event) { + // drop + } + + @Override + public String subscribe(String topicPattern, + org.apache.eventmesh.protocol.a2a.A2AMessageTransport.MessageCallback callback) { + return "noop"; + } + + @Override + public void unsubscribe(String subscriptionId) { + // no-op + } + } +} From ed0cd0ee7e940aa309da45b86805c76d483b1edf Mon Sep 17 00:00:00 2001 From: qqeasonchen Date: Fri, 28 Aug 2026 16:00:06 +0800 Subject: [PATCH 3/4] fix(a2a): test imports + variable usage distance (PR #5313 followup) A2AGatewayServiceTest.java: - Fix ImportOrder: AgentCapabilities before AgentCard (alphabetical) - Fix VariableDeclarationUsageDistance for f: inline taskId into submitTask call so the declaration is closer to first use A2AGatewaySmokeTest.java: - Fix ImportOrder: static Assertions imports before non-static TaskStore import Build PR #5313 followup. --- .../eventmesh/runtime/a2a/A2AGatewayServiceTest.java | 8 +++++--- .../apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java | 4 ++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java index 78e3ab0f47..fb4a229d99 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java @@ -24,8 +24,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import org.apache.eventmesh.protocol.a2a.AgentIdentity; -import org.apache.eventmesh.protocol.a2a.model.AgentCard; import org.apache.eventmesh.protocol.a2a.model.AgentCapabilities; +import org.apache.eventmesh.protocol.a2a.model.AgentCard; import org.apache.eventmesh.protocol.a2a.model.AgentInterface; import org.apache.eventmesh.protocol.a2a.model.AgentSkill; import org.apache.eventmesh.runtime.a2a.A2AGatewayService.TaskResult; @@ -272,9 +272,11 @@ void cancelMarksTaskCanceled() throws Exception { registry.registerCard( AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); - String taskId = "task-cancel-" + System.nanoTime(); // submit and then cancel before the timeout fires - CompletableFuture f = gateway.submitTask(taskId, silentAgent, "{}", null); + CompletableFuture f = gateway.submitTask( + "task-cancel-" + System.nanoTime(), silentAgent, "{}", null); + // extract taskId from the result for cancellation + String taskId = f.get(0, TimeUnit.SECONDS).getTaskId(); // Cancel before the response boolean ok = gateway.cancelTask(taskId); assertTrue(ok, "cancel should succeed on a PENDING task"); diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java index 4b12f948c6..acf21c79e8 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java @@ -17,11 +17,11 @@ package org.apache.eventmesh.runtime.a2a; -import org.apache.eventmesh.runtime.state.TaskStore; - import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import org.apache.eventmesh.runtime.state.TaskStore; + import java.net.HttpURLConnection; import java.net.URI; import java.net.URL; From 116d5851ce8664eeb47335faf00bc42f28554f15 Mon Sep 17 00:00:00 2001 From: qqeasonchen Date: Fri, 28 Aug 2026 16:00:54 +0800 Subject: [PATCH 4/4] fix(a2a): correct cancelMarksTaskCanceled test body (PR #5313 followup) The previous fixup broke the test logic by calling TaskResult.getTaskId() which does not exist. Restore the original taskId declaration pattern while keeping the f declaration within 3 lines of its first use: - Declare taskId first - Declare f immediately after - Inline cancelTask result into the assertTrue (no intermediate boolean) - Move f.get() call right after the cancel assertion so f's first use is within 3 lines of its declaration Build PR #5313 followup. --- .../runtime/a2a/A2AGatewayServiceTest.java | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java index fb4a229d99..dcdd7b5ad9 100644 --- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java +++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java @@ -273,22 +273,17 @@ void cancelMarksTaskCanceled() throws Exception { AgentIdentity.builder().orgId("default").unitId("default").agentId(silentAgent).build(), card); // submit and then cancel before the timeout fires - CompletableFuture f = gateway.submitTask( - "task-cancel-" + System.nanoTime(), silentAgent, "{}", null); - // extract taskId from the result for cancellation - String taskId = f.get(0, TimeUnit.SECONDS).getTaskId(); - // Cancel before the response - boolean ok = gateway.cancelTask(taskId); - assertTrue(ok, "cancel should succeed on a PENDING task"); + String taskId = "task-cancel-" + System.nanoTime(); + CompletableFuture f = gateway.submitTask(taskId, silentAgent, "{}", null); + assertTrue(gateway.cancelTask(taskId), "cancel should succeed on a PENDING task"); + // The future should be completed with a CANCELLED result + TaskResult r = f.get(1, TimeUnit.SECONDS); + assertEquals(TaskState.CANCELLED, r.getState()); // The store should now reflect CANCELED TaskRecord rec = store.getTask(taskId); assertNotNull(rec); assertEquals(Status.CANCELED, rec.status); - - // The future should be completed with a CANCELLED result - TaskResult r = f.get(1, TimeUnit.SECONDS); - assertEquals(TaskState.CANCELLED, r.getState()); } @Test