feat(a2a): add preflight task cleanup support

docs/en/API_REFERENCE.md

@@ -61,6 +61,7 @@ ExecutionResult result = ExecutePsop.builder()
 CompletableFuture<SendMessageResult> dispatch(TaskRequest request, MessageContent content, ControlPoint callbacks);
 CompletableFuture<SendMessageResult> sendMessage(String agentName, MessageContent content);
 CompletableFuture<SendMessageResult> getTask(String agentName, String taskId);
+CompletableFuture<ListTasksResult> listTasks(String agentName, ListTasksParams params);
 CompletableFuture<SendMessageResult> cancelTask(String agentName, String taskId);
 CompletableFuture<SendMessageResult> subscribeToTask(String agentName, String taskId, Consumer<Map<String,Object>> 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.

docs/en/BUSINESS_CALLBACKS.md

@@ -103,12 +103,14 @@ 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,
+    Objects.requireNonNull(
+        System.getProperty("host.task-template-uri"), "host.task-template-uri"));
 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

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

docs/zh/API_REFERENCE.md

@@ -61,6 +61,7 @@ ExecutionResult result = ExecutePsop.builder()
 CompletableFuture<SendMessageResult> dispatch(TaskRequest request, MessageContent content, ControlPoint callbacks);
 CompletableFuture<SendMessageResult> sendMessage(String agentName, MessageContent content);
 CompletableFuture<SendMessageResult> getTask(String agentName, String taskId);
+CompletableFuture<ListTasksResult> listTasks(String agentName, ListTasksParams params);
 CompletableFuture<SendMessageResult> cancelTask(String agentName, String taskId);
 CompletableFuture<SendMessageResult> subscribeToTask(String agentName, String taskId, Consumer<Map<String,Object>> 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。
 

docs/zh/BUSINESS_CALLBACKS.md

@@ -92,11 +92,13 @@ WorkflowEngineClientConfig 或 ExecutePsop 承载 LLM 配置。
 ```java
 // sdk、data、schema、模板选择均属于宿主智能体。
 MetadataContent generated = sdk.generateTaskPromptFromDataWithSchema(
-    data, schema, StandardTemplates.PRIVATE_LINE_COMPLAINT.uri());
+    data, schema,
+    Objects.requireNonNull(
+        System.getProperty("host.task-template-uri"), "host.task-template-uri"));
 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，不自行构造。样例初始化工具属于宿主示例，不是引擎依赖。
 

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 也可能返回顶层错误对象，例如：

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;
+  }
 }

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<String, String> 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<String, String> queryParams(HttpExchange exchange) {
+    String rawQuery = exchange.getRequestURI().getRawQuery();
+    if (rawQuery == null || rawQuery.isBlank()) return Map.of();
+    Map<String, String> 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<String, String> query, String name) {
+    String value = query.get(name);
+    return value == null || value.isBlank() ? null : Integer.valueOf(value);
+  }
+
+  private static Boolean booleanParam(Map<String, String> 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())

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<Void> 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;

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<TaskState> 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<AgentCard> agentCards) {
+    int listed = 0;
+    int canceled = 0;
+    int becameTerminal = 0;
+    for (AgentCard card : agentCards) {
+      Map<String, Task> 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<String, Task> listActiveTasks(WorkflowEngineClient client, String agentName) {
+    Map<String, Task> active = new LinkedHashMap<>();
+    for (TaskState state : ACTIVE_STATES) {
+      String pageToken = null;
+      Set<String> 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) {}
+}