diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f97bb02..83ee684 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,10 +14,10 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Set up JDK 17 - uses: actions/setup-java@v4 + uses: actions/setup-java@v5 with: java-version: '17' distribution: 'temurin' @@ -32,7 +32,7 @@ jobs: - name: Upload test reports if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v6 with: name: test-reports path: | @@ -46,10 +46,10 @@ jobs: steps: - name: Checkout code - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Set up JDK 17 - uses: actions/setup-java@v4 + uses: actions/setup-java@v5 with: java-version: '17' distribution: 'temurin' @@ -60,7 +60,7 @@ jobs: - name: Upload sample test reports if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v6 with: name: direct-sample-test-reports path: samples/target/surefire-reports/ diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 6b919ea..1202be7 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -45,7 +45,7 @@ jobs: steps: - name: Checkout (full history for tag ancestry validation) - uses: actions/checkout@v4 + uses: actions/checkout@v5 with: fetch-depth: 0 @@ -80,7 +80,7 @@ jobs: test -n "${SIGNING_PASSWORD}" || { echo "SIGNING_PASSWORD is not set"; exit 1; } - name: Set up JDK - uses: actions/setup-java@v4 + uses: actions/setup-java@v5 with: distribution: temurin java-version: '17' diff --git a/CHANGELOG.md b/CHANGELOG.md index 9645f22..5b6dd4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,17 @@ # Changelog -## [Unreleased] — 1.0.0 release candidate +## [Unreleased] + +- Release streaming resources when publisher subscription setup throws synchronously. +- Align published dependency examples with 0.0.2 and refresh neutral architecture terminology. +- Update GitHub Actions to Node 24-compatible action versions. + +## [0.0.2] — 2026-09-07 + +- Publish the main-branch baseline under `net.openan.workflow.sdk`, including the parent POM, + workflow-engine and spring-boot-starter. This tag has the same source tree as 0.0.1. + +## [0.0.1] — 2026-09-04 - Use published A2A-T 1.1.0 from Maven Central; remove SDK source checkout/install from CI; move content generation, validation, templates and SDK initialization to the host. @@ -25,5 +36,5 @@ - Publish a compiled host integration example and require sample regression tests. - Keep documentation host-neutral; sample class and AgentCard identifiers remain unchanged. -This repository prepares the initial 1.0.0 SDK release. Earlier internal development notes are not published API -contracts or evidence of a Maven release. Release publication/tagging is a separate, explicitly authorized operation. +Published Maven artifacts use the version selected by the release tag. The default source-build +`revision=1.0.0` is a development version, not the version to use when consuming Maven Central artifacts. diff --git a/README.md b/README.md index b027a77..dc57111 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,8 @@ subscription remains open until the host-defined terminal event, explicit cancel ### 1. Add Maven dependency +The current engine release is `0.0.2` on [Maven Central](https://repo.maven.apache.org/maven2/net/openan/workflow/sdk/). + A2A-T SDK `1.1.0` is published to Maven Central. Maven resolves it automatically; no SDK source checkout or local SDK build is required. The engine depends only on `a2a-t-core`; host agents using content generation explicitly add `a2a-t-client:1.1.0`. A dispatched-agent service that @@ -49,7 +51,7 @@ upgrade guidance. net.openan.workflow.sdk workflow-engine - 1.0.0 + 0.0.2 ``` @@ -59,7 +61,7 @@ For Spring Boot server-side integration: net.openan.workflow.sdk spring-boot-starter - 1.0.0 + 0.0.2 ``` diff --git a/docs/en/DESIGN.md b/docs/en/DESIGN.md index 550bb17..ab099df 100644 --- a/docs/en/DESIGN.md +++ b/docs/en/DESIGN.md @@ -166,7 +166,7 @@ flowchart TB direction LR ORCH["Orchestration center
workflow definitions (PSOP)"] REGT["Registry center
AgentCards"] - OMC["Dispatched agents (OMC)"] + DISPATCHED["Dispatched agents"] end WB ==>|"selects workflow and starts"| EXE @@ -181,7 +181,7 @@ flowchart TB TRANS --> OBSV TRANS --> A2AJ MSG -.->|"only reference site inside the engine"| A2ATC - A2AJ -->|"Task-T · Negotiation-T
Authorization-T · Notification-T"| OMC + A2AJ -->|"Task-T · Negotiation-T
Authorization-T · Notification-T"| DISPATCHED REGC -->|"workflow search"| ORCH REGC -->|"AgentCard retrieval"| REGT ``` @@ -190,7 +190,7 @@ Key properties of the module view: the engine is embedded in the host process as by host code and extended through the callbacks. The scheduling kernel (core/control/model packages) references no A2A-T SDK types — a boundary guarded by the `ContentDependencyBoundaryTest` architecture test. The client protocol adaptation layer (A2ATExtension, A2atMessages, DefaultWorkflowEngineClient) is the only -place inside the engine that references a2a-t-sdk. All four interaction types with dispatched agents (OMC) — +place inside the engine that references a2a-t-sdk. All four interaction types with dispatched agents — Task-T, Negotiation-T, Authorization-T, Notification-T — travel over a2a-java-sdk transports, with authorization and subscription on channels independent of the workflow. The orchestration and registry centers are only reached by the engine clients for workflow-definition search and AgentCard retrieval; the diff --git a/docs/en/DEVELOPER_GUIDE.md b/docs/en/DEVELOPER_GUIDE.md index 3e4f64a..43bf3d7 100644 --- a/docs/en/DEVELOPER_GUIDE.md +++ b/docs/en/DEVELOPER_GUIDE.md @@ -11,7 +11,7 @@ Add to your `pom.xml`: net.openan.workflow.sdk workflow-engine - 1.0.0 + 0.0.2 ``` diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index 4c0897d..18c3ffd 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -22,7 +22,7 @@ callbacks own A2A-T generation, semantic validation, schemas and any LLM calls. net.openan.workflow.sdk workflow-engine -1.0.0 +0.0.2 ``` @@ -220,7 +220,7 @@ configuration without modifying system properties. Never log plaintext or the ke **Encrypt a password** -Build the jar with `mvn -pl workflow-engine -am package`; commands below run from the repository root. +Build the jar with `mvn -pl workflow-engine -am -Drevision=0.0.2 package`; commands below run from the repository root. `set` is Windows cmd syntax (PowerShell: `$env:A2AT_CRED_KEY='...'`). This CLI needs only the SDK jar and JDK. Use disposable example values here: command-line passwords/keys can appear in shell history and process listings. For production, obtain secrets securely in the host and use the Java encryption API. @@ -228,10 +228,10 @@ For production, obtain secrets securely in the host and use the Java encryption ```bash # Option 1: set env var first set A2AT_CRED_KEY=4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b -java -cp workflow-engine/target/workflow-engine-1.0.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" +java -cp workflow-engine/target/workflow-engine-0.0.2.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" # Option 2: pass key as second argument -java -cp workflow-engine/target/workflow-engine-1.0.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b +java -cp workflow-engine/target/workflow-engine-0.0.2.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b ``` Output: @@ -247,7 +247,7 @@ Paste the output into the `value` field of the credentials JSON. 1. Generate a new key: `openssl rand -hex 32` 2. Update the host secret store / explicit `credentialEncryptionKey`, or its OS/JVM `A2AT_CRED_KEY` 3. Re-encrypt all passwords: - `java -cp workflow-engine/target/workflow-engine-1.0.0.jar dev.openan.workflow.engine.client.CredentialCrypto "plaintext" new-key` + `java -cp workflow-engine/target/workflow-engine-0.0.2.jar dev.openan.workflow.engine.client.CredentialCrypto "plaintext" new-key` 4. Update the `enc:...` results in the credentials JSON file > The `.env` file should not be committed to version control. Add it to `.gitignore`. diff --git a/docs/zh/DESIGN.md b/docs/zh/DESIGN.md index c8928e4..4c692fe 100644 --- a/docs/zh/DESIGN.md +++ b/docs/zh/DESIGN.md @@ -155,7 +155,7 @@ flowchart TB direction LR ORCH["编排中心
工作流定义（PSOP）"] REGT["注册中心
AgentCard"] - OMC["被调度智能体（OMC）"] + DISPATCHED["被调度智能体"] end WB ==>|"选择工作流并启动"| EXE @@ -170,7 +170,7 @@ flowchart TB TRANS --> OBSV TRANS --> A2AJ MSG -.->|"引擎内唯一引用位置"| A2ATC - A2AJ -->|"Task-T · Negotiation-T
Authorization-T · Notification-T"| OMC + A2AJ -->|"Task-T · Negotiation-T
Authorization-T · Notification-T"| DISPATCHED REGC -->|"检索工作流"| ORCH REGC -->|"获取 AgentCard"| REGT ``` @@ -178,7 +178,7 @@ flowchart TB 模块视图要点：引擎以 Maven 依赖嵌入宿主进程，由宿主代码启动并实现回调；调度内核（core/control/model 包）不引用任何 A2A-T SDK 类型，该边界由 `ContentDependencyBoundaryTest` 架构测试守护；client 协议 适配层（A2ATExtension、A2atMessages、DefaultWorkflowEngineClient）是引擎内引用 a2a-t-sdk 的唯一 -位置；与被调度智能体（OMC）的 Task-T、Negotiation-T、Authorization-T、Notification-T 四类交互全部 +位置；与被调度智能体的 Task-T、Negotiation-T、Authorization-T、Notification-T 四类交互全部 经 a2a-java-sdk 传输，授权与订阅使用独立于工作流的通道；编排中心与注册中心仅在工作流定义检索和 AgentCard 获取时被引擎客户端访问，检索与选择策略属宿主职责。任务创建前被对端以非 2xx 和标准 A2A 错误信封拒绝的调用，由传输层识别并投影为稳定错误码（`a2a.`），不进入协商、不自动重试； diff --git a/docs/zh/DEVELOPER_GUIDE.md b/docs/zh/DEVELOPER_GUIDE.md index 5fae170..4a4a508 100644 --- a/docs/zh/DEVELOPER_GUIDE.md +++ b/docs/zh/DEVELOPER_GUIDE.md @@ -10,7 +10,7 @@ net.openan.workflow.sdk workflow-engine - 1.0.0 + 0.0.2 ``` diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index aa86151..bfb215e 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -20,7 +20,7 @@ A2A-T 工作流执行引擎是一个 Java SDK，用于基于 A2A 协议和 A2A-T net.openan.workflow.sdk workflow-engine -1.0.0 +0.0.2 ``` @@ -212,17 +212,17 @@ A2AT_CRED_KEY=4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b **加密密码** -先执行 `mvn -pl workflow-engine -am package`，以下命令在仓库根目录运行，仅需 SDK jar 和 JDK。 +先执行 `mvn -pl workflow-engine -am -Drevision=0.0.2 package`，以下命令在仓库根目录运行，仅需 SDK jar 和 JDK。 `set` 是 Windows cmd 语法，PowerShell 应使用 `$env:A2AT_CRED_KEY='...'`。 以下参数仅用于演示；命令行口令/密钥可能进入终端历史和进程参数。生产应由集成方安全读取密钥并调用 Java 加密 API。 ```bash # 方式一：先设置环境变量 set A2AT_CRED_KEY=4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b -java -cp workflow-engine/target/workflow-engine-1.0.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" +java -cp workflow-engine/target/workflow-engine-0.0.2.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" # 方式二：密钥作为第二个参数 -java -cp workflow-engine/target/workflow-engine-1.0.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b +java -cp workflow-engine/target/workflow-engine-0.0.2.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b ``` 输出： @@ -237,7 +237,7 @@ enc:uHQcTeKZMVNRM9Ga:o5vm4weRozBXBs04phrLq7j7+/yRVyDsrw== 1. 生成新密钥：`openssl rand -hex 32` 2. 更新集成方密钥存储及显式 `credentialEncryptionKey`，或 OS/JVM 的 `A2AT_CRED_KEY`；引擎不自动加载 `.env` -3. 用新密钥重新加密所有密码：`java -cp workflow-engine/target/workflow-engine-1.0.0.jar dev.openan.workflow.engine.client.CredentialCrypto "明文密码" 新密钥` +3. 用新密钥重新加密所有密码：`java -cp workflow-engine/target/workflow-engine-0.0.2.jar dev.openan.workflow.engine.client.CredentialCrypto "明文密码" 新密钥` 4. 将新的 `enc:...` 结果更新到凭证 JSON 文件 > `.env` 文件不应提交到版本库，建议加入 `.gitignore`。 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 e36de61..b54db3c 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 @@ -641,7 +641,7 @@ private static void requireStreamingSupport(AgentCard agentCard) } } - private static void writeStream( + static void writeStream( HttpExchange exchange, Flow.Publisher publisher, ServerCallContext context) @@ -652,7 +652,7 @@ private static void writeStream( CountDownLatch done = new CountDownLatch(1); OutputStream output = exchange.getResponseBody(); StreamCancellation cancellation = new StreamCancellation(context); - publisher.subscribe( + Flow.Subscriber subscriber = new Flow.Subscriber<>() { @Override public void onSubscribe(Flow.Subscription subscription) { @@ -667,7 +667,8 @@ public void onNext(StreamingEventKind item) { formatSse(seq.incrementAndGet(), payload).getBytes(StandardCharsets.UTF_8)); output.flush(); cancellation.requestNext(); - } catch (IOException error) { + } catch (Exception error) { + log.error("[SSE] Write failed for {}: {}", exchange.getRequestURI(), error.getMessage(), error); cancellation.release(); done.countDown(); } @@ -675,6 +676,7 @@ public void onNext(StreamingEventKind item) { @Override public void onError(Throwable error) { + log.error("[SSE] Stream failed for {}: {}", exchange.getRequestURI(), error.getMessage(), error); cancellation.release(); done.countDown(); } @@ -684,11 +686,15 @@ public void onComplete() { cancellation.release(); done.countDown(); } - }); + }; try { + publisher.subscribe(subscriber); done.await(); } catch (InterruptedException error) { Thread.currentThread().interrupt(); + } catch (RuntimeException error) { + // HTTP 200 is already committed; close the stream without attempting a second response. + log.error("[SSE] Subscription failed for {}: {}", exchange.getRequestURI(), error.getMessage(), error); } finally { cancellation.release(); output.close(); diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AStreamFailureTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AStreamFailureTest.java new file mode 100644 index 0000000..0392cdc --- /dev/null +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AStreamFailureTest.java @@ -0,0 +1,113 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * All Rights Reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed 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 dev.openan.workflow.engine.examples.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.sun.net.httpserver.HttpServer; +import java.net.InetSocketAddress; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Flow; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.spec.StreamingEventKind; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class JdkHttpA2AStreamFailureTest { + + @ParameterizedTest + @ValueSource(strings = { + "before-subscribe", "after-subscribe", "during-request", "on-error", "invalid-event"}) + void committedStreamFailureClosesWithoutEscapingToTheHttpErrorHandler(String phase) + throws Exception { + AtomicInteger subscriptionCancellations = new AtomicInteger(); + AtomicInteger eventConsumerCancellations = new AtomicInteger(); + AtomicReference escaped = new AtomicReference<>(); + CountDownLatch finished = new CountDownLatch(1); + ServerCallContext context = new ServerCallContext(null, Map.of(), Set.of(), "1.0"); + context.setEventConsumerCancelCallback(eventConsumerCancellations::incrementAndGet); + Flow.Publisher publisher = receiver -> { + if (phase.equals("before-subscribe")) { + throw new IllegalStateException("Subscribe failed before registration"); + } + receiver.onSubscribe(new Flow.Subscription() { + @Override + public void request(long count) { + if (phase.equals("during-request")) { + throw new IllegalStateException("Request failed"); + } + } + + @Override + public void cancel() { + subscriptionCancellations.incrementAndGet(); + } + }); + switch (phase) { + case "on-error" -> receiver.onError(new IllegalStateException("Publisher failed")); + case "invalid-event" -> receiver.onNext(null); + default -> throw new IllegalStateException("Subscribe failed after registration"); + } + }; + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/message:stream", exchange -> { + try { + JdkHttpA2AServer.writeStream(exchange, publisher, context); + } catch (Throwable error) { + // The production outer handler would attempt a second HTTP error response. + escaped.set(error); + } finally { + exchange.close(); + finished.countDown(); + } + }); + server.start(); + try { + HttpResponse response = HttpClient.newHttpClient().send( + HttpRequest.newBuilder(URI.create( + "http://127.0.0.1:" + server.getAddress().getPort() + "/message:stream")) + .timeout(Duration.ofSeconds(5)) + .POST(HttpRequest.BodyPublishers.noBody()) + .build(), + HttpResponse.BodyHandlers.ofString()); + + assertTrue(finished.await(5, TimeUnit.SECONDS)); + assertEquals(200, response.statusCode()); + assertEquals("text/event-stream", response.headers().firstValue("Content-Type").orElseThrow()); + assertEquals("", response.body()); + assertNull(escaped.get()); + assertEquals(1, eventConsumerCancellations.get()); + assertEquals(phase.equals("before-subscribe") ? 0 : 1, subscriptionCancellations.get()); + } finally { + server.stop(0); + } + } +} 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 d5ccc15..14909d9 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 @@ -176,9 +176,10 @@ private SseEmitter subscribeToEmitter( } } }; - SseEmitter emitter = track(new SseEmitter(0L), release); + SseEmitter emitter = new SseEmitter(0L); + Runnable finish = track(emitter, release); final AtomicLong seq = new AtomicLong(0); - publisher.subscribe( + Flow.Subscriber subscriber = new Flow.Subscriber<>() { @Override public void onSubscribe(Flow.Subscription s) { @@ -222,7 +223,14 @@ public void onComplete() { release.run(); emitter.complete(); } - }); + }; + try { + publisher.subscribe(subscriber); + } catch (RuntimeException | Error error) { + // MVC never receives this emitter when setup throws, so its callbacks cannot release it. + finish.run(); + throw error; + } return emitter; } @@ -331,7 +339,7 @@ public int activeStreamCount() { return activeStreams.get(); } - private T track(T emitter, Runnable cleanup) { + private Runnable track(ResponseBodyEmitter emitter, Runnable cleanup) { activeStreams.incrementAndGet(); AtomicBoolean completed = new AtomicBoolean(); Runnable finish = @@ -347,7 +355,7 @@ private T track(T emitter, Runnable cleanup) { emitter.onCompletion(finish); emitter.onTimeout(finish); emitter.onError(ignored -> finish.run()); - return emitter; + return finish; } private static ResponseEntity toResponse(RestHandler.HTTPRestResponse response) { 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 a9f57d0..aa030a4 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 @@ -166,6 +166,78 @@ public void cancel() { assertEquals(0, controller.activeStreamCount()); } + + @org.junit.jupiter.params.ParameterizedTest + @org.junit.jupiter.params.provider.ValueSource( + strings = {"before-subscribe", "after-subscribe", "during-request"}) + void synchronousPublisherFailureReleasesResourcesBeforeMvcDispatch(String phase) throws Exception { + for (String endpoint : new String[] {"/message:stream", "/tasks/task-1:subscribe"}) { + RestHandler restHandler = mock(RestHandler.class); + when(restHandler.createErrorResponse(any())) + .thenReturn(new RestHandler.HTTPRestResponse( + 500, "application/a2a+json", + "{\"error\":{\"code\":500,\"status\":\"INTERNAL\",\"message\":\"Setup failed\"}}")); + RequestHandler requestHandler = mock(RequestHandler.class); + AgentCard agentCard = mock(AgentCard.class); + when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(true).build()); + var agentInterface = mock(org.a2aproject.sdk.spec.AgentInterface.class); + when(agentInterface.protocolVersion()).thenReturn("1.0"); + when(agentCard.supportedInterfaces()).thenReturn(java.util.List.of(agentInterface)); + AtomicInteger subscriptionCancellations = new AtomicInteger(); + AtomicInteger eventConsumerCancellations = new AtomicInteger(); + AtomicReference> subscriber = new AtomicReference<>(); + org.mockito.stubbing.Answer> answer = + invocation -> { + ServerCallContext context = invocation.getArgument(1); + context.setEventConsumerCancelCallback(eventConsumerCancellations::incrementAndGet); + return receiver -> { + subscriber.set(receiver); + if (!phase.equals("before-subscribe")) { + receiver.onSubscribe(new Flow.Subscription() { + @Override + public void request(long count) { + if (phase.equals("during-request")) { + throw new IllegalStateException("Request failed"); + } + } + + @Override + public void cancel() { + subscriptionCancellations.incrementAndGet(); + } + }); + } + throw new IllegalStateException("Subscribe failed"); + }; + }; + when(requestHandler.onSubscribeToTask(any(), any())).thenAnswer(answer); + when(requestHandler.onMessageSendStream(any(), any())).thenAnswer(answer); + var controller = new A2AController(restHandler, requestHandler, agentCard); + MockMvc mockMvc = MockMvcBuilders.standaloneSetup(controller) + .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json") + .build(); + + mockMvc.perform(post("/a2a/json" + endpoint) + .header("A2A-Version", "1.0") + .contentType("application/json") + .accept("text/event-stream") + .content("{\"message\":{\"messageId\":\"m1\",\"role\":\"ROLE_USER\"," + + "\"parts\":[{\"text\":\"test\"}]}}")) + .andExpect(status().isInternalServerError()) + .andExpect(request().asyncNotStarted()) + .andExpect(content().contentTypeCompatibleWith("application/a2a+json")) + .andExpect(content().string(containsString("\"status\":\"INTERNAL\""))); + + assertEquals(0, controller.activeStreamCount()); + assertTrue(controller.awaitStreamsDrained(java.time.Duration.ofMillis(10))); + assertEquals(1, eventConsumerCancellations.get()); + assertEquals(phase.equals("before-subscribe") ? 0 : 1, subscriptionCancellations.get()); + subscriber.get().onComplete(); + assertEquals(0, controller.activeStreamCount()); + assertEquals(1, eventConsumerCancellations.get()); + } + } + @Test void nonStreamingA2AResponseUsesProtocolMediaType() throws Exception { RestHandler restHandler = mock(RestHandler.class); diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/package-info.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/package-info.java index 9bb54ac..884c50e 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/package-info.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/package-info.java @@ -23,5 +23,3 @@ * through the callback contracts in {@code control}. */ package dev.openan.workflow.engine; - -