diff --git a/CHANGELOG.md b/CHANGELOG.md index 3803f5d..a9b20b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,22 @@ # Changelog +## [Unreleased] + +- Keep Notification-T subscriptions open independently of task/result terminal states; maintain + stream activity tracking and explicit cancellation/resource ownership. +- Add independent registry HTTP authentication/TLS policy and orchestration account login for PSOP + loading. These APIs are current-source capabilities, not part of the published 0.1.1 artifact. +- Refresh bilingual examples, release boundaries and constructor/transport ownership guidance. + +## [0.1.1] — 2026-09-23 + +Released from tag `v0.1.1` (`952cde2`). Compared with 0.1.0: + +- Harden workflow joins, cancellation, failure propagation and sensitive-data logging. +- Keep active notification streams alive beyond the ordinary request deadline and release + streaming resources reliably. Later state-independent terminal-result handling is under Unreleased. +- Correct build-analysis exclusions and remove development-only vendor references from main. + ## [0.1.0] — 2026-09-18 - Send SSE heartbeat comments on streaming endpoints (`message:stream`, `tasks/{id}:subscribe`) at a configurable @@ -137,4 +154,4 @@ - Keep documentation host-neutral; sample class and AgentCard identifiers remain unchanged. Published Maven artifacts use the version selected by the release tag. The default source-build -`revision=0.0.9-SNAPSHOT` is not a Maven Central release version. +`revision=0.1.0-SNAPSHOT` is not a Maven Central release version. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c6f9dae..146563d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -24,9 +24,10 @@ workflow-engine-sdk-java/ | +-- src/main/java/dev/openan/workflow/engine/ | |-- client/ A2A transport, authentication, extensions, runtime APIs | |-- control/ User-facing: ControlPoint, EventCallback, EventType -| |-- core/ Internal: WorkflowExecutor, ContextBuilder (package-private) +| |-- core/ WorkflowExecutor (public), ContextBuilder (package-private) | |-- model/ Data models | |-- registry/ LoadPsop, RegistryClient +| |-- util/ Shared redaction and utility support | +-- runner/ ExecutePsop (entry point) |-- samples/ Demo applications |-- docs/ Documentation diff --git a/README.md b/README.md index 06586ac..ad51a12 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # A2A-T Workflow Execution Engine (Java) +> **Release boundary**: Maven Central currently publishes `0.1.1` (tag v0.1.1, 952cde2). This guide follows current source. Independent center HTTP authentication/TLS, PSOP account login and task-state-independent Notification-T handling were added after that tag and are unreleased. Build an exact reviewed source revision with a unique unpublished version for those capabilities; do not overwrite a released coordinate. Source revision=0.1.0-SNAPSHOT is a development coordinate, not a newer release. + [![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](LICENSE) [![Java](https://img.shields.io/badge/Java-17+-orange.svg)](https://adoptium.net/) [![Maven](https://img.shields.io/badge/Maven-3.6+-red.svg)](https://maven.apache.org/) @@ -32,13 +34,14 @@ never sent under an A2A-T URI as a fallback. The SDK is embedded in a host agent. It schedules workflow tasks to dispatched agents and returns complete results to the host callbacks. Task-T, Authorization-T, and Notification-T use independent transport/runtime/context instances. Authorization and Notification are independently triggered host-agent operations rather than DAG nodes; a notification -subscription remains open until the host-defined terminal event, explicit cancellation, or shutdown. +subscription is independent of task/result status. It remains open while transport activity continues; +explicit cancellation, host shutdown, stream exit/error, or the configured idle timeout ends it. ## Quick Start ### 1. Add Maven dependency -The current engine release is `0.1.0` on [Maven Central](https://repo.maven.apache.org/maven2/net/openan/workflow/sdk/). +The current engine release is `0.1.1` on [Maven Central](https://repo.maven.apache.org/maven2/net/openan/workflow/sdk/). A2A-T SDK `1.1.1` is published to Maven Central. Maven resolves it automatically; no SDK source checkout or local SDK build is required. The engine depends only on @@ -51,7 +54,7 @@ upgrade guidance. net.openan.workflow.sdk workflow-engine - 0.1.0 + 0.1.1 ``` @@ -61,7 +64,7 @@ For Spring Boot server-side integration: net.openan.workflow.sdk spring-boot-starter - 0.1.0 + 0.1.1 ``` @@ -134,6 +137,7 @@ graph TD root --> model root --> registry root --> runner + root --> util ``` | Package | Key Classes | Description | @@ -143,6 +147,7 @@ graph TD | `core` | `WorkflowExecutor`, `ContextBuilder` | DAG traversal, context assembly | | `model` | `Workflow`, `WorkflowStep`, `Task`, `TaskRequest`, `MessageContent`, `TaskResult`, `NegotiationRequest`, `NegotiationReply`, `ExecutionResult` | Data models | | `registry` | `LoadPsop`, `RegistryClient` | PSOP loading and AgentCard registry | +| `util` | `SensitiveDataRedactor` | Shared sensitive-data redaction | | `runner` | `ExecutePsop` | Entry point for workflow execution | > **Note:** `LlmHelper` lives in the **samples** module (`dev.openan.workflow.engine.examples`), not in the `client` diff --git a/SECURITY.md b/SECURITY.md index 19e6224..40f2125 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -4,8 +4,8 @@ | Version | Supported | |---------|--------------------| -| 1.0.x | :white_check_mark: | -| < 1.0 | :x: | +| Latest released 0.1.x | Report vulnerabilities against the latest release; fixes are evaluated on main | +| Older releases | Upgrade; no separate backport guarantee is documented | ## Reporting a Vulnerability diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index ec587ff..6baa668 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -154,6 +154,7 @@ Builder-based configuration for the workflow engine client. | `sendExecutorCoreSize` | `int` | `4` | Send executor core threads | | `sendExecutorMaxSize` | `int` | `16` | Send executor maximum threads | | `sendExecutorQueueCapacity` | `int` | `256` | Bounded send executor queue capacity | +| `preferredProtocol` | `String` | null | Preferred declared AgentCard protocol; use supported transport identifiers, not a vendor name. | | `authProvider` | `AuthProvider` | null | Custom auth provider | | `credentialsConfigPath` | `String` | null | Path to credentials JSON; explicit missing or malformed files fail startup | | `credentialEncryptionKey` | `String` | null | Host-supplied decryption key; never loaded from LLM .env | @@ -197,7 +198,7 @@ the same header name throw `SecurityException` instead of silently overwriting e Submit custom extensions via MessageContent(parts, metadata, extensions); no engine content-handler registration. A2atMessages.from(MetadataContent, List>) preserves SDK metadata and activates the corresponding URI; contextOf -(ReceivedMessage) or contextOf(Map) reads/checks canonical negotiation context. This adapter uses +(ReceivedMessage) is the public context reader. The Map overload is package-private and cannot be called by host code. This adapter uses a2a-t-core only, never content generation or semantic validation. ### A2AJavaClientRuntime @@ -213,6 +214,20 @@ public interface A2AJavaClientRuntime { Consumer eventSink, Consumer logSink); + // Optional task operations; default implementations throw UnsupportedOperationException. + default Task getTask(AgentCard card, String taskId, ClientCallContext context) { + throw new UnsupportedOperationException(); + } + default ListTasksResult listTasks(AgentCard card, ListTasksParams params, ClientCallContext context) { + throw new UnsupportedOperationException(); + } + default Task cancelTask(AgentCard card, String taskId, ClientCallContext context) { + throw new UnsupportedOperationException(); + } + default CompletableFuture subscribeToTask( + AgentCard card, String taskId, ClientCallContext context, Consumer sink) { + throw new UnsupportedOperationException(); + } void close(); } ``` diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index 04c7013..223c01b 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -1,6 +1,6 @@ # Business callback integration contract -Release `0.1.0` API; A2A-T SDK `1.1.1`, A2A Java `1.2.0.Final`. The engine owns DAG scheduling, standard A2A envelopes, +Current-source API baseline; A2A-T SDK `1.1.1`, A2A Java `1.2.0.Final`. The engine owns DAG scheduling, standard A2A envelopes, auth, transport and task waiting. The host owns content, schemas, templates, semantic validation and any LLM calls. The business contracts are independent of the transport runtime. @@ -19,7 +19,7 @@ onTask returns final parts/metadata/extensions; the engine sends them without ge onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No echo-success, implicit route approval or automatic consent. -See [Business callback contract](BUSINESS_CALLBACKS.md) for fields and working examples. +See the callback fields and examples below. ```java ControlPoint callbacks = ControlPoint.builder() @@ -220,7 +220,7 @@ sends; it does not automatically stop external business/LLM work. Hosts own canc callback entry points must return promptly; place blocking operations in an asynchronous executor. Missing/null/exceptional callbacks fail; uncertain sends are not blindly retried. -### 6.1 Standalone task +### 6.3 Standalone task Use `WorkflowEngineClient.sendTask(agentName, finalContent)` for a task outside the DAG. The overload with `NegotiationStrategy` supplies only that call's negotiation decisions. The host still owns content generation and @@ -234,8 +234,13 @@ ExtensionSender.openNotification(agentName, finalContent, (handle, received) -> Use host-generated Authorization-T or Notification-T content with the matching activated extension. The handle is registered before I/O; early callbacks can call handle.close() without capturing an unassigned variable. acknowledgement() reports actual ACK or failure; timeout is not a synthesized success. close() requests shutdown; -completion() completes after the stream actually exits. Retain it until the host-defined terminal event, explicit -cancellation, or host-agent shutdown, not just workflow completion. +completion() completes after the stream actually exits. Retain it across task/result terminal statuses and workflow +completion. Explicit cancellation, host-agent shutdown, stream exit/error, or the configured idle timeout ends it. + +The bundled SPN sample has a separate business policy: `WorkbenchExtensionLifecycle.handleNotification` explicitly +closes its handle after a validated recovery result. This is sample-controlled cancellation, not SDK termination on +Task-T/Notification-T result status. For a continuously subscribed production host, replace that sample callback and +retain the handle across recovery results until an explicit unsubscribe or shutdown. Task, authorization and notification use separate transports/runtimes/contexts. Neither extension success nor failure is a prerequisite for workflow execution. Authorization controls only the host-defined action associated with that policy. @@ -265,7 +270,7 @@ not the engine. Each role calls only the APIs needed for the message it owns. For natural-language inputs choose generateTaskPromptFromText(text, templateUri), generateAuthPromptFromText(text, templateUri), or generateNotificationPromptFromText(text, templateUri). Negotiation also offers phase-specific FromText and typed FromData alternatives. Structured generation may still call the SDK's configured LLM. Wrap the returned -MetadataContent with A2atMessages.from(generated, List.of (new TextPart("business summary"))). Preserve full metadata, +MetadataContent with A2atMessages.from(generated, List.of(new TextPart("business summary"))). Preserve full metadata, including templateUri and negotiationContext; provide nonempty parts. The parts summary is not the formal extension validation input. @@ -276,6 +281,7 @@ host disclosure denial produces SDK Reject. Authorization policy is a separate c Authorization and Notification generation APIs create requests, not arbitrary business results. Authorization receipts and notification events are dispatched-agent business outputs and must follow the selected template and application -contract. A subscription closes only on a host-defined terminal event, explicit cancellation, or shutdown; an ACK alone -is not a terminal business event. SDK rejection should preserve its safe code, slot errors, and extracted parameter keys. +contract. Task/result terminal statuses do not close a notification subscription; neither do ACKs. Explicit cancellation, +shutdown, stream exit/error, or the configured idle timeout ends it. SDK rejection should preserve its safe code, +slot errors, and extracted parameter keys. Authorization and subscription failures still do not block workflow execution. diff --git a/docs/en/DESIGN.md b/docs/en/DESIGN.md index b43ae6e..d8b5596 100644 --- a/docs/en/DESIGN.md +++ b/docs/en/DESIGN.md @@ -1,7 +1,7 @@ # A2A-T Engine SDK - Design > Architecture and design rationale for the A2A-T Workflow Execution Engine. -> This document describes the system as shipped in v1.0. It is written for +> This document describes the system as implemented by current source (see CHANGELOG for release boundaries). It is written for > engineers integrating or extending the SDK, not as a walkthrough of any > particular bug-fix history. @@ -363,14 +363,15 @@ sequenceDiagram ES->>A: Independent long-lived stream A-->>H: ACK via acknowledgement() A-->>H: listener(handle, ReceivedMessage) - H->>ES: handle.close() on terminal event/cancel/shutdown + Note over H,ES: Task/result terminal statuses do not close the notification stream + H->>ES: handle.close() on explicit cancellation/shutdown ES-->>H: completion() after stream exits ``` ## 9. Dependencies workflow-engine uses A2A Java `1.2.0.Final` (REST/JSON-RPC/gRPC), a matching gRPC runtime, -`net.openan.a2a-t.sdk:a2a-t-core:1.1.0`, Jackson and SLF4J. Pure engine consumers do not transitively receive A2A-T +`net.openan.a2a-t.sdk:a2a-t-core:1.1.1`, Jackson and SLF4J. Pure engine consumers do not transitively receive A2A-T client/server, LLM, prompt or resources. Host agents explicitly depend on a2a-t-client; dispatched-agent services that validate received content additionally use a2a-t-server. Registry/orchestration discovery is host-owned; RegistryClient/LoadPsop are optional helpers. diff --git a/docs/en/DEVELOPER_GUIDE.md b/docs/en/DEVELOPER_GUIDE.md index 9801d85..a77476d 100644 --- a/docs/en/DEVELOPER_GUIDE.md +++ b/docs/en/DEVELOPER_GUIDE.md @@ -1,5 +1,7 @@ # Developer Guide +> **Release boundary**: Maven Central currently publishes `0.1.1` (tag v0.1.1, 952cde2). This guide follows current source. Independent center HTTP authentication/TLS, PSOP account login and task-state-independent Notification-T handling were added after that tag and are unreleased. Build an exact reviewed source revision with a unique unpublished version for those capabilities; do not overwrite a released coordinate. Source revision=0.1.0-SNAPSHOT is a development coordinate, not a newer release. + This guide is for contributors and advanced users who want to understand the internal architecture, extend the SDK, or contribute patches. @@ -11,11 +13,11 @@ Add to your `pom.xml`: net.openan.workflow.sdk workflow-engine - 0.1.0 + 0.1.1 ``` -Version `0.1.0` is published to Maven Central. The repository currently builds `0.1.0-SNAPSHOT`. +Version `0.1.1` is published to Maven Central. The repository currently builds `0.1.0-SNAPSHOT`. The engine pulls in A2A protocol transports and a2a-t-core only. Host agents generating A2A-T content explicitly add a2a-t-client; dispatched-agent services that validate received content also add a2a-t-server. @@ -112,14 +114,15 @@ Events come from three layers: the runner (lifecycle bracket), the executor (ste ## 6. Mid-Level (Layer 1: WorkflowExecutor) ```java -try(var client = new DefaultWorkflowEngineClient(agentCards, a2aRuntime, - WorkflowEngineClientConfig.builder() - .sslVerify(false) - .credentialsConfigPath("etc/conf/agent_credentials.json") - .build())){ +WorkflowEngineClientConfig clientConfig = WorkflowEngineClientConfig.builder() + .sslVerify(true) + .credentialsConfigPath("etc/conf/agent_credentials.json") + .build(); +try (A2ATransport transport = new A2ATransport(agentCards, a2aRuntime, clientConfig); + DefaultWorkflowEngineClient client = new DefaultWorkflowEngineClient(transport, clientConfig)) { WorkflowExecutor executor = new WorkflowExecutor( workflow, - new MyControlPoint(), + controlPoint, client, new EventCallback(), "Analyze request", @@ -129,6 +132,8 @@ ExecutionResult result = executor.run().join(); } ``` +The transport owns the supplied runtime (or creates one when `a2aRuntime=null`). Use a dedicated runtime; do not pass a shared runtime owned elsewhere. The client facade here is non-owning and is closed before the transport. AgentCards, workflow and controlPoint are host inputs; supply a trusted CA/client identity as your deployment requires. + ### 6.1 Negotiation Auto-Loop When the host business allows negotiation for a task, call @@ -157,7 +162,7 @@ task success, even if the dispatched agent acknowledges it with COMPLETED. | Field | Where | Meaning | |-----------------------|-----------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `steps[].stepType` | `WorkflowStep` | `AllSuccess` (default): every subtask must succeed; `AnySuccess`: any subtask success suffices; `SelfLoop`: the host agent handles the step locally via `onSelfTask` (no A2A-T message to the named agent). | +| `steps[].stepType` | `WorkflowStep` | `ALL_SUCCESS` (default): every subtask must succeed; `ANY_SUCCESS`: any subtask success suffices; `SELF_LOOP`: the host agent handles the step locally via `onSelfTask` (no A2A-T message to the named agent). | | `steps[].subtasks[]` | `Task` | Each has `agent`, `skill`, `description`. One `onTask` (or `onSelfTask` for SelfLoop) call per subtask. | | `steps[].next[]` | `List` | Branch targets. `step` = next step name; `condition` = rule text. | | `steps[].layer` | `WorkflowStep` | Orchestration-level hint; actual readiness is derived from DAG predecessors. | @@ -326,3 +331,5 @@ Formatting changes presentation only (JSON whitespace and SSE display labels), n stays raw. Redaction and capacity limits still apply. To verify negotiation traffic, look for `A2A-Extensions: .../Negotiation-T/v1`, the matching metadata key and `negotiationContext`; there is no dedicated Negotiation-T HTTP header. + +Java enum constants are ALL_SUCCESS/ANY_SUCCESS/SELF_LOOP; their display/wire values are AllSuccess/AnySuccess/SelfLoop. StepType.fromValue accepts both forms. They are not extra step types. diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index 0cc5883..563205b 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -1,5 +1,7 @@ # A2A-T Workflow Execution Engine - Integration Guide +> **Release boundary**: Maven Central currently publishes `0.1.1` (tag v0.1.1, 952cde2). This guide follows current source. Independent center HTTP authentication/TLS, PSOP account login and task-state-independent Notification-T handling were added after that tag and are unreleased. Build an exact reviewed source revision with a unique unpublished version for those capabilities; do not overwrite a released coordinate. Source revision=0.1.0-SNAPSHOT is a development coordinate, not a newer release. + ## 1. Overview The A2A-T Workflow Execution Engine is a Java SDK for orchestrating multi-agent workflows using the A2A protocol with @@ -22,11 +24,11 @@ callbacks own A2A-T generation, semantic validation, schemas and any LLM calls. net.openan.workflow.sdk workflow-engine - 0.1.0 + 0.1.1 ``` -Version `0.1.0` is published to Maven Central. The repository currently builds `0.1.0-SNAPSHOT`. +Version `0.1.1` is published to Maven Central. The repository currently builds `0.1.0-SNAPSHOT`. ## 4. Quick Start @@ -447,7 +449,9 @@ NotificationSubscription openNotification(String agentName, MessageContent conte The host agent generates final Authorization-T/Notification-T content. Authorization and notification each use a transport/runtime/context independent of the task client and of one another. The listener receives the handle and -complete ReceivedMessage, and closes on the host-defined terminal event. acknowledgement() and completion() separately +complete ReceivedMessage. Task/result terminal statuses do not close the subscription; the host explicitly closes the +handle when cancelling the subscription or shutting down. Stream exit/error and the configured idle timeout also end +the subscription. acknowledgement() and completion() separately represent ACK and actual stream exit; neither is a workflow prerequisite. `heartbeat().lastEventAt()` is the last transport activity. A standard SSE comment heartbeat such as `: heartbeat` diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index e97c5ef..ffdfb6b 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -141,6 +141,7 @@ handle 再开始 I/O，监听器直接收到 handle 和完整 ReceivedMessage。 | `sendExecutorCoreSize` | `int` | `4` | 发送线程池核心线程数 | | `sendExecutorMaxSize` | `int` | `16` | 发送线程池最大线程数 | | `sendExecutorQueueCapacity` | `int` | `256` | 发送线程池有界队列容量 | +| `preferredProtocol` | `String` | null | 首选已声明的 AgentCard 协议，使用支持的传输标识，不是厂商名称。 | | `authProvider` | `AuthProvider` | null | 自定义认证提供器 | | `credentialsConfigPath` | `String` | null | 凭证 JSON 文件路径；显式配置缺失或损坏时启动失败 | | `credentialEncryptionKey` | `String` | null | 宿主显式传入凭据解密密钥，不从 LLM .env 加载 | @@ -181,8 +182,7 @@ Header 生成不同值，引擎会抛出 `SecurityException`，不会静默覆 ### A2atMessages 通过 MessageContent(parts, metadata, extensions) 提交自定义扩展内容，不注册引擎内容处理器。A2atMessages.from -(MetadataContent, List>) 保留 SDK metadata 原位置并激活对应 URI；contextOf(ReceivedMessage) 或 contextOf(Map< -String,Object>) 读取并检查规范协商上下文。只有 a2a-t-core 依赖，没有内容生成或语义校验。 +(MetadataContent, List>) 保留 SDK metadata 原位置并激活对应 URI；公开的 contextOf(ReceivedMessage) 读取并检查规范协商上下文，Map 重载为包内实现，宿主不能调用。只有 a2a-t-core 依赖，没有内容生成或语义校验。 ### A2AJavaClientRuntime @@ -196,6 +196,20 @@ public interface A2AJavaClientRuntime { Consumer eventSink, Consumer logSink); + // Optional task operations; default implementations throw UnsupportedOperationException. + default Task getTask(AgentCard card, String taskId, ClientCallContext context) { + throw new UnsupportedOperationException(); + } + default ListTasksResult listTasks(AgentCard card, ListTasksParams params, ClientCallContext context) { + throw new UnsupportedOperationException(); + } + default Task cancelTask(AgentCard card, String taskId, ClientCallContext context) { + throw new UnsupportedOperationException(); + } + default CompletableFuture subscribeToTask( + AgentCard card, String taskId, ClientCallContext context, Consumer sink) { + throw new UnsupportedOperationException(); + } void close(); } ``` diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index e5fac0c..1c96524 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -1,6 +1,6 @@ # 业务回调集成契约 -`0.1.0` API，A2A-T SDK `1.1.1`，A2A Java `1.2.0.Final`。 引擎负责 DAG、标准 A2A 信封、认证、传输和任务等待；宿主负责消息内容、schema、模板、语义校验和 +当前源码 API，A2A-T SDK `1.1.1`，A2A Java `1.2.0.Final`。 引擎负责 DAG、标准 A2A 信封、认证、传输和任务等待；宿主负责消息内容、schema、模板、语义校验和 LLM。业务契约独立于具体传输实现。 ## 1. 回调接口 @@ -17,7 +17,7 @@ interface ControlPoint { onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。onSelfTask 返回本地 TaskResult；onRoute 对一条条件边独立返回放行或拒绝；onNegotiation 返回 Send 或 Stop。无条件边不调用 onRoute 并始终放行。未实现的回调明确失败， 不回显成功、不默认放行条件边、不自动同意。 -字段与完整示例见 [业务回调集成契约](BUSINESS_CALLBACKS.md)。 +字段与完整示例见下文。 ```java ControlPoint callbacks = ControlPoint.builder() @@ -190,7 +190,7 @@ sendTimeoutSeconds=600；路由单独限制回调等待时间，任务交互／ 但不等于自动取消宿主正在运行的 LLM 或业务操作，宿主负责清理其资源。 同步回调入口应迅速返回，阻塞任务应交给异步执行器。 回调缺失、返回 null 或异常均明确失败；不确定发送失败不自动重发。 -### 6.1 独立任务 +### 6.3 独立任务 DAG 外任务使用 `WorkflowEngineClient.sendTask(agentName, finalContent)`。带 `NegotiationStrategy` 的重载只提供本次调用的 协商决策。内容生成和语义校验仍由宿主负责；引擎保持远端 task/context、等待终态，并在本地交互无法继续时清理已知的非终态任务。 @@ -201,7 +201,9 @@ ExtensionSender.sendAuthorization(agentName, finalContent) 返回 CompletableFut ExtensionSender.openNotification(agentName, finalContent, (handle, received) -> ...) 立即返回 handle。 传入宿主生成的最终授权／订阅内容并激活对应扩展。 I/O 前注册 handle，早到事件可直接 handle.close()，不用捕获尚未赋值的外部变量。 acknowledgement() 代表真实 ACK 或失败；超时不伪造成功。 close() 请求关闭，completion() 在流实际退出后完成。 -订阅保持到宿主定义的终态事件、显式取消或宿主退出，不随单次工作流结束自动关闭。 +任务或通知结果终态不关闭订阅，单次工作流结束也不关闭订阅。有传输活动时持续保持；显式取消、宿主退出、流退出或异常、配置的空闲超时才结束订阅。 + +随附 SPN 样例另有业务策略：`WorkbenchExtensionLifecycle.handleNotification` 在校验完整抢通结果后显式关闭自己的 handle。这是样例业务主动取消，不是 SDK 随 Task-T/Notification-T 结果状态自动终止。生产宿主需要持续订阅时，应替换该样例回调，跨抢通结果保留 handle，直到显式取消订阅或宿主退出。 任务、授权、通知使用独立 transport/runtime/context。授权和订阅成功与否均不作为工作流执行前提；授权策略只控制与该策略关联的宿主自定义业务操作。 @@ -228,10 +230,10 @@ acknowledgement() 代表真实 ACK 或失败；超时不伪造成功。 close() FromText 和强类型 FromData。 这些是输入方式的选择，不是每条消息必须全部执行的流水线。FromDataWithSchema 也可能调用 SDK 配置的 LLM， 不能等同于完全不调用模型。 -无论选择哪一种生成入口，都保留返回 MetadataContent 的完整 metadata，再通过 A2atMessages.from(generated, List.of (new +无论选择哪一种生成入口，都保留返回 MetadataContent 的完整 metadata，再通过 A2atMessages.from(generated, List.of(new TextPart("业务摘要"))) 转成引擎 MessageContent。 parts 必须有合法内容；摘要不是正式扩展正文，不能拿摘要替代校验输入。不要只复制 promptText 而丢失 templateUri、negotiationContext 或扩展 URI。 回复必须匹配 A2A taskId/contextId 及协商 id/round/maxRounds，正式正文从 Negotiation-T metadata 读取。 一次有效回复只消费一次。正文错误、字段不完整、任务不匹配或跨协商的回复不得执行业务。无法核实权威信息时生成 SDK Abort，不编造数据；宿主披露策略拒绝时生成 SDK Reject。授权策略是独立关注点。 -Authorization-T 和 Notification-T 的生成接口用于生成请求，不用于包装任意业务结果。授权回执与通知事件是被调度智能体的业务输出，必须遵循选定模板和应用契约。订阅只在宿主定义的终态事件、显式取消或关闭时结束；ACK 本身不是终态业务事件。SDK 拒绝应保留安全的错误码、slot 错误与已提取参数键。授权或订阅失败仍不阻断工作流执行。 +Authorization-T 和 Notification-T 的生成接口用于生成请求，不用于包装任意业务结果。授权回执与通知事件是被调度智能体的业务输出，必须遵循选定模板和应用契约。任务或通知结果终态、ACK 均不关闭通知订阅；显式取消、关闭、流退出或异常、配置的空闲超时才结束订阅。SDK 拒绝应保留安全的错误码、slot 错误与已提取参数键。授权或订阅失败仍不阻断工作流执行。 diff --git a/docs/zh/DESIGN.md b/docs/zh/DESIGN.md index 5c107a2..cf61e17 100644 --- a/docs/zh/DESIGN.md +++ b/docs/zh/DESIGN.md @@ -1,7 +1,7 @@ # A2A-T 工作流执行引擎 - 架构设计 > A2A-T 工作流执行引擎的架构设计与设计原理。 -> 本文档描述 v1.0 发布版本。面向集成或扩展 SDK 的工程师 +> 本文档描述当前源码实现（发布边界见 CHANGELOG），面向集成或扩展 SDK 的工程师。 --- @@ -324,14 +324,15 @@ sequenceDiagram ES->>A: Independent long-lived stream A-->>H: ACK via acknowledgement() A-->>H: listener(handle, ReceivedMessage) - H->>ES: 终态事件 / 取消 / 停机时 handle.close() + Note over H,ES: 任务或通知结果终态不关闭通知流 + H->>ES: 显式取消订阅 / 停机时 handle.close() ES-->>H: completion() after stream exits ``` ## 9. 依赖 workflow-engine：A2A Java `1.2.0.Final`（REST/JSON-RPC/gRPC）、匹配的 gRPC runtime、 -`net.openan.a2a-t.sdk:a2a-t-core:1.1.0`、Jackson、SLF4J。 纯引擎消费者不会传递引入 A2A-T client/server、LLM、prompt 或 +`net.openan.a2a-t.sdk:a2a-t-core:1.1.1`、Jackson、SLF4J。 纯引擎消费者不会传递引入 A2A-T client/server、LLM、prompt 或 resources。宿主智能体显式依赖 a2a-t-client，校验接收内容的被调度智能体服务另依赖 a2a-t-server。注册中心和编排中心由宿主调用，可选择 RegistryClient/LoadPsop 辅助接口或自己的实现。 模板和 slot schema 来自锁定 SDK jar，样例不覆盖同名资源。 diff --git a/docs/zh/DEVELOPER_GUIDE.md b/docs/zh/DEVELOPER_GUIDE.md index ac7b4e1..f5ac328 100644 --- a/docs/zh/DEVELOPER_GUIDE.md +++ b/docs/zh/DEVELOPER_GUIDE.md @@ -1,5 +1,7 @@ # 开发者指南 +> **发布边界**：Maven Central 最新发布为 `0.1.1`（标签 v0.1.1，952cde2）。本文按当前源码说明；中心独立 HTTP 认证/TLS、PSOP 账号登录及通知结果状态无关处理是在该标签之后加入，尚未发布。使用这些能力需构建精确检视源码的独立版本，不能覆盖已发布坐标。源码 revision=0.1.0-SNAPSHOT 是开发坐标，不是更新的正式版本。 + 本指南面向贡献者和希望理解内部架构、扩展 SDK 或提交补丁的高级用户。 ## 1. 安装 @@ -10,11 +12,11 @@ net.openan.workflow.sdk workflow-engine - 0.1.0 + 0.1.1 ``` -`0.1.0` 已发布到 Maven Central；当前仓库构建版本为 `0.1.0-SNAPSHOT`。 +`0.1.1` 已发布到 Maven Central；当前仓库构建版本为 `0.1.0-SNAPSHOT`。 引擎会传递性引入 A2A 协议 SDK（`a2a-java-sdk-client`，含 REST、JSON-RPC、gRPC 传输） 和最小 A2A-T 核心（`a2a-t-core`）。生成 A2A-T 内容的宿主智能体显式依赖 a2a-t-client，校验接收内容的被调度智能体服务另引入 a2a-t-server。 @@ -108,14 +110,15 @@ ExecutionResult result = ExecutePsop.builder() ## 6. 中间层（Layer 1: WorkflowExecutor） ```java -try (var client = new DefaultWorkflowEngineClient(agentCards, a2aRuntime, - WorkflowEngineClientConfig.builder() - .sslVerify(false) - .credentialsConfigPath("etc/conf/agent_credentials.json") - .build())) { +WorkflowEngineClientConfig clientConfig = WorkflowEngineClientConfig.builder() + .sslVerify(true) + .credentialsConfigPath("etc/conf/agent_credentials.json") + .build(); +try (A2ATransport transport = new A2ATransport(agentCards, a2aRuntime, clientConfig); + DefaultWorkflowEngineClient client = new DefaultWorkflowEngineClient(transport, clientConfig)) { WorkflowExecutor executor = new WorkflowExecutor( workflow, - new MyControlPoint(), + controlPoint, client, new EventCallback(), "分析请求", @@ -125,6 +128,8 @@ try (var client = new DefaultWorkflowEngineClient(agentCards, a2aRuntime, } ``` +transport 接管传入 runtime（`a2aRuntime=null` 时创建默认 runtime）。使用专用 runtime，不传入其他组件拥有的共享实例；示例中的非 owning client 先关闭，再关闭 transport。AgentCards、workflow 和 controlPoint 由宿主提供，CA 与客户端证书按部署要求配置。 + ### 6.1 协商自动循环 宿主业务需要允许当前任务协商时，在首轮 Task-T `MessageContent` 上调用 @@ -148,7 +153,7 @@ COMPLETED 确认，也不能判为任务成功。 | 字段 | 位置 | 含义 | |-----------------------|-----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `steps[].stepType` | `WorkflowStep` | `AllSuccess`（默认）：所有子任务必须成功；`AnySuccess`：任一成功即可；`SelfLoop`：宿主智能体通过 `onSelfTask` 本地处理（不向命名智能体发送 A2A-T 消息）。 | +| `steps[].stepType` | `WorkflowStep` | `ALL_SUCCESS`（默认）：所有子任务必须成功；`ANY_SUCCESS`：任一成功即可；`SELF_LOOP`：宿主智能体通过 `onSelfTask` 本地处理（不向命名智能体发送 A2A-T 消息）。 | | `steps[].subtasks[]` | `Task` | 每个含 `agent`、`skill`、`description`。每个子任务触发一次 `onTask`（或 SelfLoop 的 `onSelfTask`）调用。 | | `steps[].next[]` | `List` | 分支目标。`step` = 下一步名称；`condition` = 规则文本。 | | `steps[].layer` | `WorkflowStep` | 编排层级提示；实际就绪条件由 DAG 前驱关系决定。 | @@ -308,3 +313,5 @@ SSE 数据区标记），不改实际发送字节、metadata、数值或扩展 字符串中的 `\\n` 保留转义，避免把业务字符串误写成非法 JSON；非 JSON／不完整正文保持原样。日志仍执行脱敏和容量限制。验证协商流量时， 检查 `A2A-Extensions: .../Negotiation-T/v1`；不存在专用的 Negotiation-T HTTP 头。正文检查对应 URI 的 metadata 和 `negotiationContext`，不要只检索大写 `NEGOTIATION-T`。 + +Java 枚举常量为 ALL_SUCCESS/ANY_SUCCESS/SELF_LOOP；显示/协议值为 AllSuccess/AnySuccess/SelfLoop。StepType.fromValue 接受两种形式，不代表另有三类步骤。 diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index 02046bd..1d46057 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -1,5 +1,7 @@ # A2A-T 工作流执行引擎 - 二次开发集成指南 +> **发布边界**：Maven Central 最新发布为 `0.1.1`（标签 v0.1.1，952cde2）。本文按当前源码说明；中心独立 HTTP 认证/TLS、PSOP 账号登录及通知结果状态无关处理是在该标签之后加入，尚未发布。使用这些能力需构建精确检视源码的独立版本，不能覆盖已发布坐标。源码 revision=0.1.0-SNAPSHOT 是开发坐标，不是更新的正式版本。 + ## 1. 概述 A2A-T 工作流执行引擎是一个 Java SDK，用于基于 A2A 协议和 A2A-T 电信扩展编排多智能体工作流。 @@ -20,11 +22,11 @@ A2A-T 工作流执行引擎是一个 Java SDK，用于基于 A2A 协议和 A2A-T net.openan.workflow.sdk workflow-engine - 0.1.0 + 0.1.1 ``` -`0.1.0` 已发布到 Maven Central；当前仓库构建版本为 `0.1.0-SNAPSHOT`。 +`0.1.1` 已发布到 Maven Central；当前仓库构建版本为 `0.1.0-SNAPSHOT`。 ## 4. 快速上手 @@ -258,7 +260,7 @@ public interface AuthProvider { } ``` -每次发消息前都会调用 `applyAuth`，实现方往 `headers` 里塞认证头即可。 +每次发消息前都会调用 `applyAuth`，实现方在 `headers` 中设置所需认证头。 **场景 1：企业 SSO / 外部 Token 服务** @@ -420,7 +422,7 @@ NotificationSubscription openNotification(String agentName, MessageContent conte ``` 宿主智能体生成最终 Authorization-T/Notification-T 内容。授权与通知分别使用独立于任务客户端且彼此隔离的 -transport/runtime/context。订阅监听器收到 handle 与完整 ReceivedMessage，在宿主定义的终态事件上关闭。 +transport/runtime/context。订阅监听器收到 handle 与完整 ReceivedMessage。任务或通知结果终态不关闭订阅；宿主取消订阅或退出时显式关闭 handle，流退出或异常、配置的空闲超时也会结束订阅。 handle.acknowledgement() 和 completion() 分别表示 ACK 和真实流退出，两者都不是工作流前提。 `heartbeat().lastEventAt()` 表示最后一次传输活动；标准 SSE comment 心跳（如 `: heartbeat`）会刷新该时间， 但不会增加只统计已解码 diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/DocumentedCallbackExampleTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/DocumentedCallbackExampleTest.java index 7093fbf..a5c3a6c 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/DocumentedCallbackExampleTest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/DocumentedCallbackExampleTest.java @@ -35,6 +35,22 @@ private static String block(String doc, String marker) { return doc.substring(begin, end); } + @Test + void publicNotificationGuidesDoNotCloseOnResultStatus() throws Exception { + Path root = Path.of("").toAbsolutePath(); + if (!Files.isDirectory(root.resolve("docs"))) root = root.getParent(); + for (String language : List.of("en", "zh")) { + for (String guide : List.of("BUSINESS_CALLBACKS", "INTEGRATION_GUIDE", "DESIGN")) { + String doc = Files.readString(root.resolve("docs/" + language + "/" + guide + ".md")); + assertFalse(doc.contains("host-defined terminal event"), guide); + assertFalse(doc.contains("宿主定义的终态事件"), guide); + assertTrue( + doc.contains(language.equals("en") ? "terminal statuses do not close" : "终态不关闭"), + guide); + } + } + } + @Test void bilingualCallbackAndSdkExamplesCompileAgainstCurrentPublicApi() throws Exception { Path root = Path.of("").toAbsolutePath(); @@ -45,24 +61,29 @@ void bilingualCallbackAndSdkExamplesCompileAgainstCurrentPublicApi() throws Exce String generated = block(doc, "MetadataContent generated ="); String callbackSource = """ - import java.util.*; - import java.util.concurrent.*; - import dev.openan.workflow.engine.control.*; - import dev.openan.workflow.engine.model.*; - import dev.openan.workflow.engine.client.*; - import net.openan.a2at.sdk.client.A2ATClient; - import net.openan.a2at.sdk.core.model.*; - import org.a2aproject.sdk.spec.TextPart; - class DocumentedCallbacks { - ControlPoint callbacks() { - """ + import java.util.*; + import java.util.concurrent.*; + import dev.openan.workflow.engine.control.*; + import dev.openan.workflow.engine.model.*; + import dev.openan.workflow.engine.client.*; + import net.openan.a2at.sdk.client.A2ATClient; + import net.openan.a2at.sdk.core.model.*; + import org.a2aproject.sdk.spec.TextPart; + class DocumentedCallbacks { + ControlPoint callbacks() { + """ + callbacks - + "\nreturn callbacks;\n}\n" - + "MessageContent generate(A2ATClient sdk, Map data, Map schema) {\n" + + "\n" + + "return callbacks;\n" + + "}\n" + + "MessageContent generate(A2ATClient sdk, Map data," + + " Map schema) {\n" + generated + "\nreturn outgoing;\n}\n}"; - String integration = Files.readString(root.resolve("docs/" + language + "/INTEGRATION_GUIDE.md")); - String quickStartSource = """ + String integration = + Files.readString(root.resolve("docs/" + language + "/INTEGRATION_GUIDE.md")); + String quickStartSource = + """ import java.util.*; import java.util.concurrent.*; import com.fasterxml.jackson.databind.ObjectMapper; @@ -75,69 +96,97 @@ ControlPoint callbacks() { class DocumentedIntegration { void run() throws Exception { """ - + block(integration, "Workflow workflow =") + "\n" - + block(integration, "HttpAuthProvider registryAuth =") + "\n" - + block(integration, "ControlPoint callbacks =") + "\n" - + block(integration, "CompletableFuture execution =") + "\n}}"; - String centerSource = """ + + block(integration, "Workflow workflow =") + + "\n" + + block(integration, "HttpAuthProvider registryAuth =") + + "\n" + + block(integration, "ControlPoint callbacks =") + + "\n" + + block(integration, "CompletableFuture execution =") + + "\n}}"; + String centerSource = + """ import java.util.*; import dev.openan.workflow.engine.registry.*; import dev.openan.workflow.engine.model.*; class DocumentedCenters { void run() throws Exception { """ - + block(integration, "TlsPolicy orchTls =") + "\n}}"; - String development = Files.readString(root.resolve("docs/" + language + "/DEVELOPER_GUIDE.md")); + + block(integration, "TlsPolicy orchTls =") + + "\n}}"; + String development = + Files.readString(root.resolve("docs/" + language + "/DEVELOPER_GUIDE.md")); assertTrue(development.contains("orchAuth, orchTls")); assertFalse(development.contains("psopId, token, false")); assertTrue(development.contains("subscribeOn(Schedulers.boundedElastic())")); assertTrue(development.contains("produces = MediaType.TEXT_EVENT_STREAM_VALUE")); String developmentSource = - "import dev.openan.workflow.engine.registry.*; import dev.openan.workflow.engine.model.*; " - + "class DocumentedDevelopment { String baseUrl = \"https://orchestration.example\";\n" - + block(development, "private final TlsPolicy orchTls =") + "\n" + "import dev.openan.workflow.engine.registry.*; import dev.openan.workflow.engine.model.*;" + + " class DocumentedDevelopment { String baseUrl =" + + " \"https://orchestration.example\";\n" + + block(development, "private final TlsPolicy orchTls =") + + "\n" + "Workflow load(String psopId) throws Exception { " + "return LoadPsop.load(baseUrl, psopId, orchAuth, orchTls); }}"; - for (String source : List.of(callbackSource, quickStartSource, centerSource, developmentSource)) { - JavaCompiler compiler = ToolProvider.getSystemJavaCompiler(); - assertNotNull(compiler, "Tests require a JDK, not a JRE"); - DiagnosticCollector diagnostics = new DiagnosticCollector<>(); - JavaFileObject input = - new SimpleJavaFileObject( - URI.create("string:///DocumentedCallbacks.java"), JavaFileObject.Kind.SOURCE) { - @Override - public CharSequence getCharContent(boolean ignoreEncodingErrors) { - return source; - } - }; - try (StandardJavaFileManager standard = - compiler.getStandardFileManager(diagnostics, null, null); - var output = - new ForwardingJavaFileManager(standard) { - @Override - public JavaFileObject getJavaFileForOutput( - Location location, String name, JavaFileObject.Kind kind, FileObject sibling) { - return new SimpleJavaFileObject( - URI.create("bytes:///" + name + kind.extension), kind) { - @Override - public java.io.OutputStream openOutputStream() { - return new java.io.ByteArrayOutputStream(); - } - }; - } - }) { - boolean ok = - compiler - .getTask( - null, - output, - diagnostics, - List.of("-proc:none", "-classpath", System.getProperty("java.class.path")), - null, - List.of(input)) - .call(); - assertTrue(ok, () -> language + ": " + diagnostics.getDiagnostics()); - } + String midLevelSource = + """ + import java.util.*; + import org.a2aproject.sdk.spec.AgentCard; + import dev.openan.workflow.engine.client.*; + import dev.openan.workflow.engine.control.*; + import dev.openan.workflow.engine.core.WorkflowExecutor; + import dev.openan.workflow.engine.model.*; + class DocumentedMidLevel { + void run(List agentCards, A2AJavaClientRuntime a2aRuntime, + Workflow workflow, ControlPoint controlPoint) { + """ + + block(development, "WorkflowEngineClientConfig clientConfig =") + + "\n}}"; + for (String source : + List.of( + callbackSource, quickStartSource, centerSource, developmentSource, midLevelSource)) { + JavaCompiler compiler = ToolProvider.getSystemJavaCompiler(); + assertNotNull(compiler, "Tests require a JDK, not a JRE"); + DiagnosticCollector diagnostics = new DiagnosticCollector<>(); + JavaFileObject input = + new SimpleJavaFileObject( + URI.create("string:///DocumentedCallbacks.java"), JavaFileObject.Kind.SOURCE) { + @Override + public CharSequence getCharContent(boolean ignoreEncodingErrors) { + return source; + } + }; + try (StandardJavaFileManager standard = + compiler.getStandardFileManager(diagnostics, null, null); + var output = + new ForwardingJavaFileManager(standard) { + @Override + public JavaFileObject getJavaFileForOutput( + Location location, + String name, + JavaFileObject.Kind kind, + FileObject sibling) { + return new SimpleJavaFileObject( + URI.create("bytes:///" + name + kind.extension), kind) { + @Override + public java.io.OutputStream openOutputStream() { + return new java.io.ByteArrayOutputStream(); + } + }; + } + }) { + boolean ok = + compiler + .getTask( + null, + output, + diagnostics, + List.of("-proc:none", "-classpath", System.getProperty("java.class.path")), + null, + List.of(input)) + .call(); + assertTrue(ok, () -> language + ": " + diagnostics.getDiagnostics()); + } } } }