> 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/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..dcdd7b5ad9
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewayServiceTest.java
@@ -0,0 +1,333 @@
+/*
+ * 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.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;
+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 and then cancel before the timeout fires
+ 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);
+ }
+
+ @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..acf21c79e8
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/a2a/A2AGatewaySmokeTest.java
@@ -0,0 +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 {
+
+ 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
+ }
+ }
+}