From dd5a8faed57fd4e4c003f510f6ed1529908634a2 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Tue, 1 Sep 2026 15:19:24 +0800 Subject: [PATCH 1/3] docs: generalize A2A-T callback examples Signed-off-by: Zhoujie628 --- docs/en/BUSINESS_CALLBACKS.md | 6 +++--- docs/zh/BUSINESS_CALLBACKS.md | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index 35550b6f..a65ef8b0 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -103,12 +103,12 @@ A2ATClient in the host, not in WorkflowEngineClientConfig or ExecutePsop. ```java // sdk, data, schema and template selection are owned by host-agent code. MetadataContent generated = sdk.generateTaskPromptFromDataWithSchema( - data, schema, StandardTemplates.PRIVATE_LINE_COMPLAINT.uri()); + data, schema, templateUri); MessageContent outgoing = A2atMessages.from(generated, List.of(new TextPart("Process this task"))); ``` -Natural-language equivalent: `sdk.generateTaskPromptFromText(text, StandardTemplates.PRIVATE_LINE_COMPLAINT.uri())`, then use the same -A2atMessages.from conversion. +Natural-language equivalent: `sdk.generateTaskPromptFromText(text, templateUri)`, then use the same A2atMessages.from +conversion. The host agent selects `templateUri` for its business domain. Natural-language generation/validation and all other A2A-T content APIs are called directly on the host's SDK. A2atMessages.from copies generated metadata and activates exactly its extension; supplied parts stay unchanged. It does diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 87a00761..95759196 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -92,11 +92,11 @@ WorkflowEngineClientConfig 或 ExecutePsop 承载 LLM 配置。 ```java // sdk、data、schema、模板选择均属于宿主智能体。 MetadataContent generated = sdk.generateTaskPromptFromDataWithSchema( - data, schema, StandardTemplates.PRIVATE_LINE_COMPLAINT.uri()); + data, schema, templateUri); MessageContent outgoing = A2atMessages.from(generated, List.of(new TextPart("处理当前任务"))); ``` -自然语言入口为 `sdk.generateTaskPromptFromText(text, StandardTemplates.PRIVATE_LINE_COMPLAINT.uri())`，随后同样调用 A2atMessages.from。 +自然语言入口为 `sdk.generateTaskPromptFromText(text, templateUri)`，随后同样调用 A2atMessages.from；`templateUri` 由宿主智能体按业务选择。 自然语言、结构化生成和校验等能力直接调用宿主 SDK。A2atMessages.from 只复制生成的 metadata 并激活对应扩展，parts 保持业务提供的内容；不生成、不校验业务语义。应选择当前 A2A-T SDK 发布的模板 URI，不自行构造。样例初始化工具属于宿主示例，不是引擎依赖。 From 5c90efffea8f7a211318dbb018eb9fadec4c5daa Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Tue, 1 Sep 2026 15:22:13 +0800 Subject: [PATCH 2/3] docs: keep generic callback example compilable Signed-off-by: Zhoujie628 --- docs/en/BUSINESS_CALLBACKS.md | 4 +++- docs/zh/BUSINESS_CALLBACKS.md | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index a65ef8b0..e0fd8be9 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -103,7 +103,9 @@ A2ATClient in the host, not in WorkflowEngineClientConfig or ExecutePsop. ```java // sdk, data, schema and template selection are owned by host-agent code. MetadataContent generated = sdk.generateTaskPromptFromDataWithSchema( - data, schema, templateUri); + data, schema, + Objects.requireNonNull( + System.getProperty("host.task-template-uri"), "host.task-template-uri")); MessageContent outgoing = A2atMessages.from(generated, List.of(new TextPart("Process this task"))); ``` diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 95759196..ebcb6439 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -92,7 +92,9 @@ WorkflowEngineClientConfig 或 ExecutePsop 承载 LLM 配置。 ```java // sdk、data、schema、模板选择均属于宿主智能体。 MetadataContent generated = sdk.generateTaskPromptFromDataWithSchema( - data, schema, templateUri); + data, schema, + Objects.requireNonNull( + System.getProperty("host.task-template-uri"), "host.task-template-uri")); MessageContent outgoing = A2atMessages.from(generated, List.of(new TextPart("处理当前任务"))); ``` From 8884e29a3507ffabf7f7f6977a3fd836104eea38 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Tue, 1 Sep 2026 20:40:08 +0800 Subject: [PATCH 3/3] feat(a2a): add preflight task cleanup support Signed-off-by: Zhoujie628 --- docs/en/API_REFERENCE.md | 6 + docs/en/INTEGRATION_GUIDE.md | 14 ++ docs/zh/API_REFERENCE.md | 5 + docs/zh/INTEGRATION_GUIDE.md | 12 ++ .../config/WorkbenchClientProperties.java | 36 +++++ .../examples/server/JdkHttpA2AServer.java | 44 ++++++ .../SpringWorkbenchExtensionLifecycle.java | 18 ++- .../examples/workbench/StaleTaskCleaner.java | 132 ++++++++++++++++++ .../WorkbenchExtensionLifecycle.java | 108 +++++++++++++- samples/src/main/resources/application.yml | 4 + .../demo/SpringSpnDemoFailureE2ETest.java | 14 ++ .../server/EmbeddedA2AServerTest.java | 33 +++++ .../workbench/StaleTaskCleanerTest.java | 126 +++++++++++++++++ .../workflow/engine/spring/A2AController.java | 31 ++++ .../spring/A2AAutoConfigurationTest.java | 15 ++ .../engine/client/A2AJavaClientRuntime.java | 8 ++ .../workflow/engine/client/A2ATransport.java | 26 ++++ .../client/DefaultA2AJavaClientRuntime.java | 15 ++ .../client/DefaultWorkflowEngineClient.java | 17 +++ .../engine/client/WorkflowEngineClient.java | 5 + .../engine/StubWorkflowEngineClient.java | 7 + .../engine/client/A2ATransportHeaderTest.java | 39 +++++- 22 files changed, 710 insertions(+), 5 deletions(-) create mode 100644 samples/src/main/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleaner.java create mode 100644 samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index f20dd20c..f3db72c7 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -61,6 +61,7 @@ ExecutionResult result = ExecutePsop.builder() CompletableFuture dispatch(TaskRequest request, MessageContent content, ControlPoint callbacks); CompletableFuture sendMessage(String agentName, MessageContent content); CompletableFuture getTask(String agentName, String taskId); +CompletableFuture listTasks(String agentName, ListTasksParams params); CompletableFuture cancelTask(String agentName, String taskId); CompletableFuture subscribeToTask(String agentName, String taskId, Consumer> callback); long callbackTimeoutSeconds(); @@ -69,6 +70,11 @@ void setEventCallback(EventCallback callback); void close(); ``` +`listTasks` and `cancelTask` expose the standard A2A task-management operations through the same +AgentCard, authentication and custom runtime as message dispatch. A list result is authorization-scoped: +it contains only tasks visible to the authenticated identity. Callers must paginate with +`nextPageToken`; cancellation is not implied by closing a local client or notification stream. + The executor calls dispatch; onTask does not send. Content is final parts/metadata/extensions. The engine manages envelopes and interactions, never instantiates A2ATClient or generates content from AgentCard declarations. Template queries and generation belong to the host SDK. diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index f73bc23d..dac010d9 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -535,6 +535,20 @@ The engine consumes AgentCards and never starts dispatched-agent servers. The ho endpoint reachability checks, and any development fixtures. When using externally managed dispatched agents, do not start local fixtures on the same endpoints. Keep production AgentCards and credentials outside tracked sample resources. +### 13.3 Demo preflight task cleanup + +The demo queries each dispatched agent for `SUBMITTED`, `WORKING`, `INPUT_REQUIRED` and `AUTH_REQUIRED` tasks before +opening extension channels or starting a workflow, then cancels every visible result through the standard A2A task API. +It follows `nextPageToken`, de-duplicates tasks that change state during the scan, and accepts the race where a task +becomes terminal before cancellation. Query and cancellation use a dedicated short-lived authenticated transport; they +never reuse workflow, authorization or notification lifecycles. + +Cleanup is enabled and fail-fast by default so stale tasks cannot silently lead to a capacity error. Configure it with +`A2A_TASK_CLEANUP_ENABLED`, `A2A_TASK_CLEANUP_FAIL_FAST`, `A2A_TASK_CLEANUP_PAGE_SIZE` (1–100), and +`A2A_TASK_CLEANUP_MAX_TASKS`. The list operation is authorization-scoped. If several installations share one identity, +the demo may cancel active tasks created by another installation; use an isolated identity or disable cleanup after +providing an equivalent ownership-aware policy. + ## 14. Remote problem responses A successful HTTP envelope does not guarantee a successful task: SSE data can contain a top-level diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index c95569cb..c69dbf6d 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -61,6 +61,7 @@ ExecutionResult result = ExecutePsop.builder() CompletableFuture dispatch(TaskRequest request, MessageContent content, ControlPoint callbacks); CompletableFuture sendMessage(String agentName, MessageContent content); CompletableFuture getTask(String agentName, String taskId); +CompletableFuture listTasks(String agentName, ListTasksParams params); CompletableFuture cancelTask(String agentName, String taskId); CompletableFuture subscribeToTask(String agentName, String taskId, Consumer> callback); long callbackTimeoutSeconds(); @@ -69,6 +70,10 @@ void setEventCallback(EventCallback callback); void close(); ``` +`listTasks` 与 `cancelTask` 通过消息下发使用的同一 AgentCard、认证配置和自定义运行时暴露标准 A2A +任务管理操作。查询结果受认证身份权限约束，只包含当前身份可见的任务；调用方必须使用 +`nextPageToken` 完成分页。关闭本地客户端或通知流不等同于取消远端任务。 + 执行器内部调用 dispatch；onTask 不自行发送。内容是最终 parts/metadata/extensions，引擎只管理信封和交互，不创建 A2ATClient，不按 AgentCard 声明生成内容。模板查询和生成接口请直接使用宿主 SDK。 diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index 87510315..fdfe9768 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -502,6 +502,18 @@ LoadPsop 不修改 JVM 全局 SSLContext、默认 SocketFactory 或 HostnameVeri 引擎只消费 AgentCard，不启动被调度智能体服务。宿主智能体负责 AgentCard 发现、端点可达性校验与开发测试资源。使用外部管理的被调度智能体时，不得在相同端点上启动本地测试服务。生产 AgentCard 和凭据应保存在仓库外部。 +### 13.3 Demo 启动前任务清理 + +Demo 在打开独立协议通道和启动工作流之前，先查询每个被调度智能体中处于 `SUBMITTED`、`WORKING`、 +`INPUT_REQUIRED`、`AUTH_REQUIRED` 状态的任务，再通过标准 A2A 任务接口取消当前认证身份可见的全部结果。 +查询会跟随 `nextPageToken`，对扫描过程中发生状态变化的任务去重，并正确处理“查询后、取消前任务已进入终态”的竞态。 +查询和取消使用独立的短生命周期认证传输，不复用工作流、授权或通知的生命周期。 + +清理默认开启并采用 fail-fast，避免遗留任务静默累积后触发容量错误。可通过 +`A2A_TASK_CLEANUP_ENABLED`、`A2A_TASK_CLEANUP_FAIL_FAST`、`A2A_TASK_CLEANUP_PAGE_SIZE`（1–100）和 +`A2A_TASK_CLEANUP_MAX_TASKS` 配置。任务查询受认证身份权限约束；若多个实例共用同一身份，Demo 可能取消其他实例创建的活跃任务。 +应使用隔离身份，或在提供等价的任务归属清理策略后关闭该功能。 + ## 14. 远端错误响应 响应头成功不代表任务成功。HTTP 200 的 SSE data 也可能返回顶层错误对象，例如： diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/config/WorkbenchClientProperties.java b/samples/src/main/java/dev/openan/workflow/engine/examples/config/WorkbenchClientProperties.java index a57790cd..df8e65ae 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/config/WorkbenchClientProperties.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/config/WorkbenchClientProperties.java @@ -16,6 +16,10 @@ public class WorkbenchClientProperties { private boolean sslVerify; private String a2atEnvPath = ""; private boolean demoNegotiationEnabled; + private boolean taskCleanupEnabled = true; + private boolean taskCleanupFailFast = true; + private int taskCleanupPageSize = 100; + private int taskCleanupMaxTasks = 1000; /** * Internal sample-host setting; SpringSpnDemo supplies its local-only default per application @@ -61,4 +65,36 @@ public String getA2atEnvPath() { public void setA2atEnvPath(String a2atEnvPath) { this.a2atEnvPath = a2atEnvPath; } + + public boolean isTaskCleanupEnabled() { + return taskCleanupEnabled; + } + + public void setTaskCleanupEnabled(boolean taskCleanupEnabled) { + this.taskCleanupEnabled = taskCleanupEnabled; + } + + public boolean isTaskCleanupFailFast() { + return taskCleanupFailFast; + } + + public void setTaskCleanupFailFast(boolean taskCleanupFailFast) { + this.taskCleanupFailFast = taskCleanupFailFast; + } + + public int getTaskCleanupPageSize() { + return taskCleanupPageSize; + } + + public void setTaskCleanupPageSize(int taskCleanupPageSize) { + this.taskCleanupPageSize = taskCleanupPageSize; + } + + public int getTaskCleanupMaxTasks() { + return taskCleanupMaxTasks; + } + + public void setTaskCleanupMaxTasks(int taskCleanupMaxTasks) { + this.taskCleanupMaxTasks = taskCleanupMaxTasks; + } } diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java b/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java index a672451a..68a325bc 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java @@ -419,6 +419,22 @@ private void handleExchange(HttpExchange exchange, RestHandler restHandler) thro handleStream(exchange, restHandler, readBody(exchange)); return; } + if ("GET".equalsIgnoreCase(method) && "/tasks".equals(path)) { + Map query = queryParams(exchange); + var resp = + restHandler.listTasks( + buildCallContext(exchange), + "", + query.get("contextId"), + query.get("status"), + integerParam(query, "pageSize"), + query.get("pageToken"), + integerParam(query, "historyLength"), + query.get("statusTimestampAfter"), + booleanParam(query, "includeArtifacts")); + sendJson(exchange, resp.getStatusCode(), resp.getBody()); + return; + } // Task routes (?? A2A: ??/??/??) String taskId = taskIdFromPath(path, null); if ("GET".equalsIgnoreCase(method) && taskId != null) { @@ -449,6 +465,34 @@ private void handleExchange(HttpExchange exchange, RestHandler restHandler) thro } } + private static Map queryParams(HttpExchange exchange) { + String rawQuery = exchange.getRequestURI().getRawQuery(); + if (rawQuery == null || rawQuery.isBlank()) return Map.of(); + Map values = new LinkedHashMap<>(); + for (String pair : rawQuery.split("&")) { + String[] fields = pair.split("=", 2); + String key = URLDecoder.decode(fields[0], StandardCharsets.UTF_8); + String value = + fields.length == 2 ? URLDecoder.decode(fields[1], StandardCharsets.UTF_8) : ""; + values.put(key, value); + } + return values; + } + + private static Integer integerParam(Map query, String name) { + String value = query.get(name); + return value == null || value.isBlank() ? null : Integer.valueOf(value); + } + + private static Boolean booleanParam(Map query, String name) { + String value = query.get(name); + if (value == null || value.isBlank()) return null; + if (!"true".equalsIgnoreCase(value) && !"false".equalsIgnoreCase(value)) { + throw new IllegalArgumentException(name + " must be true or false"); + } + return Boolean.valueOf(value); + } + @SuppressWarnings("unchecked") private void handleLogin(HttpExchange exchange) throws IOException { if (!"PUT".equalsIgnoreCase(exchange.getRequestMethod()) diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/SpringWorkbenchExtensionLifecycle.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/SpringWorkbenchExtensionLifecycle.java index 2ef07898..bf3998fc 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/SpringWorkbenchExtensionLifecycle.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/SpringWorkbenchExtensionLifecycle.java @@ -7,6 +7,7 @@ import dev.openan.workflow.engine.examples.config.WorkbenchClientProperties; import dev.openan.workflow.engine.examples.util.EnvResolver; import jakarta.annotation.PreDestroy; +import org.a2aproject.sdk.spec.AgentCard; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.context.event.ApplicationReadyEvent; @@ -20,12 +21,15 @@ public final class SpringWorkbenchExtensionLifecycle { LoggerFactory.getLogger(SpringWorkbenchExtensionLifecycle.class); private final WorkbenchClientProperties properties; + private final String localAgentName; private WorkbenchExtensionLifecycle lifecycle; private volatile java.util.concurrent.CompletableFuture firstRecovery = new java.util.concurrent.CompletableFuture<>(); - public SpringWorkbenchExtensionLifecycle(WorkbenchClientProperties properties) { + public SpringWorkbenchExtensionLifecycle( + WorkbenchClientProperties properties, AgentCard localAgentCard) { this.properties = properties; + this.localAgentName = localAgentCard.name(); } @EventListener(ApplicationReadyEvent.class) @@ -40,10 +44,20 @@ public synchronized void start() { properties.isSslVerify(), resolveEnvPath(), null, - this::onNotification); + this::onNotification, + null, + properties.isTaskCleanupEnabled(), + properties.isTaskCleanupFailFast(), + properties.getTaskCleanupPageSize(), + properties.getTaskCleanupMaxTasks(), + localAgentName); try { candidate.start(); lifecycle = candidate; + } catch (WorkbenchExtensionLifecycle.TaskCleanupException e) { + candidate.close(); + lifecycle = null; + throw e; } catch (RuntimeException e) { candidate.close(); lifecycle = null; diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleaner.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleaner.java new file mode 100644 index 00000000..32d7dd67 --- /dev/null +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleaner.java @@ -0,0 +1,132 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.examples.workbench; + +import dev.openan.workflow.engine.client.WorkflowEngineClient; +import dev.openan.workflow.engine.model.SendMessageResult; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletionException; +import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.ListTasksParams; +import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskState; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Cancels non-terminal A2A tasks visible to the demo's authenticated identity before startup. */ +final class StaleTaskCleaner { + private static final Logger log = LoggerFactory.getLogger(StaleTaskCleaner.class); + private static final List ACTIVE_STATES = + List.of( + TaskState.TASK_STATE_SUBMITTED, + TaskState.TASK_STATE_WORKING, + TaskState.TASK_STATE_INPUT_REQUIRED, + TaskState.TASK_STATE_AUTH_REQUIRED); + + private final int pageSize; + private final int maxTasks; + + StaleTaskCleaner(int pageSize, int maxTasks) { + if (pageSize < 1 || pageSize > 100) { + throw new IllegalArgumentException("task cleanup pageSize must be between 1 and 100"); + } + if (maxTasks < 1) { + throw new IllegalArgumentException("task cleanup maxTasks must be positive"); + } + this.pageSize = pageSize; + this.maxTasks = maxTasks; + } + + CleanupReport cleanup(WorkflowEngineClient client, List agentCards) { + int listed = 0; + int canceled = 0; + int becameTerminal = 0; + for (AgentCard card : agentCards) { + Map active = listActiveTasks(client, card.name()); + listed += active.size(); + for (Task task : active.values()) { + if (cancel(client, card.name(), task)) canceled++; + else becameTerminal++; + } + } + return new CleanupReport(listed, canceled, becameTerminal); + } + + private Map listActiveTasks(WorkflowEngineClient client, String agentName) { + Map active = new LinkedHashMap<>(); + for (TaskState state : ACTIVE_STATES) { + String pageToken = null; + Set seenTokens = new java.util.HashSet<>(); + do { + ListTasksParams.Builder query = + ListTasksParams.builder() + .status(state) + .pageSize(pageSize) + .historyLength(0) + .includeArtifacts(false); + if (pageToken != null) query.pageToken(pageToken); + var page = client.listTasks(agentName, query.build()).join(); + for (Task task : page.tasks()) { + if (task.status() != null && ACTIVE_STATES.contains(task.status().state())) { + active.putIfAbsent(task.id(), task); + if (active.size() > maxTasks) { + throw new IllegalStateException( + "Task cleanup limit exceeded for " + agentName + ": " + maxTasks); + } + } + } + pageToken = page.nextPageToken(); + if (pageToken != null && !pageToken.isBlank() && !seenTokens.add(pageToken)) { + throw new IllegalStateException( + "Task cleanup received a repeated page token from " + agentName); + } + } while (pageToken != null && !pageToken.isBlank()); + } + log.info("[TaskCleanup] LIST_DONE agent={}, activeTasks={}", agentName, active.size()); + return active; + } + + private boolean cancel(WorkflowEngineClient client, String agentName, Task task) { + try { + SendMessageResult result = client.cancelTask(agentName, task.id()).join(); + if (result.getTask() == null + || result.getTask().status() == null + || result.getTask().status().state() != TaskState.TASK_STATE_CANCELED) { + throw new IllegalStateException( + "Cancellation did not reach CANCELED for " + agentName + "/" + task.id()); + } + log.info( + "[TaskCleanup] CANCEL_DONE agent={}, taskId={}, previousState={}", + agentName, + task.id(), + task.status().state()); + return true; + } catch (CompletionException | IllegalStateException cancellationError) { + SendMessageResult current; + try { + current = client.getTask(agentName, task.id()).join(); + } catch (RuntimeException queryError) { + cancellationError.addSuppressed(queryError); + throw cancellationError; + } + if (current.getTask() != null + && current.getTask().status() != null + && current.getTask().status().state().isFinal()) { + log.info( + "[TaskCleanup] CANCEL_SKIP agent={}, taskId={}, reason=became_terminal, state={}", + agentName, + task.id(), + current.getTask().status().state()); + return false; + } + throw cancellationError; + } + } + + record CleanupReport(int listed, int canceled, int becameTerminal) {} +} diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java index 27e4ed9c..e77a0cd0 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java @@ -29,6 +29,11 @@ public final class WorkbenchExtensionLifecycle implements AutoCloseable { private final boolean sslVerify; private final String a2atEnvPath; private final Supplier runtimeSupplier; + private final boolean taskCleanupEnabled; + private final boolean taskCleanupFailFast; + private final int taskCleanupPageSize; + private final int taskCleanupMaxTasks; + private final String localAgentName; private final java.util.function.BiConsumer< NotificationSubscription, dev.openan.workflow.engine.model.ReceivedMessage> notificationCallback; @@ -43,7 +48,18 @@ public WorkbenchExtensionLifecycle( java.util.function.BiConsumer< NotificationSubscription, dev.openan.workflow.engine.model.ReceivedMessage> notificationCallback) { - this(credentialsPath, sslVerify, a2atEnvPath, runtimeSupplier, notificationCallback, null); + this( + credentialsPath, + sslVerify, + a2atEnvPath, + runtimeSupplier, + notificationCallback, + null, + false, + true, + 100, + 1000, + null); } public WorkbenchExtensionLifecycle( @@ -51,12 +67,45 @@ public WorkbenchExtensionLifecycle( Supplier runtimeSupplier, java.util.function.BiConsumer notificationCallback, dev.openan.workflow.engine.client.AuthProvider authProvider) { + this( + credentialsPath, + sslVerify, + a2atEnvPath, + runtimeSupplier, + notificationCallback, + authProvider, + false, + true, + 100, + 1000, + null); + } + + public WorkbenchExtensionLifecycle( + String credentialsPath, + boolean sslVerify, + String a2atEnvPath, + Supplier runtimeSupplier, + java.util.function.BiConsumer< + NotificationSubscription, dev.openan.workflow.engine.model.ReceivedMessage> + notificationCallback, + dev.openan.workflow.engine.client.AuthProvider authProvider, + boolean taskCleanupEnabled, + boolean taskCleanupFailFast, + int taskCleanupPageSize, + int taskCleanupMaxTasks, + String localAgentName) { this.authProvider = authProvider; this.credentialsPath = credentialsPath; this.sslVerify = sslVerify; this.a2atEnvPath = a2atEnvPath; this.runtimeSupplier = runtimeSupplier; this.notificationCallback = notificationCallback; + this.taskCleanupEnabled = taskCleanupEnabled; + this.taskCleanupFailFast = taskCleanupFailFast; + this.taskCleanupPageSize = taskCleanupPageSize; + this.taskCleanupMaxTasks = taskCleanupMaxTasks; + this.localAgentName = localAgentName; } private static long elapsedMillis(long startedNanos) { @@ -71,6 +120,7 @@ public synchronized void start() { return; } List agentCards = new WorkbenchAgentCatalog().load(); + cleanupStaleTasks(agentCards); A2AJavaClientRuntime authorizationRuntime = null; A2AJavaClientRuntime notificationRuntime = null; A2ATransport authorizationTransport = null; @@ -125,6 +175,56 @@ public synchronized void start() { } } + private void cleanupStaleTasks(List agentCards) { + if (!taskCleanupEnabled) { + log.info("[TaskCleanup] SKIP reason=disabled"); + return; + } + long started = System.nanoTime(); + A2AJavaClientRuntime cleanupRuntime = null; + A2ATransport cleanupTransport = null; + try { + cleanupRuntime = runtimeSupplier != null ? runtimeSupplier.get() : null; + WorkflowEngineClientConfig config = + WorkflowEngineClientConfig.builder() + .sslVerify(sslVerify) + .credentialsConfigPath(credentialsPath) + .authProvider(authProvider) + .build(); + cleanupTransport = new A2ATransport(agentCards, cleanupRuntime, config); + List cleanupTargets = + localAgentName == null + ? agentCards + : agentCards.stream() + .filter(card -> !localAgentName.equals(card.name())) + .toList(); + var report = + new StaleTaskCleaner(taskCleanupPageSize, taskCleanupMaxTasks) + .cleanup( + new dev.openan.workflow.engine.client.DefaultWorkflowEngineClient(cleanupTransport), + cleanupTargets); + log.info( + "[TaskCleanup] DONE listed={}, canceled={}, becameTerminal={}, elapsedMs={}", + report.listed(), + report.canceled(), + report.becameTerminal(), + elapsedMillis(started)); + } catch (RuntimeException error) { + log.error( + "[TaskCleanup] FAILED elapsedMs={}, errorType={}, message={}, failFast={}", + elapsedMillis(started), + error.getClass().getSimpleName(), + error.getMessage(), + taskCleanupFailFast, + error); + if (taskCleanupFailFast) { + throw new TaskCleanupException("Failed to clean stale remote tasks", error); + } + } finally { + closeResource(cleanupTransport, cleanupRuntime); + } + } + private static void closeResource(A2ATransport transport, A2AJavaClientRuntime runtime) { try { if (transport != null) transport.close(); @@ -168,4 +268,10 @@ private void handleNotification( } } } + + public static final class TaskCleanupException extends RuntimeException { + TaskCleanupException(String message, Throwable cause) { + super(message, cause); + } + } } diff --git a/samples/src/main/resources/application.yml b/samples/src/main/resources/application.yml index d13bab05..bce52403 100644 --- a/samples/src/main/resources/application.yml +++ b/samples/src/main/resources/application.yml @@ -20,3 +20,7 @@ a2at: a2a: orch-url: https://127.0.0.1:5001 ssl-verify: ${A2A_SSL_VERIFY:false} + task-cleanup-enabled: ${A2A_TASK_CLEANUP_ENABLED:true} + task-cleanup-fail-fast: ${A2A_TASK_CLEANUP_FAIL_FAST:true} + task-cleanup-page-size: ${A2A_TASK_CLEANUP_PAGE_SIZE:100} + task-cleanup-max-tasks: ${A2A_TASK_CLEANUP_MAX_TASKS:1000} diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java index acde806a..7b1f1031 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java @@ -117,6 +117,20 @@ private static String reason(int status) { private static HttpServer omc(int status, AtomicInteger calls) throws Exception { HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); server.createContext("/", exchange -> { + if ("GET".equals(exchange.getRequestMethod()) + && exchange.getRequestURI().getPath().endsWith("/tasks")) { + byte[] bytes = + JSON.writeValueAsBytes( + Map.of("tasks", List.of(), "totalSize", 0, "pageSize", 0)); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, bytes.length); + try (var output = exchange.getResponseBody()) { + output.write(bytes); + } finally { + exchange.close(); + } + return; + } var request = JSON.readTree(exchange.getRequestBody()); boolean task = request.path("message").path("metadata").has(A2ATExtension.TASK_T.uri()); String response; diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java index 300466ba..1fd25f9c 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java @@ -45,6 +45,8 @@ import net.openan.a2at.sdk.core.model.*; import net.openan.a2at.sdk.negotiation.content.*; import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.ListTasksParams; +import org.a2aproject.sdk.spec.TaskState; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; @@ -531,6 +533,37 @@ void taskRouteParserDecodesExactlyOnePathSegment() { assertNull(JdkHttpA2AServer.taskIdFromPath("/tasks/id:subscribe", ":cancel")); } + @Test + void standardTaskListAndCancelManageAnInputRequiredTask() { + SendMessageResult pending = startIncomplete(); + + var listed = + client + .listTasks( + AGENT_NAME, + ListTasksParams.builder() + .status(TaskState.TASK_STATE_INPUT_REQUIRED) + .pageSize(100) + .build()) + .join(); + + assertTrue(listed.tasks().stream().anyMatch(task -> task.id().equals(pending.getTask().id()))); + SendMessageResult canceled = client.cancelTask(AGENT_NAME, pending.getTask().id()).join(); + assertEquals("TASK_STATE_CANCELED", canceled.getTaskState()); + assertTrue( + client + .listTasks( + AGENT_NAME, + ListTasksParams.builder() + .status(TaskState.TASK_STATE_INPUT_REQUIRED) + .pageSize(100) + .build()) + .join() + .tasks() + .stream() + .noneMatch(task -> task.id().equals(pending.getTask().id()))); + } + private class RecordingCity1 extends SpnDomainAgentCity1Executor { @Override protected String executeBusiness( diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java new file mode 100644 index 00000000..a9576f8c --- /dev/null +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java @@ -0,0 +1,126 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.examples.workbench; + +import static org.junit.jupiter.api.Assertions.*; + +import dev.openan.workflow.engine.client.WorkflowEngineClient; +import dev.openan.workflow.engine.control.ControlPoint; +import dev.openan.workflow.engine.control.EventCallback; +import dev.openan.workflow.engine.model.MessageContent; +import dev.openan.workflow.engine.model.SendMessageResult; +import dev.openan.workflow.engine.model.TaskRequest; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.function.Consumer; +import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; +import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.ListTasksParams; +import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskState; +import org.a2aproject.sdk.spec.TaskStatus; +import org.junit.jupiter.api.Test; + +class StaleTaskCleanerTest { + + @Test + void cancelsEveryVisibleNonTerminalTask() { + FakeClient client = new FakeClient(); + client.tasks.add(task("working", TaskState.TASK_STATE_WORKING)); + client.tasks.add(task("waiting", TaskState.TASK_STATE_INPUT_REQUIRED)); + client.tasks.add(task("done", TaskState.TASK_STATE_COMPLETED)); + + var report = new StaleTaskCleaner(100, 10).cleanup(client, List.of(agentCard())); + + assertEquals(2, report.listed()); + assertEquals(2, report.canceled()); + assertEquals(List.of("working", "waiting"), client.canceledTaskIds); + } + + @Test + void acceptsTaskThatBecomesTerminalBeforeCancellation() { + FakeClient client = new FakeClient(); + client.tasks.add(task("racing", TaskState.TASK_STATE_WORKING)); + client.failCancellation = true; + + var report = new StaleTaskCleaner(100, 10).cleanup(client, List.of(agentCard())); + + assertEquals(1, report.listed()); + assertEquals(0, report.canceled()); + assertEquals(1, report.becameTerminal()); + } + + private static AgentCard agentCard() { + return new WorkbenchAgentCatalog().load().get(0); + } + + private static Task task(String id, TaskState state) { + return Task.builder() + .id(id) + .contextId("ctx-" + id) + .status(new TaskStatus(state)) + .build(); + } + + private static final class FakeClient implements WorkflowEngineClient { + private final List tasks = new ArrayList<>(); + private final List canceledTaskIds = new ArrayList<>(); + private boolean failCancellation; + + @Override + public CompletableFuture listTasks(String agentName, ListTasksParams params) { + return CompletableFuture.completedFuture( + new ListTasksResult( + tasks.stream().filter(task -> task.status().state() == params.status()).toList())); + } + + @Override + public CompletableFuture cancelTask(String agentName, String taskId) { + if (failCancellation) { + return CompletableFuture.failedFuture(new IllegalStateException("already terminal")); + } + canceledTaskIds.add(taskId); + return CompletableFuture.completedFuture(result(task(taskId, TaskState.TASK_STATE_CANCELED))); + } + + @Override + public CompletableFuture getTask(String agentName, String taskId) { + return CompletableFuture.completedFuture(result(task(taskId, TaskState.TASK_STATE_COMPLETED))); + } + + private static SendMessageResult result(Task task) { + return SendMessageResult.builder().task(task).taskState(task.status().state().name()).build(); + } + + @Override + public CompletableFuture dispatch( + TaskRequest request, MessageContent content, ControlPoint callbacks) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture sendMessage( + String agentName, MessageContent content) { + throw new UnsupportedOperationException(); + } + + @Override + public void setControlPoint(ControlPoint controlPoint) {} + + @Override + public void setEventCallback(EventCallback callback) {} + + @Override + public CompletableFuture subscribeToTask( + String agentName, String taskId, Consumer> eventCallback) { + throw new UnsupportedOperationException(); + } + + @Override + public void close() {} + } +} diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java index 271e9b84..0bbd9661 100644 --- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java +++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java @@ -47,6 +47,7 @@ import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @@ -168,6 +169,36 @@ public ResponseEntity getTask(HttpServletRequest req, @PathVariable("id" .body(resp.getBody()); } + @GetMapping("${a2at.server.path-prefix}/tasks") + public ResponseEntity listTasks( + HttpServletRequest req, + @RequestParam(required = false) String contextId, + @RequestParam(required = false) String status, + @RequestParam(required = false) Integer pageSize, + @RequestParam(required = false) String pageToken, + @RequestParam(required = false) Integer historyLength, + @RequestParam(required = false) String statusTimestampAfter, + @RequestParam(required = false) Boolean includeArtifacts) { + var resp = + restHandler.listTasks( + buildContext(req), + "", + contextId, + status, + pageSize, + pageToken, + historyLength, + statusTimestampAfter, + includeArtifacts); + return ResponseEntity.status(resp.getStatusCode()) + .contentType( + MediaType.parseMediaType( + resp.getContentType() != null + ? resp.getContentType() + : MediaType.APPLICATION_JSON_VALUE)) + .body(resp.getBody()); + } + @PostMapping("${a2at.server.path-prefix}/tasks/{id}:cancel") public ResponseEntity cancelTask( HttpServletRequest req, diff --git a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java index f87a1da5..779b5f07 100644 --- a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java +++ b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java @@ -30,6 +30,21 @@ void everyTaskRouteUsesConfiguredA2aPathPrefix() throws Exception { .getMethod("getTask", jakarta.servlet.http.HttpServletRequest.class, String.class) .getAnnotation(GetMapping.class) .value()[0]); + assertEquals( + placeholder + "/tasks", + A2AController.class + .getMethod( + "listTasks", + jakarta.servlet.http.HttpServletRequest.class, + String.class, + String.class, + Integer.class, + String.class, + Integer.class, + String.class, + Boolean.class) + .getAnnotation(GetMapping.class) + .value()[0]); assertEquals( placeholder + "/tasks/{id}:cancel", A2AController.class diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java index f7ca6a27..d352e9eb 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java @@ -23,7 +23,9 @@ import java.util.function.Consumer; import org.a2aproject.sdk.client.ClientEvent; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.ListTasksParams; import org.a2aproject.sdk.spec.MessageSendParams; import org.a2aproject.sdk.spec.Task; @@ -63,6 +65,12 @@ default Task getTask(AgentCard agentCard, String taskId, ClientCallContext callC throw new UnsupportedOperationException("getTask not supported by this runtime"); } + /** List tasks visible to the authenticated caller (A2A GET tasks). */ + default ListTasksResult listTasks( + AgentCard agentCard, ListTasksParams params, ClientCallContext callContext) { + throw new UnsupportedOperationException("listTasks not supported by this runtime"); + } + /** Cancel a task by ID (A2A POST tasks/{id}:cancel). */ default Task cancelTask(AgentCard agentCard, String taskId, ClientCallContext callContext) { throw new UnsupportedOperationException("cancelTask not supported by this runtime"); diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java index 8cbb426e..fc9eb8ab 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java @@ -45,8 +45,10 @@ import org.a2aproject.sdk.client.TaskEvent; import org.a2aproject.sdk.client.TaskUpdateEvent; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.AgentCard; import org.a2aproject.sdk.spec.Artifact; +import org.a2aproject.sdk.spec.ListTasksParams; import org.a2aproject.sdk.spec.Message; import org.a2aproject.sdk.spec.MessageSendParams; import org.a2aproject.sdk.spec.Part; @@ -759,6 +761,30 @@ public CompletableFuture getTask( asyncExecutor); } + public CompletableFuture listTasks( + AgentCard agentCard, String agentName, ListTasksParams params) { + if (closed.get()) { + return CompletableFuture.failedFuture(new IllegalStateException("A2A transport is closed")); + } + Map trace = new HashMap<>(WireLog.context()); + trace.put("agent", agentName); + trace.put("operation", "listTasks"); + return CompletableFuture.supplyAsync( + () -> + WireLog.call( + trace, + () -> { + try { + ClientCallContext ctx = buildClientCallContext(agentCard, agentName, Map.of()); + return a2aClientRuntime.listTasks(agentCard, params, ctx); + } catch (Exception e) { + throw new RuntimeException( + "listTasks failed for " + agentName + ": " + e.getMessage(), e); + } + }), + asyncExecutor); + } + public CompletableFuture cancelTask( AgentCard agentCard, String agentName, String taskId) { if (closed.get()) { diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java index caeb796b..14faba10 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java @@ -50,10 +50,12 @@ import org.a2aproject.sdk.client.transport.rest.RestTransport; import org.a2aproject.sdk.client.transport.rest.RestTransportConfig; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.A2AClientException; import org.a2aproject.sdk.spec.AgentCard; import org.a2aproject.sdk.spec.AgentInterface; import org.a2aproject.sdk.spec.CancelTaskParams; +import org.a2aproject.sdk.spec.ListTasksParams; import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; import org.a2aproject.sdk.spec.TaskIdParams; @@ -519,6 +521,19 @@ public Task getTask(AgentCard agentCard, String taskId, ClientCallContext callCo } } + @Override + public ListTasksResult listTasks( + AgentCard agentCard, ListTasksParams params, ClientCallContext callContext) { + if (closed.get()) throw new IllegalStateException("A2A client runtime is closed"); + Client client = getOrCreateClient(agentCard, extractAgentUrl(agentCard)); + try { + return client.listTasks(params, callContext); + } catch (A2AClientException e) { + throw new RuntimeException( + "A2A listTasks failed for " + agentCard.name() + ": " + e.getMessage(), e); + } + } + @Override public Task cancelTask(AgentCard agentCard, String taskId, ClientCallContext callContext) { if (closed.get()) throw new IllegalStateException("A2A client runtime is closed"); diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java index 771778d9..be70bcd6 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java @@ -33,7 +33,9 @@ import org.a2aproject.sdk.client.ClientEvent; import org.a2aproject.sdk.client.MessageEvent; import org.a2aproject.sdk.client.TaskUpdateEvent; +import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.ListTasksParams; import org.a2aproject.sdk.spec.Message; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; import org.slf4j.Logger; @@ -442,6 +444,21 @@ public CompletableFuture getTask(String agentName, String tas return transport.getTask(agentCard, agentName, taskId); } + @Override + public CompletableFuture listTasks(String agentName, ListTasksParams params) { + if (closed.get()) + return CompletableFuture.failedFuture(new IllegalStateException("Workflow client closed")); + if (agentName == null || agentName.isBlank() || params == null) { + return CompletableFuture.failedFuture( + new IllegalArgumentException("agentName and params must not be null or blank")); + } + AgentCard agentCard = transport.getCard(agentName); + if (agentCard == null) { + return CompletableFuture.failedFuture(new RuntimeException("Agent not found: " + agentName)); + } + return transport.listTasks(agentCard, agentName, params); + } + @Override public CompletableFuture cancelTask(String agentName, String taskId) { if (closed.get()) diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java index 6934cf85..a22b65bf 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java @@ -27,6 +27,8 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; +import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; +import org.a2aproject.sdk.spec.ListTasksParams; /** Sends final content and coordinates task interaction. Content generation belongs to the host. */ public interface WorkflowEngineClient { @@ -49,6 +51,9 @@ default long callbackTimeoutSeconds() { /** Queries an existing remote task. */ CompletableFuture getTask(String agentName, String taskId); + /** Lists remote tasks visible to the configured authentication identity. */ + CompletableFuture listTasks(String agentName, ListTasksParams params); + /** Cancels an existing remote task; this is not a Negotiation-T Abort. */ CompletableFuture cancelTask(String agentName, String taskId); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java index c065e871..2d884c1c 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java @@ -142,6 +142,13 @@ public CompletableFuture getTask(String agentName, String tas return cannedResult(agentName, defaultTaskState); } + @Override + public CompletableFuture listTasks( + String agentName, org.a2aproject.sdk.spec.ListTasksParams params) { + return CompletableFuture.completedFuture( + new org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult(List.of())); + } + @Override public CompletableFuture cancelTask(String agentName, String taskId) { return cannedResult(agentName, "TASK_STATE_CANCELED"); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java index 6ec28836..086b298a 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java @@ -22,7 +22,9 @@ import java.util.function.Consumer; import org.a2aproject.sdk.client.ClientEvent; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.ListTasksParams; import org.a2aproject.sdk.spec.MessageSendParams; import org.junit.jupiter.api.Test; @@ -222,6 +224,35 @@ void authProviderCanSatisfySecurityRequirementWithoutCredentials() throws Except assertEquals("Bearer gateway-token", capturedHeaders.get().get("Authorization")); } + @Test + void listTasksUsesTheSameAuthenticatedTransportBoundary() throws Exception { + AtomicReference> capturedHeaders = new AtomicReference<>(); + A2AJavaClientRuntime runtime = + new CapturingRuntime(capturedHeaders) { + @Override + public ListTasksResult listTasks( + AgentCard agentCard, ListTasksParams params, ClientCallContext callContext) { + capturedHeaders.set(Map.copyOf(callContext.getHeaders())); + return new ListTasksResult(List.of()); + } + }; + AgentCard card = securedAgentCard(); + WorkflowEngineClientConfig config = + WorkflowEngineClientConfig.builder() + .authProvider( + (agentName, ignored, headers) -> + headers.put("Authorization", "Bearer cleanup-token")) + .build(); + + try (A2ATransport transport = new A2ATransport(List.of(card), runtime, config)) { + new DefaultWorkflowEngineClient(transport) + .listTasks(card.name(), ListTasksParams.builder().pageSize(10).build()) + .join(); + } + + assertEquals("Bearer cleanup-token", capturedHeaders.get().get("Authorization")); + } + @Test void authenticationHeaderConflictsAreCaseInsensitive() { Map headers = new HashMap<>(); @@ -335,8 +366,12 @@ void finalStructuredContentNeedsNoContentSdk() throws Exception { } } - private record CapturingRuntime(AtomicReference> capturedHeaders) - implements A2AJavaClientRuntime { + private static class CapturingRuntime implements A2AJavaClientRuntime { + private final AtomicReference> capturedHeaders; + + private CapturingRuntime(AtomicReference> capturedHeaders) { + this.capturedHeaders = capturedHeaders; + } @Override public Iterable sendMessage(