diff --git a/CHANGELOG.md b/CHANGELOG.md index 65923e6d..893368fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,19 @@ # Changelog -## [Unreleased] +## [0.0.9] — 2026-09-14 + +- Make Negotiation-T activation host-owned. The business activates Negotiation-T per send via the new + `MessageContent.withExtension(...)`; the engine mirrors that selection into `message.extensions` and the + `A2A-Extensions` request header and validates that the target AgentCard declares the extension before dispatch. + An unsolicited negotiation (host did not activate Negotiation-T) fails fast instead of silently entering the + auto-loop. +- Accept Negotiation-T Propose carried on a taskless bare message (A2A-T pre-task negotiation, where the remote + proposes before creating any task): `onNegotiation` now also fires for a valid Propose without an + `INPUT_REQUIRED` task status. The exchange is correlated by contextId and negotiation id, and the follow-up send + carries no taskId. Invalid negotiation metadata on a bare message now fails explicitly instead of being silently + treated as a normal successful response. +- Converge non-functional formatting and javadoc drift between `main` and `dev` so the branches differ only in the + Eastcom platform integration. ## [0.0.8] — 2026-09-13 diff --git a/README.md b/README.md index 2511aa51..4792cfbf 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ return final message content and own any A2A-T generation, schema, validation or ## Features -- **A2A-T Extension Support**: Task-T (structured task prompts), Negotiation-T (stateless auto negotiation loop), +- **A2A-T Extension Support**: Task-T (structured task prompts), Negotiation-T (host-activated auto negotiation loop), Authorization-T (independent authorization operation), Notification-T (independent long-lived SSE subscription) - **Content-neutral callbacks**: final MessageContent, complete ReceivedMessage, local multi-output TaskResult and explicit NegotiationReply.Send/Stop @@ -38,7 +38,7 @@ subscription remains open until the host-defined terminal event, explicit cancel ### 1. Add Maven dependency -The current engine release is `0.0.8` on [Maven Central](https://repo.maven.apache.org/maven2/net/openan/workflow/sdk/). +The current engine release is `0.0.9` 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 @@ -51,7 +51,7 @@ upgrade guidance. net.openan.workflow.sdk workflow-engine - 0.0.8 + 0.0.9 ``` @@ -61,7 +61,7 @@ For Spring Boot server-side integration: net.openan.workflow.sdk spring-boot-starter - 0.0.8 + 0.0.9 ``` diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index 451dd3f3..7486c16b 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -22,21 +22,21 @@ Entry point for executing a PSOP workflow. Uses the Builder pattern. #### ExecutePsop.Builder -| Method | Type | Default | Description | -|------------------------------------------|----------|-------------|----------------------------------------------| -| `psop(Workflow)` | required | - | PSOP workflow definition | -| `agentCards(List)` | optional | `List.of()` | Cards for dispatched agents; required for remote steps unless a configured `engineClient` supplies transport | -| `controlPoint(ControlPoint)` | required | - | User decision implementation | +| Method | Type | Default | Description | +|------------------------------------------|----------|-------------|-----------------------------------------------------------------------------------------------------------------------------------| +| `psop(Workflow)` | required | - | PSOP workflow definition | +| `agentCards(List)` | optional | `List.of()` | Cards for dispatched agents; required for remote steps unless a configured `engineClient` supplies transport | +| `controlPoint(ControlPoint)` | required | - | User decision implementation | | `engineClient(WorkflowEngineClient)` | optional | null | Pre-configured client (null = auto-create); cannot be combined with AgentCard, runtime, TLS, or credentials construction settings | -| `runtimeIntent(String)` | optional | `""` | Natural-language intent for context assembly | -| `lang(String)` | optional | `"zh"` | Language hint (`"zh"` or `"en"`) | -| `credentialsConfigPath(String)` | optional | null | Path to credentials JSON file | -| `sslVerify(boolean)` | optional | `true` | Whether to verify TLS certificates | -| `caCertsPath(String)` | optional | null | Path to CA certificates PEM file | -| `a2aClientRuntime(A2AJavaClientRuntime)` | optional | null | Custom runtime (null = auto-create) | -| `eventCallback(EventCallback)` | optional | null | Real-time event callback | -| `onFinish(BiConsumer)` | optional | null | Called when execution completes | -| `onEvent(Function)` | optional | null | Per-event transformation hook | +| `runtimeIntent(String)` | optional | `""` | Natural-language intent for context assembly | +| `lang(String)` | optional | `"zh"` | Language hint (`"zh"` or `"en"`) | +| `credentialsConfigPath(String)` | optional | null | Path to credentials JSON file | +| `sslVerify(boolean)` | optional | `true` | Whether to verify TLS certificates | +| `caCertsPath(String)` | optional | null | Path to CA certificates PEM file | +| `a2aClientRuntime(A2AJavaClientRuntime)` | optional | null | Custom runtime (null = auto-create) | +| `eventCallback(EventCallback)` | optional | null | Real-time event callback | +| `onFinish(BiConsumer)` | optional | null | Called when execution completes | +| `onEvent(Function)` | optional | null | Per-event transformation hook | ```java ExecutionResult result = ExecutePsop.builder() @@ -76,28 +76,33 @@ 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 Future returned by `subscribeToTask` represents the subscription result, not an independently -closeable SSE handle. Cancelling it does not guarantee transport termination; close the owning Client/Runtime -to release a non-terminal subscription. +`nextPageToken`; cancellation is not implied by closing a local client or notification stream. The Future returned by +`subscribeToTask` represents the subscription result, not an independently closeable SSE handle. Cancelling it does not +guarantee transport termination; close the owning Client/Runtime to release a non-terminal subscription. -`sendTask` executes final host-provided content outside a DAG. Every call has a fresh context and -uses the same send, wait, task-query and Negotiation-T loop as workflow execution. The overload takes -a per-call strategy without changing the client-wide ControlPoint. If a known remote task remains -non-final after a timeout, missing handler or other local interaction failure, the client attempts -to cancel it before completing exceptionally. `sendTask` is the only public task-submission name; -protocol-runtime implementations may still use the A2A operation name `sendMessage` internally. +`sendTask` executes final host-provided content outside a DAG. Every call has a fresh context and uses the same send, +wait, task-query and Negotiation-T loop as workflow execution. The overload takes a per-call strategy without changing +the client-wide ControlPoint. If a known remote task remains non-final after a timeout, missing handler or other local +interaction failure, the client attempts to cancel it before completing exceptionally. `sendTask` is the only public +task-submission name; protocol-runtime implementations may still use the A2A operation name `sendMessage` internally. -Plain A2A content does not require Task-T. When Task-T is activated, its metadata must be present and -the target AgentCard must declare Task-T support. +Plain A2A content does not require Task-T. When Task-T is activated, its metadata must be present and the target +AgentCard must declare Task-T support. After `onTask` returns, the executor calls `sendTask` internally; `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. -Only a remote `INPUT_REQUIRED` carrying valid Negotiation-T Propose enters `onNegotiation`. Terminal responses never -restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. The host validates/interprets the proposal and +Declaring Negotiation-T in an AgentCard does not activate it. When the host business allows negotiation for the current +task, call `withExtension(A2ATExtension.NEGOTIATION_T.uri())` on the initial Task-T `MessageContent`; the engine mirrors +that selection into the message extensions and the `A2A-Extensions` request header. `onNegotiation` and +`NegotiationStrategy` only handle a received Propose and do not modify the initial extension set. + +Only a remote `INPUT_REQUIRED` or a taskless bare message carrying a valid Negotiation-T Propose enters +`onNegotiation` (the latter is the A2A-T pre-task mode: correlated by contextId and negotiation id, with taskId-less +follow-up sends). Terminal responses never restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. Invalid +negotiation metadata on a bare message also fails explicitly instead of being treated as a normal response. The host validates/interprets the proposal and generates the final Accept/Reject/Abort with its own A2A-T client. Use `A2atMessages.contextOf(request.received())` to obtain the received context; reply with the same id, round and maxRounds. The last allowed round can still be answered. Do not call nextRound for an ending reply or return a new Propose. @@ -122,14 +127,14 @@ NotificationSubscription openNotification(String agentName, MessageContent conte Authorization and notification accept host-generated final content. Use separate transport/runtime/context instances; their outcomes do not gate the workflow. openNotification registers a handle before I/O, and passes it plus ReceivedMessage directly to the listener. acknowledgement() is the real ACK; timeout fails. close() requests closure; -completion() observes actual stream termination. -For compatibility, `heartbeat().lastEventAt()` keeps its old name but represents the last transport activity. -A standard SSE `: heartbeat` refreshes it without increasing `eventCount`, which counts decoded A2A business events only. +completion() observes actual stream termination. For compatibility, `heartbeat().lastEventAt()` keeps its old name but +represents the last transport activity. A standard SSE `: heartbeat` refreshes it without increasing `eventCount`, which +counts decoded A2A business events only. `lastBusinessEventAt()` returns the most recent business-event time. `new DefaultExtensionSender(transport)` owns and closes the transport. Use -`DefaultExtensionSender.nonOwning(transport)` only when another component owns the transport, and then close subscriptions -and the transport explicitly. +`DefaultExtensionSender.nonOwning(transport)` only when another component owns the transport, and then close +subscriptions and the transport explicitly. ### WorkflowEngineClientConfig @@ -154,8 +159,8 @@ Builder-based configuration for the workflow engine client. | `credentialsConfig` | `Map` | null | Inline credentials config; must match AgentCard security requirements | | `maxNegotiationExchanges` | `int` | `3` | Local interaction budget, independent of SDK maxRounds | -`credentialsConfigPath` and `credentialsConfig` are mutually exclusive built-in credential sources; -configuring both fails when the configuration is built. +`credentialsConfigPath` and `credentialsConfig` are mutually exclusive built-in credential sources; configuring both +fails when the configuration is built. ```java WorkflowEngineClientConfig config = WorkflowEngineClientConfig.builder() @@ -289,19 +294,19 @@ Map normalized = AgentCardNormalizer.normalize(rawMap); The following public types support advanced integrations. Prefer the higher-level interfaces above unless a custom runtime, diagnostic adapter, or explicit lifecycle requires them. -| Type | Purpose | -|-------------------------------|---------| -| `A2ATExtension` | Canonical extension names and URIs | -| `A2ATransport` | Low-level transport, authentication, response assembly, and subscription lifecycle | -| `DefaultWorkflowEngineClient` | Default `WorkflowEngineClient` implementation | +| Type | Purpose | +|-------------------------------|--------------------------------------------------------------------------------------------------| +| `A2ATExtension` | Canonical extension names and URIs | +| `A2ATransport` | Low-level transport, authentication, response assembly, and subscription lifecycle | +| `DefaultWorkflowEngineClient` | Default `WorkflowEngineClient` implementation | | `DefaultExtensionSender` | Default `ExtensionSender`; its public constructor owns the transport, while `nonOwning` does not | -| `DefaultA2AJavaClientRuntime` | Default A2A Java SDK runtime for HTTP/JSON-RPC/gRPC | -| `CredentialCrypto` | AES-GCM credential encryption utility and command-line entry point | -| `EnvFileLoader` | Explicit `.env` parser for host-owned configuration | -| `SslContextFactory` | TLS context construction used by transport and discovery helpers | -| `ProtocolResponses` | A2A event/result assembly helpers | -| `ClientEventMapper` | Stable event projection for callbacks and diagnostics | -| `WireLog` | Correlation context and protocol-observation facade | +| `DefaultA2AJavaClientRuntime` | Default A2A Java SDK runtime for HTTP/JSON-RPC/gRPC | +| `CredentialCrypto` | AES-GCM credential encryption utility and command-line entry point | +| `EnvFileLoader` | Explicit `.env` parser for host-owned configuration | +| `SslContextFactory` | TLS context construction used by transport and discovery helpers | +| `ProtocolResponses` | A2A event/result assembly helpers | +| `ClientEventMapper` | Stable event projection for callbacks and diagnostics | +| `WireLog` | Correlation context and protocol-observation facade | | `RemoteA2AErrorException` | Standard A2A HTTP error projection (`code`, `status`, `message`, typed details and safe headers) | --- @@ -408,8 +413,8 @@ LoadPsop convenience overloads default to `sslVerify=true`, using JVM trust and Explicit `false` skips certificate-chain and hostname verification on that orchestration HTTPS connection only, allowing development without a local CA file. The server still needs a TLS certificate; this does not bypass mTLS, modify JVM-wide defaults, or change other clients' TLS policies. Production must verify trust and matching SANs. -This is distinct from the engine's southbound HTTP/JSON-RPC verification policy. -Connection and read timeouts both default to 30 seconds. Supply +This is distinct from the engine's southbound HTTP/JSON-RPC verification policy. Connection and read timeouts both +default to 30 seconds. Supply `new LoadPsop.Timeouts(connectTimeout, readTimeout)` to set positive per-call values. ### RegistryClient @@ -651,7 +656,8 @@ not add private negotiation-state keys to the wire message. ## Thread Safety -- The default engine client supports concurrent tasks. Each workflow invocation binds its own negotiation strategy and event callback without mutating client defaults. +- The default engine client supports concurrent tasks. Each workflow invocation binds its own negotiation strategy and + event callback without mutating client defaults. - `ControlPoint` implementations must be thread-safe if used from multiple workflow executions concurrently. - `EventCallback.onEvent` is called from multiple threads (main + SSE worker threads). Use synchronization if needed. diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index 04e23baf..7096e7d7 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -1,6 +1,6 @@ # Business callback integration contract -Release `0.0.8` API; A2A-T SDK `1.1.0`, A2A Java `1.2.0.Final`. The engine owns DAG scheduling, standard A2A envelopes, +Release `0.0.9` API; A2A-T SDK `1.1.0`, 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. @@ -120,13 +120,22 @@ one. Sample initialization helpers are host-side examples, not engine dependenci ## 5. Negotiation +The host business explicitly decides whether the current task allows negotiation. An AgentCard declaration only says +that the target supports Negotiation-T. To allow it, call +`withExtension(A2ATExtension.NEGOTIATION_T.uri())` on the initial Task-T content. The engine then writes both +`message.extensions` and the `A2A-Extensions` request header without adding premature Negotiation-T metadata. Merely +implementing `onNegotiation` or supplying a `NegotiationStrategy` does not activate negotiation; the target must not +start Negotiation-T when it was not activated. + NegotiationRequest(task, originalSubmission, received, previousExchanges, remainingWait): task is the original TaskRequest; originalSubmission is the exact initial content; received is the complete current response; previousExchanges contains only this session's completed Exchange (received, reply); remainingWait is the remaining local interaction deadline. There are no engine-defined business proposal schemas. -Only a remote `INPUT_REQUIRED` carrying valid Negotiation-T Propose enters `onNegotiation`. Terminal responses never -restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. The host validates/interprets the proposal and +Only a remote `INPUT_REQUIRED` or a taskless bare message carrying a valid Negotiation-T Propose enters +`onNegotiation` (the latter is the A2A-T pre-task mode: correlated by contextId and negotiation id, with taskId-less +follow-up sends). Terminal responses never restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. Invalid +negotiation metadata on a bare message also fails explicitly instead of being treated as a normal response. The host validates/interprets the proposal and generates the final Accept/Reject/Abort with its own A2A-T client. Use `A2atMessages.contextOf(request.received())` to obtain the received context; reply with the same id, round and maxRounds. The last allowed round can still be answered. Do not call nextRound for an ending reply or return a new Propose. @@ -162,7 +171,7 @@ Each `next` entry is a `JumpCondition(step, condition)`; an empty condition make ```java WorkflowStep diagnosis = WorkflowStep.builder() .name("diagnosis") - .subtasks(List.of(Task.builder().agent("Scheduled Agent A").description("diagnose").build())) + .subtasks(List.of(Task.builder().agent("SPN Domain Agent City1").description("diagnose").build())) .next(List.of( new JumpCondition("notify", ""), // unconditional: always active, never reaches onRoute new JumpCondition("hardware_recovery", "fault.hardware"), @@ -205,17 +214,18 @@ output comes as JSON in outputs; natural-language output calls for text matching Callbacks for different conditional edges may run concurrently; do not hold shared mutable current-task state. Each workflow task activation (preparation and dispatch combined) is bounded by the client timeout (default sendTimeoutSeconds=600). Routing has a -callback timeout; task interaction/negotiation additionally has its own total wait deadline. Cancellation/timeout prevents late +callback timeout; task interaction/negotiation additionally has its own total wait deadline. Cancellation/timeout +prevents late sends; it does not automatically stop external business/LLM work. Hosts own cancellation of their resources. Synchronous 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 -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 semantic validation; the engine preserves the remote task/context, waits for -a final state and cleans up a known non-final task when local interaction cannot continue. +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 +semantic validation; the engine preserves the remote task/context, waits for a final state and cleans up a known +non-final task when local interaction cannot continue. ## 7. Independent extensions diff --git a/docs/en/DESIGN.md b/docs/en/DESIGN.md index 966acb39..b43ae6e4 100644 --- a/docs/en/DESIGN.md +++ b/docs/en/DESIGN.md @@ -261,8 +261,14 @@ generation. Negotiation-T: -Only a remote `INPUT_REQUIRED` carrying valid Negotiation-T Propose enters `onNegotiation`. Terminal responses never -restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. The host validates/interprets the proposal and +The host business decides whether a task allows negotiation through the initial `MessageContent.extensions`. Neither +the AgentCard declaration nor the callback itself activates it. The engine only maps that selection to the message and +the `A2A-Extensions` request header, then manages subsequent negotiation correlation. + +Only a remote `INPUT_REQUIRED` or a taskless bare message carrying a valid Negotiation-T Propose enters +`onNegotiation` (the latter is the A2A-T pre-task mode: correlated by contextId and negotiation id, with taskId-less +follow-up sends). Terminal responses never restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. Invalid +negotiation metadata on a bare message also fails explicitly instead of being treated as a normal response. The host validates/interprets the proposal and generates the final Accept/Reject/Abort with its own A2A-T client. Use `A2atMessages.contextOf(request.received())` to obtain the received context; reply with the same id, round and maxRounds. The last allowed round can still be answered. Do not call nextRound for an ending reply or return a new Propose. diff --git a/docs/en/DEVELOPER_GUIDE.md b/docs/en/DEVELOPER_GUIDE.md index 1a66b1b9..a71932e6 100644 --- a/docs/en/DEVELOPER_GUIDE.md +++ b/docs/en/DEVELOPER_GUIDE.md @@ -11,11 +11,11 @@ Add to your `pom.xml`: net.openan.workflow.sdk workflow-engine - 0.0.8 + 0.0.9 ``` -Version `0.0.8` is published to Maven Central and contains the APIs documented in this guide. +Version `0.0.9` is published to Maven Central and contains the APIs documented in this guide. 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. @@ -131,8 +131,15 @@ ExecutionResult result = executor.run().join(); ### 6.1 Negotiation Auto-Loop -Only a remote `INPUT_REQUIRED` carrying valid Negotiation-T Propose enters `onNegotiation`. Terminal responses never -restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. The host validates/interprets the proposal and +When the host business allows negotiation for a task, call +`withExtension(A2ATExtension.NEGOTIATION_T.uri())` on the initial Task-T `MessageContent`. An AgentCard declaration or a +negotiation callback does not add the request header automatically; the engine only maps the business selection to +`message.extensions` and `A2A-Extensions`. + +Only a remote `INPUT_REQUIRED` or a taskless bare message carrying a valid Negotiation-T Propose enters +`onNegotiation` (the latter is the A2A-T pre-task mode: correlated by contextId and negotiation id, with taskId-less +follow-up sends). Terminal responses never restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. Invalid +negotiation metadata on a bare message also fails explicitly instead of being treated as a normal response. The host validates/interprets the proposal and generates the final Accept/Reject/Abort with its own A2A-T client. Use `A2atMessages.contextOf(request.received())` to obtain the received context; reply with the same id, round and maxRounds. The last allowed round can still be answered. Do not call nextRound for an ending reply or return a new Propose. diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index 88049192..d183aaa8 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -22,11 +22,11 @@ callbacks own A2A-T generation, semantic validation, schemas and any LLM calls. net.openan.workflow.sdk workflow-engine -0.0.8 +0.0.9 ``` -Version `0.0.8` is published to Maven Central and contains the APIs documented in this guide. +Version `0.0.9` is published to Maven Central and contains the APIs documented in this guide. ## 4. Quick Start @@ -227,7 +227,7 @@ configuration without modifying system properties. Never log plaintext or the ke **Encrypt a password** -Build the jar with `mvn -pl workflow-engine -am "-Drevision=0.0.8" package`; commands below run from the repository root. +Build the jar with `mvn -pl workflow-engine -am "-Drevision=0.0.9" 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. @@ -235,10 +235,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-0.0.8.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" +java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" # Option 2: pass key as second argument -java -cp workflow-engine/target/workflow-engine-0.0.8.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b +java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b ``` Output: @@ -254,7 +254,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-0.0.8.jar dev.openan.workflow.engine.client.CredentialCrypto "plaintext" new-key` + `java -cp workflow-engine/target/workflow-engine-0.0.9.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`. @@ -392,8 +392,23 @@ authentication. ## 7. A2A-T Extensions -Only a remote `INPUT_REQUIRED` carrying valid Negotiation-T Propose enters `onNegotiation`. Terminal responses never -restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. The host validates/interprets the proposal and +Declaring Negotiation-T in the AgentCard means the target supports it; it does not enable negotiation for every task. +The host business selects it on the initial final content: + +```java +MessageContent outgoing = A2atMessages.from(generatedTask, parts) + .withExtension(A2ATExtension.NEGOTIATION_T.uri()); +``` + +The engine writes this extension set to both the A2A Message and the `A2A-Extensions` request header. Initial metadata +still contains only the Task-T body; do not manufacture Negotiation-T metadata before receiving a Propose. Configuring +a negotiation callback alone does not activate the extension, and tasks that disallow negotiation must not call +`withExtension`. + +Only a remote `INPUT_REQUIRED` or a taskless bare message carrying a valid Negotiation-T Propose enters +`onNegotiation` (the latter is the A2A-T pre-task mode: correlated by contextId and negotiation id, with taskId-less +follow-up sends). Terminal responses never restart negotiation; ordinary `INPUT_REQUIRED` fails explicitly. Invalid +negotiation metadata on a bare message also fails explicitly instead of being treated as a normal response. The host validates/interprets the proposal and generates the final Accept/Reject/Abort with its own A2A-T client. Use `A2atMessages.contextOf(request.received())` to obtain the received context; reply with the same id, round and maxRounds. The last allowed round can still be answered. Do not call nextRound for an ending reply or return a new Propose. @@ -407,10 +422,10 @@ exhausted budget or a missing handler fails locally; no implicit Accept or synth SUBMITTED/WORKING remain pending and are observed without resending the command. A business-sent Abort is never task success, even if the dispatched agent acknowledges it with COMPLETED. -For a task outside a workflow DAG, call the task client directly. It uses a fresh context but keeps -the same remote task identity through waiting and negotiation. The optional per-call strategy does -not change the client-wide ControlPoint. Plain A2A content is allowed; activated Task-T content must -include Task-T metadata and the target AgentCard must declare the extension. +For a task outside a workflow DAG, call the task client directly. It uses a fresh context but keeps the same remote task +identity through waiting and negotiation. The optional per-call strategy does not change the client-wide ControlPoint. +Plain A2A content is allowed; activated Task-T content must include Task-T metadata and the target AgentCard must +declare the extension. ```java CompletableFuture sendTask(String agentName, MessageContent content); @@ -418,9 +433,9 @@ CompletableFuture sendTask(String agentName, MessageContent c NegotiationStrategy negotiationStrategy); ``` -If a known remote task is still non-final when local interaction cannot continue, the client makes -a best-effort cancellation before completing exceptionally. Standard A2A request errors remain -exceptional results and are not converted into business task failures. +If a known remote task is still non-final when local interaction cannot continue, the client makes a best-effort +cancellation before completing exceptionally. Standard A2A request errors remain exceptional results and are not +converted into business task failures. ```java CompletableFuture sendAuthorization(String agentName, MessageContent content); @@ -659,17 +674,17 @@ Logging configuration, pretty display and observer callbacks do not determine ta ## 15. Interface Reference -| Interface/Class | Purpose | -|--------------------------------------------------------|-------------------------------------------------------------------------| -| `ExecutePsop.Builder` | Workflow execution entry point | -| `ControlPoint` / `DefaultControlPoint` | Business decisions (onTask, onSelfTask, onRoute, onNegotiation, etc.) | -| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | Workflow and standalone task interaction | -| `ExtensionSender` / `DefaultExtensionSender` | Independent Authorization-T operations and Notification-T subscriptions | -| `A2ATransport` | Shared wire layer (A2A Java client runtime, auth, SSE consumer) | -| `WorkflowEngineClientConfig` | Configuration (TLS, auth, deadlines, executor limits, negotiation exchange budget) | -| `AuthProvider` | Custom authentication | -| `EventCallback` / `EventType` | Event callback | -| `LoadPsop` / `RegistryClient` | Workflow loading / AgentCard fetching | -| `Workflow` / `WorkflowStep` / `Task` / `JumpCondition` | Workflow definition | -| `ExecutionResult` | Execution result | -| `SendMessageResult` / `TaskResult` | Message/task response | +| Interface/Class | Purpose | +|--------------------------------------------------------|------------------------------------------------------------------------------------| +| `ExecutePsop.Builder` | Workflow execution entry point | +| `ControlPoint` / `DefaultControlPoint` | Business decisions (onTask, onSelfTask, onRoute, onNegotiation, etc.) | +| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | Workflow and standalone task interaction | +| `ExtensionSender` / `DefaultExtensionSender` | Independent Authorization-T operations and Notification-T subscriptions | +| `A2ATransport` | Shared wire layer (A2A Java client runtime, auth, SSE consumer) | +| `WorkflowEngineClientConfig` | Configuration (TLS, auth, deadlines, executor limits, negotiation exchange budget) | +| `AuthProvider` | Custom authentication | +| `EventCallback` / `EventType` | Event callback | +| `LoadPsop` / `RegistryClient` | Workflow loading / AgentCard fetching | +| `Workflow` / `WorkflowStep` / `Task` / `JumpCondition` | Workflow definition | +| `ExecutionResult` | Execution result | +| `SendMessageResult` / `TaskResult` | Message/task response | diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index 1d900ed6..be4b1cb8 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -22,21 +22,21 @@ #### ExecutePsop.Builder -| 方法 | 类型 | 默认值 | 说明 | -|------------------------------------------|------|-------------|---------------------------------| -| `psop(Workflow)` | 必填 | - | PSOP 工作流定义 | -| `agentCards(List)` | 可选 | `List.of()` | 被调度智能体的 AgentCard；存在远程步骤且未传入已配置 `engineClient` 时必须提供 | +| 方法 | 类型 | 默认值 | 说明 | +|------------------------------------------|------|-------------|-------------------------------------------------------------------------------------------------| +| `psop(Workflow)` | 必填 | - | PSOP 工作流定义 | +| `agentCards(List)` | 可选 | `List.of()` | 被调度智能体的 AgentCard；存在远程步骤且未传入已配置 `engineClient` 时必须提供 | | `engineClient(WorkflowEngineClient)` | 可选 | null | 预配置客户端（null=自动创建）；传入后不可再设置 AgentCard、runtime、TLS 或 credentials 构造参数 | -| `controlPoint(ControlPoint)` | 必填 | - | 用户决策实现 | -| `runtimeIntent(String)` | 可选 | `""` | 自然语言意图，用于上下文组装 | -| `lang(String)` | 可选 | `"zh"` | 语言提示（`"zh"` 或 `"en"`） | -| `credentialsConfigPath(String)` | 可选 | null | 凭证 JSON 文件路径 | -| `sslVerify(boolean)` | 可选 | `true` | 是否验证 TLS 证书 | -| `caCertsPath(String)` | 可选 | null | CA 证书 PEM 文件路径 | -| `a2aClientRuntime(A2AJavaClientRuntime)` | 可选 | null | 自定义运行时（null = 自动创建） | -| `eventCallback(EventCallback)` | 可选 | null | 实时事件回调 | -| `onFinish(BiConsumer)` | 可选 | null | 执行完成回调 | -| `onEvent(Function)` | 可选 | null | 单事件转换钩子 | +| `controlPoint(ControlPoint)` | 必填 | - | 用户决策实现 | +| `runtimeIntent(String)` | 可选 | `""` | 自然语言意图，用于上下文组装 | +| `lang(String)` | 可选 | `"zh"` | 语言提示（`"zh"` 或 `"en"`） | +| `credentialsConfigPath(String)` | 可选 | null | 凭证 JSON 文件路径 | +| `sslVerify(boolean)` | 可选 | `true` | 是否验证 TLS 证书 | +| `caCertsPath(String)` | 可选 | null | CA 证书 PEM 文件路径 | +| `a2aClientRuntime(A2AJavaClientRuntime)` | 可选 | null | 自定义运行时（null = 自动创建） | +| `eventCallback(EventCallback)` | 可选 | null | 实时事件回调 | +| `onFinish(BiConsumer)` | 可选 | null | 执行完成回调 | +| `onEvent(Function)` | 可选 | null | 单事件转换钩子 | ```java ExecutionResult result = ExecutePsop.builder() @@ -75,27 +75,33 @@ void close(); `listTasks` 与 `cancelTask` 通过消息下发使用的同一 AgentCard、认证配置和自定义运行时暴露标准 A2A 任务管理操作。查询结果受认证身份权限约束，只包含当前身份可见的任务；调用方必须使用 -`nextPageToken` 完成分页。关闭本地客户端或通知流不等同于取消远端任务。 -当前 `subscribeToTask` 返回的 Future 表示订阅结果，不是底层 SSE 的独立关闭句柄；取消 Future -不保证终止传输。非终态订阅需要通过 owning Client/Runtime 的 `close()` 释放。 +`nextPageToken` 完成分页。关闭本地客户端或通知流不等同于取消远端任务。 当前 `subscribeToTask` 返回的 Future 表示订阅结果，不是底层 +SSE 的独立关闭句柄；取消 Future 不保证终止传输。非终态订阅需要通过 owning Client/Runtime 的 `close()` 释放。 -`sendTask` 在 DAG 外执行宿主提供的最终任务内容。每次调用使用新的 context，并复用工作流任务的发送、等待、查询和 -Negotiation-T 闭环。重载方法接受本次调用独有的协商策略，不修改客户端级 ControlPoint。超时、缺少处理器或其他本地交互失败后， +`sendTask` 在 DAG 外执行宿主提供的最终任务内容。每次调用使用新的 context，并复用工作流任务的发送、等待、查询和 Negotiation-T +闭环。重载方法接受本次调用独有的协商策略，不修改客户端级 ControlPoint。超时、缺少处理器或其他本地交互失败后， 如果已知远端任务仍处于非终态，客户端会先尝试取消，再以异常完成。`sendTask` 是公开任务提交的唯一名称；传输运行时内部仍可使用 A2A 标准动作名称 `sendMessage`。 普通 A2A 内容不要求 Task-T；一旦激活 Task-T，就必须同时携带对应 metadata，且目标 AgentCard 必须声明支持 Task-T。 -`onTask` 返回后由执行器内部调用 `sendTask`；`onTask` 不自行发送。内容是最终 parts/metadata/extensions，引擎只管理信封和交互，不创建 A2ATClient，不按 +`onTask` 返回后由执行器内部调用 `sendTask`；`onTask` 不自行发送。内容是最终 parts/metadata/extensions，引擎只管理信封和交互，不创建 +A2ATClient，不按 AgentCard 声明生成内容。模板查询和生成接口请直接使用宿主 SDK。 -只有远端 `INPUT_REQUIRED` 携带有效 Negotiation-T Propose 才进入 `onNegotiation`。 终态不会重启协商，普通 INPUT_REQUIRED -明确报告不支持的交互。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 +AgentCard 声明 Negotiation-T 不会激活协商。宿主业务需要允许当前任务协商时，在首轮 Task-T `MessageContent` 上调用 +`withExtension(A2ATExtension.NEGOTIATION_T.uri())`；引擎将业务选择同步到消息扩展和 `A2A-Extensions` 请求头。 +`onNegotiation` 或 `NegotiationStrategy` 只处理已收到的 Propose，不隐式修改首轮扩展。 + +只有远端 `INPUT_REQUIRED` 或无任务裸 message 携带有效 Negotiation-T Propose 才进入 `onNegotiation`（后者对应 +A2A-T 先协商后建任务模式：按 contextId 与协商 id 关联，续发请求不带 taskId）。 终态不会重启协商，普通 INPUT_REQUIRED +明确报告不支持的交互；无任务裸 message 上的无效协商元数据同样显式失败，不会静默当作普通结果。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， -引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 +getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 @@ -114,8 +120,8 @@ handle 再开始 I/O，监听器直接收到 handle 和完整 ReceivedMessage。 `heartbeat().lastEventAt()` 为兼容保留旧名称，实际表示最后一次传输活动；标准 SSE `: heartbeat` 会刷新它，但不会增加只统计已解码 A2A 业务事件的 `eventCount`。`lastBusinessEventAt()` 返回最近业务事件时间。 -`new DefaultExtensionSender(transport)` 默认拥有并在 `close()` 时关闭 transport；仅当 transport -由其他组件统一管理时使用 `DefaultExtensionSender.nonOwning(transport)`，并由调用方关闭订阅和 transport。 +`new DefaultExtensionSender(transport)` 默认拥有并在 `close()` 时关闭 transport；仅当 transport 由其他组件统一管理时使用 +`DefaultExtensionSender.nonOwning(transport)`，并由调用方关闭订阅和 transport。 ### WorkflowEngineClientConfig @@ -265,20 +271,20 @@ Map normalized = AgentCardNormalizer.normalize(rawMap); 以下公开类型面向高级集成。除非需要自定义 runtime、观测适配或显式生命周期管理，优先使用上述高层接口。 -| 类型 | 用途 | -|-------------------------------|------| -| `A2ATExtension` | 规范扩展名称与 URI | -| `A2ATransport` | 底层传输、认证、响应组装与订阅生命周期 | -| `DefaultWorkflowEngineClient` | `WorkflowEngineClient` 默认实现 | +| 类型 | 用途 | +|-------------------------------|-----------------------------------------------------------------------------------------------------| +| `A2ATExtension` | 规范扩展名称与 URI | +| `A2ATransport` | 底层传输、认证、响应组装与订阅生命周期 | +| `DefaultWorkflowEngineClient` | `WorkflowEngineClient` 默认实现 | | `DefaultExtensionSender` | 基于 `A2ATransport` 的 `ExtensionSender` 默认实现；公开构造器拥有 transport，`nonOwning` 工厂不拥有 | -| `DefaultA2AJavaClientRuntime` | HTTP/JSON-RPC/gRPC 的默认 A2A Java SDK runtime | -| `CredentialCrypto` | AES-GCM 凭据加密工具与命令行入口 | -| `EnvFileLoader` | 显式加载宿主自有 `.env` 配置 | -| `SslContextFactory` | 传输与发现辅助 API 的 TLS 上下文构造 | -| `ProtocolResponses` | A2A 事件与结果组装辅助方法 | -| `ClientEventMapper` | 面向回调和诊断的稳定事件投影 | -| `WireLog` | 关联上下文与协议观测门面 | -| `RemoteA2AErrorException` | 标准 A2A HTTP 错误投影（`code`、`status`、`message`、类型化详情及安全响应头） | +| `DefaultA2AJavaClientRuntime` | HTTP/JSON-RPC/gRPC 的默认 A2A Java SDK runtime | +| `CredentialCrypto` | AES-GCM 凭据加密工具与命令行入口 | +| `EnvFileLoader` | 显式加载宿主自有 `.env` 配置 | +| `SslContextFactory` | 传输与发现辅助 API 的 TLS 上下文构造 | +| `ProtocolResponses` | A2A 事件与结果组装辅助方法 | +| `ClientEventMapper` | 面向回调和诊断的稳定事件投影 | +| `WireLog` | 关联上下文与协议观测门面 | +| `RemoteA2AErrorException` | 标准 A2A HTTP 错误投影（`code`、`status`、`message`、类型化详情及安全响应头） | --- @@ -381,8 +387,8 @@ POST `/api/v1/orchestrate/search`。返回按自然语言意图匹配的工作 LoadPsop 的简便重载默认 `sslVerify=true`，使用 JVM 信任库并校验主机名。 显式传 `false` 时，仅该编排中心 HTTPS 连接跳过证书链和主机名校验，可免配本地 CA 文件联调。 服务端仍须提供 HTTPS 证书；此开关不绕过 mTLS，也不修改其他客户端或 JVM 的全局 TLS 策略。 -生产环境必须保持验证并配置正确的服务端 SAN 与信任库；本设置不是引擎南向 HTTP/JSON-RPC 的 TLS 策略变更。 -连接与读取超时默认均为 30 秒；可通过 `new LoadPsop.Timeouts(connectTimeout, readTimeout)` 为单次调用设置正值。 +生产环境必须保持验证并配置正确的服务端 SAN 与信任库；本设置不是引擎南向 HTTP/JSON-RPC 的 TLS 策略变更。 连接与读取超时默认均为 +30 秒；可通过 `new LoadPsop.Timeouts(connectTimeout, readTimeout)` 为单次调用设置正值。 ### RegistryClient diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 03629d75..465d5348 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -1,6 +1,6 @@ # 业务回调集成契约 -`0.0.8` API，A2A-T SDK `1.1.0`，A2A Java `1.2.0.Final`。 引擎负责 DAG、标准 A2A 信封、认证、传输和任务等待；宿主负责消息内容、schema、模板、语义校验和 +`0.0.9` API，A2A-T SDK `1.1.0`，A2A Java `1.2.0.Final`。 引擎负责 DAG、标准 A2A 信封、认证、传输和任务等待；宿主负责消息内容、schema、模板、语义校验和 LLM。业务契约独立于具体传输实现。 ## 1. 回调接口 @@ -105,17 +105,24 @@ MessageContent outgoing = A2atMessages.from(generated, List.of(new TextPart("处 ## 5. 协商 +是否允许本次任务协商由宿主业务显式决定。AgentCard 声明 Negotiation-T 只表示目标具备能力；需要允许协商时，宿主在首轮 +Task-T 内容上调用 `withExtension(A2ATExtension.NEGOTIATION_T.uri())`。引擎据此同时写入 `message.extensions` 和 +`A2A-Extensions` 请求头，但不添加首轮 Negotiation-T metadata。仅实现 `onNegotiation` 或传入 `NegotiationStrategy` +不会自动激活协商；未激活时对端不得发起 Negotiation-T。 + NegotiationRequest(task, originalSubmission, received, previousExchanges, remainingWait)： task 是原始 TaskRequest，originalSubmission 是首次最终提交， received 是当前完整响应，previousExchanges 只含该会话的 Exchange (received, reply)， remainingWait 是本次交互剩余时间。引擎不规定业务 proposal 分类或 schema。 -只有远端 `INPUT_REQUIRED` 携带有效 Negotiation-T Propose 才进入 `onNegotiation`。 终态不会重启协商，普通 INPUT_REQUIRED -明确报告不支持的交互。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 +只有远端 `INPUT_REQUIRED` 或无任务裸 message 携带有效 Negotiation-T Propose 才进入 `onNegotiation`（后者对应 +A2A-T 先协商后建任务模式：按 contextId 与协商 id 关联，续发请求不带 taskId）。 终态不会重启协商，普通 INPUT_REQUIRED +明确报告不支持的交互；无任务裸 message 上的无效协商元数据同样显式失败，不会静默当作普通结果。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， -引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 +getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 @@ -139,7 +146,7 @@ RouteDecision.allow (reason) 激活该边，或返回 RouteDecision.deny (reason ```java WorkflowStep diagnosis = WorkflowStep.builder() .name("diagnosis") - .subtasks(List.of(Task.builder().agent("被调度智能体 A").description("diagnose").build())) + .subtasks(List.of(Task.builder().agent("SPN Domain Agent City1").description("diagnose").build())) .next(List.of( new JumpCondition("notify", ""), // 无条件：始终激活，不经 onRoute new JumpCondition("hardware_recovery", "fault.hardware"), diff --git a/docs/zh/DESIGN.md b/docs/zh/DESIGN.md index aee0bb9d..5c107a24 100644 --- a/docs/zh/DESIGN.md +++ b/docs/zh/DESIGN.md @@ -236,13 +236,18 @@ Task-T：宿主生成最终内容，引擎只封装并发送。AgentCard 声明 Negotiation-T： -只有远端 `INPUT_REQUIRED` 携带有效 Negotiation-T Propose 才进入 `onNegotiation`。 终态不会重启协商，普通 INPUT_REQUIRED -明确报告不支持的交互。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 +宿主业务通过首轮 `MessageContent.extensions` 决定本次任务是否允许协商；AgentCard 声明和协商回调本身不触发激活。引擎只把该选择 +映射为消息扩展和 `A2A-Extensions` 请求头，并管理后续协商关联。 + +只有远端 `INPUT_REQUIRED` 或无任务裸 message 携带有效 Negotiation-T Propose 才进入 `onNegotiation`（后者对应 +A2A-T 先协商后建任务模式：按 contextId 与协商 id 关联，续发请求不带 taskId）。 终态不会重启协商，普通 INPUT_REQUIRED +明确报告不支持的交互；无任务裸 message 上的无效协商元数据同样显式失败，不会静默当作普通结果。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， -引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 +getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 diff --git a/docs/zh/DEVELOPER_GUIDE.md b/docs/zh/DEVELOPER_GUIDE.md index 5c0e76f4..fcbed543 100644 --- a/docs/zh/DEVELOPER_GUIDE.md +++ b/docs/zh/DEVELOPER_GUIDE.md @@ -10,11 +10,11 @@ net.openan.workflow.sdk workflow-engine - 0.0.8 + 0.0.9 ``` -`0.0.8` 已发布到 Maven Central，包含本文说明的接口。 +`0.0.9` 已发布到 Maven Central，包含本文说明的接口。 引擎会传递性引入 A2A 协议 SDK（`a2a-java-sdk-client`，含 REST、JSON-RPC、gRPC 传输） 和最小 A2A-T 核心（`a2a-t-core`）。生成 A2A-T 内容的宿主智能体显式依赖 a2a-t-client，校验接收内容的被调度智能体服务另引入 a2a-t-server。 @@ -127,13 +127,19 @@ try (var client = new DefaultWorkflowEngineClient(agentCards, a2aRuntime, ### 6.1 协商自动循环 -只有远端 `INPUT_REQUIRED` 携带有效 Negotiation-T Propose 才进入 `onNegotiation`。 终态不会重启协商，普通 INPUT_REQUIRED -明确报告不支持的交互。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 +宿主业务需要允许当前任务协商时，在首轮 Task-T `MessageContent` 上调用 +`withExtension(A2ATExtension.NEGOTIATION_T.uri())`。AgentCard 声明或实现协商回调不会自动增加请求头；引擎只把业务选择映射到 +`message.extensions` 和 `A2A-Extensions`。 + +只有远端 `INPUT_REQUIRED` 或无任务裸 message 携带有效 Negotiation-T Propose 才进入 `onNegotiation`（后者对应 +A2A-T 先协商后建任务模式：按 contextId 与协商 id 关联，续发请求不带 taskId）。 终态不会重启协商，普通 INPUT_REQUIRED +明确报告不支持的交互；无任务裸 message 上的无效协商元数据同样显式失败，不会静默当作普通结果。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， -引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 +getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index af8e81bd..3080aa46 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -20,11 +20,11 @@ A2A-T 工作流执行引擎是一个 Java SDK，用于基于 A2A 协议和 A2A-T net.openan.workflow.sdk workflow-engine -0.0.8 +0.0.9 ``` -`0.0.8` 已发布到 Maven Central，包含本文说明的接口。 +`0.0.9` 已发布到 Maven Central，包含本文说明的接口。 ## 4. 快速上手 @@ -154,8 +154,8 @@ try { 必填项：`psop`、`controlPoint`。其余配置项都有默认值。 -若宿主传入已配置好的 `engineClient`，AgentCard、runtime、TLS 和 credentials 均由该 client -负责，不要再设置对应的 `ExecutePsop.Builder` 构建参数；混用会在启动执行前直接报错，避免配置被静默忽略。 +若宿主传入已配置好的 `engineClient`，AgentCard、runtime、TLS 和 credentials 均由该 client 负责，不要再设置对应的 +`ExecutePsop.Builder` 构建参数；混用会在启动执行前直接报错，避免配置被静默忽略。 ## 5. 配置 @@ -217,17 +217,17 @@ A2AT_CRED_KEY=4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b **加密密码** -先执行 `mvn -pl workflow-engine -am "-Drevision=0.0.8" package`，以下命令在仓库根目录运行，仅需 SDK jar 和 JDK。 +先执行 `mvn -pl workflow-engine -am "-Drevision=0.0.9" 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-0.0.8.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" +java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" # 方式二：密钥作为第二个参数 -java -cp workflow-engine/target/workflow-engine-0.0.8.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b +java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b ``` 输出： @@ -242,7 +242,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-0.0.8.jar dev.openan.workflow.engine.client.CredentialCrypto "明文密码" 新密钥` +3. 用新密钥重新加密所有密码：`java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "明文密码" 新密钥` 4. 将新的 `enc:...` 结果更新到凭证 JSON 文件 > `.env` 文件不应提交到版本库，建议加入 `.gitignore`。 @@ -376,20 +376,32 @@ AgentCard 通过 `capabilities.extensions` 声明扩展点： ## 7. A2A-T 扩展能力 -只有远端 `INPUT_REQUIRED` 携带有效 Negotiation-T Propose 才进入 `onNegotiation`。 终态不会重启协商，普通 INPUT_REQUIRED -明确报告不支持的交互。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 +AgentCard 声明 Negotiation-T 只表示目标支持协商，不代表本次任务已经启用。是否允许协商由宿主业务在首轮最终内容中选择： + +```java +MessageContent outgoing = A2atMessages.from(generatedTask, parts) + .withExtension(A2ATExtension.NEGOTIATION_T.uri()); +``` + +引擎将 `extensions` 同时写入 A2A Message 和 `A2A-Extensions` 请求头。首轮 metadata 仍然只有 Task-T 正文；不要在收到 +Propose 前伪造 Negotiation-T metadata。仅配置协商回调不会自动激活扩展，不允许协商的任务不要调用 `withExtension`。 + +只有远端 `INPUT_REQUIRED` 或无任务裸 message 携带有效 Negotiation-T Propose 才进入 `onNegotiation`（后者对应 +A2A-T 先协商后建任务模式：按 contextId 与协商 id 关联，续发请求不带 taskId）。 终态不会重启协商，普通 INPUT_REQUIRED +明确报告不支持的交互；无任务裸 message 上的无效协商元数据同样显式失败，不会静默当作普通结果。 宿主自行校验、理解 Propose，并用自己的 A2A-T client 生成最终 Accept/Reject/Abort。 通过 `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， -引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 +getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 对于 DAG 外的独立任务，直接调用任务客户端。每次调用使用新的 context，但等待和协商续发始终保持同一个远端 taskId。 -可选的单次调用策略不会修改客户端级 ControlPoint。普通 A2A 内容可以直接发送；激活 Task-T 时必须携带 Task-T metadata， -且目标 AgentCard 必须声明该扩展。 +可选的单次调用策略不会修改客户端级 ControlPoint。普通 A2A 内容可以直接发送；激活 Task-T 时必须携带 Task-T metadata， 且目标 +AgentCard 必须声明该扩展。 ```java CompletableFuture sendTask(String agentName, MessageContent content); @@ -410,8 +422,8 @@ NotificationSubscription openNotification(String agentName, MessageContent conte transport/runtime/context。订阅监听器收到 handle 与完整 ReceivedMessage，在宿主定义的终态事件上关闭。 handle.acknowledgement() 和 completion() 分别表示 ACK 和真实流退出，两者都不是工作流前提。 -`heartbeat().lastEventAt()` 表示最后一次传输活动；标准 SSE comment 心跳（如 `: heartbeat`）会刷新该时间， -但不会增加只统计已解码 A2A 业务事件的 `eventCount`。需要判断最后一条业务消息时间时使用 +`heartbeat().lastEventAt()` 表示最后一次传输活动；标准 SSE comment 心跳（如 `: heartbeat`）会刷新该时间， 但不会增加只统计已解码 +A2A 业务事件的 `eventCount`。需要判断最后一条业务消息时间时使用 `lastBusinessEventAt()`。取消 `subscribeToTask` 返回的 Future 只停止本地等待；结束独立订阅应关闭 `NotificationSubscription`，结束客户端管理的任务订阅则关闭其 client/runtime。 @@ -616,7 +628,7 @@ contextId 与 executionId，供跨层定位。已识别的 A2A 错误使用 WARN |--------------------------------------------------------|---------------------------------------------------------------| | `ExecutePsop.Builder` | 工作流执行入口 | | `ControlPoint` / `DefaultControlPoint` | 业务决策实现（onTask、onSelfTask、onRoute、onNegotiation 等） | -| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | 工作流与独立任务交互 | +| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | 工作流与独立任务交互 | | `ExtensionSender` / `DefaultExtensionSender` | 独立 Authorization-T 操作与 Notification-T 长连接订阅 | | `A2ATransport` | 共享通信层（A2A Java 客户端 runtime、认证、SSE 消费） | | `WorkflowEngineClientConfig` | 配置（SSL、认证、A2A-T、协商轮数、自定义 Handler） | diff --git a/pom.xml b/pom.xml index 5f3b0e0e..431297b9 100644 --- a/pom.xml +++ b/pom.xml @@ -48,7 +48,7 @@ - 0.0.8-SNAPSHOT + 0.0.9-SNAPSHOT 17 UTF-8 1.2.0.Final diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationBaseAgentExecutor.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationBaseAgentExecutor.java index 5b3f0ec9..27dee6af 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationBaseAgentExecutor.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationBaseAgentExecutor.java @@ -59,9 +59,9 @@ * Accept reply it validates and consumes the filled parameters before diagnosis. Reject and Abort * are validated separately and end the task without executing diagnosis. * - *

Negotiation is triggered by incomplete SDK-validated task data, not merely by extension - * activation. A configured A2A-T client/server is mandatory for protocol generation and validation; - * the sample never manufactures protocol-shaped fallback text. + *

Negotiation is triggered only when the caller activates Negotiation-T and SDK-validated task + * data is incomplete. A configured A2A-T client/server is mandatory for protocol generation and + * validation; the sample never manufactures protocol-shaped fallback text. */ public abstract class NegotiationBaseAgentExecutor extends BaseAgentExecutor { @@ -391,6 +391,7 @@ private void handleNewTask(RequestContext ctx, AgentEmitter emitter) { } List missing = invalidTaskFields(data); if (!missing.isEmpty()) { + activateNegotiation(ctx); requestNegotiation(ctx, emitter, data, missing); } else { log.info("[{}] Parameters sufficient, skipping negotiation", getClass().getSimpleName()); @@ -405,6 +406,7 @@ private boolean isNegotiationReply(RequestContext ctx) { /** Replies are read from Negotiation-T metadata, validated by phase, then consumed once. */ private void handleFollowUp(RequestContext ctx, AgentEmitter emitter) { + activateNegotiation(ctx); Map metadata = ctx.getMessage().metadata(); var context = java.util.Objects.requireNonNull( @@ -453,6 +455,16 @@ private void handleFollowUp(RequestContext ctx, AgentEmitter emitter) { } } + private void activateNegotiation(RequestContext ctx) { + var callContext = ctx.getCallContext(); + String extension = A2ATExtension.NEGOTIATION_T.uri(); + if (callContext == null || !callContext.isExtensionRequested(extension)) { + throw new IllegalArgumentException( + "Negotiation-T was not activated in the A2A-Extensions request header"); + } + callContext.activateExtension(extension); + } + private void acceptReply( RequestContext ctx, AgentEmitter emitter, diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java index 8f6543dc..5e77c798 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java @@ -294,7 +294,8 @@ private void case_7_3() { List.of( new org.a2aproject.sdk.spec.TextPart(SpnCasePrompts.TASK_TEXT + "(参数缺失)")), metadata, - java.util.Set.of(A2ATExtension.TASK_T.uri()))) + java.util.Set.of(A2ATExtension.TASK_T.uri())) + .withExtension(A2ATExtension.NEGOTIATION_T.uri())) .join(); requireCompletedNegotiation("7.3", result, before); log.info( @@ -320,7 +321,8 @@ private void case_7_4() { List.of( new org.a2aproject.sdk.spec.TextPart(SpnCasePrompts.TASK_TEXT + "(语义错误)")), metadata, - java.util.Set.of(A2ATExtension.TASK_T.uri()))) + java.util.Set.of(A2ATExtension.TASK_T.uri())) + .withExtension(A2ATExtension.NEGOTIATION_T.uri())) .join(); requireCompletedNegotiation("7.4", result, before); log.info( diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java index e091678d..958737f1 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java @@ -130,7 +130,9 @@ public CompletableFuture onTask(TaskRequest request) { "[onTask] DEMO_NEGOTIATION agent={}, fault=missing-port," + " source=explicit-sample-switch", request.getAgentName()); - return new MessageContent(content.parts(), metadata, content.extensions()); + return new MessageContent(content.parts(), metadata, content.extensions()) + .withExtension( + dev.openan.workflow.engine.client.A2ATExtension.NEGOTIATION_T.uri()); } return content; })); diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoE2ETest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoE2ETest.java index 4822529b..7d5d0605 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoE2ETest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoE2ETest.java @@ -137,7 +137,16 @@ protected String negotiationWireBoundary() { private void assertNegotiationWireEvidence(String evidence) { String uri = dev.openan.workflow.engine.client.A2ATExtension.NEGOTIATION_T.uri(); + String taskUri = dev.openan.workflow.engine.client.A2ATExtension.TASK_T.uri(); var records = java.util.Arrays.asList(evidence.split("(?m)^.*? PROTOCOL - ")); + org.junit.jupiter.api.Assertions.assertTrue( + records.stream() + .anyMatch( + record -> + record.startsWith("[" + negotiationWireBoundary() + "] REQUEST") + && record.contains("A2A-Extensions: " + taskUri) + && record.contains(uri)), + "Missing Negotiation-T activation on the initial Task-T request"); String request = records.stream() .filter(record -> record.startsWith("[" + negotiationWireBoundary() + "] REQUEST")) 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 f2d82609..0de706a4 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 @@ -359,7 +359,8 @@ void testNegotiationAutoLoopCompletes() { new dev.openan.workflow.engine.model.MessageContent( List.of(new org.a2aproject.sdk.spec.TextPart("diagnose SPN fault")), metadata, - java.util.Set.of(A2ATExtension.TASK_T.uri())), + java.util.Set.of(A2ATExtension.TASK_T.uri())) + .withExtension(A2ATExtension.NEGOTIATION_T.uri()), negotiation -> callbacks.onNegotiation( new dev.openan.workflow.engine.model.NegotiationRequest( @@ -389,7 +390,8 @@ private SendMessageResult startIncomplete() { new MessageContent( List.of(new org.a2aproject.sdk.spec.TextPart("diagnose")), SpnCasePrompts.taskTMetadata(SpnCasePrompts.privateLineComplaintPromptBlankObject()), - java.util.Set.of(A2ATExtension.TASK_T.uri())); + java.util.Set.of(A2ATExtension.TASK_T.uri())) + .withExtension(A2ATExtension.NEGOTIATION_T.uri()); var result = transport .send( @@ -400,6 +402,25 @@ private SendMessageResult startIncomplete() { return result; } + @Test + void incompleteTaskWithoutNegotiationActivationDoesNotStartNegotiation() { + var content = + new MessageContent( + List.of(new org.a2aproject.sdk.spec.TextPart("diagnose")), + SpnCasePrompts.taskTMetadata(SpnCasePrompts.privateLineComplaintPromptBlankObject()), + java.util.Set.of(A2ATExtension.TASK_T.uri())); + + var result = + transport + .send( + agentCard, AGENT_NAME, content, java.util.UUID.randomUUID().toString(), null, null) + .join(); + + assertEquals("TASK_STATE_FAILED", result.getTaskState()); + assertTrue(result.getText().contains("Negotiation-T was not activated"), result.getText()); + assertEquals(0, diagnosisCount); + } + private NegotiationContext negotiationContext(SendMessageResult pending) { return pending.getReceivedMessages().stream() .map(A2atMessages::contextOf) @@ -505,7 +526,8 @@ void unknownPortNegotiatesOnlyObjectAndKeepsValidatedContext() { AGENT_NAME, A2atMessages.from( generated, - List.of(new org.a2aproject.sdk.spec.TextPart("diagnose unknown port"))), + List.of(new org.a2aproject.sdk.spec.TextPart("diagnose unknown port"))) + .withExtension(A2ATExtension.NEGOTIATION_T.uri()), java.util.UUID.randomUUID().toString(), null, null) diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java index 9b0dd486..559f9630 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java @@ -27,7 +27,8 @@ *

Each enum constant encapsulates the full extension URI so callers never need to hardcode URI * strings. Use these when explicitly activating final message content. * - *

The engine handles these extensions automatically: + *

The engine supports these extensions while the host keeps business activation and content + * decisions: * *

    *
  • {@link #TASK_T} - host-generated task content (workflow or standalone task) @@ -41,7 +42,7 @@ public enum A2ATExtension { /** Host-generated task content. The engine does not generate it. */ TASK_T(ExtensionUriConstants.TASK_T_EXTENSION_URI), - /** Negotiation text exchange. Handled automatically via auto-loop. */ + /** Negotiation exchange. The host activates it; the engine handles the resulting auto-loop. */ NEGOTIATION_T(ExtensionUriConstants.NEGOTIATION_T_EXTENSION_URI), /** Authorization whitelist. Pre-positioned before workflow starts. */ 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 4630e55b..1c514e06 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 @@ -671,23 +671,25 @@ public NotificationSubscription openNotificationStream( streamThread.start(); ScheduledFuture acknowledgementTimeout = timeoutScheduler.schedule( - () -> { - if (subscription.isActive() && !subscription.acknowledgement().isDone()) { - log.warn( - "[Transport] Notification-T subscription: no acknowledgement in {}s; closing unconfirmed stream", - notificationAckTimeoutSeconds); - subscription.failAcknowledgement( - new java.util.concurrent.TimeoutException( - "No subscription acknowledgement received")); - subscription.close(); - } - }, - notificationAckTimeoutSeconds, - TimeUnit.SECONDS); + () -> { + if (subscription.isActive() && !subscription.acknowledgement().isDone()) { + log.warn( + "[Transport] Notification-T subscription: no acknowledgement in {}s; closing unconfirmed stream", + notificationAckTimeoutSeconds); + subscription.failAcknowledgement( + new java.util.concurrent.TimeoutException( + "No subscription acknowledgement received")); + subscription.close(); + } + }, + notificationAckTimeoutSeconds, + TimeUnit.SECONDS); subscription .acknowledgement() .whenComplete((ignored, error) -> acknowledgementTimeout.cancel(false)); - subscription.completion().whenComplete((ignored, error) -> acknowledgementTimeout.cancel(false)); + subscription + .completion() + .whenComplete((ignored, error) -> acknowledgementTimeout.cancel(false)); return subscription; } 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 efe072fc..e8d5403b 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 @@ -329,8 +329,7 @@ public Iterable sendMessage( callContext == null ? null : callContext.getState().get(A2AJavaClientRuntime.TRANSPORT_ACTIVITY_STATE_KEY); - Runnable activityListener = - configuredActivity instanceof Runnable runnable ? runnable : null; + Runnable activityListener = configuredActivity instanceof Runnable runnable ? runnable : null; return TransportActivityMonitor.call( activityListener, () -> sendMessageObserved(agentCard, params, callContext, eventSink, logSink)); @@ -717,20 +716,19 @@ private A2AHttpClient createHttpClient() { Duration.ofSeconds(60), httpClientExecutor); return customizeHttpClient( - new A2AErrorDetectingHttpClient( - new JdkA2AHttpClient(new ObservedHttpClient(httpClient)))); + new A2AErrorDetectingHttpClient(new JdkA2AHttpClient(new ObservedHttpClient(httpClient)))); } /** - * Customizes the fully configured HTTP client used by REST and JSON-RPC transports. + * Decorates the HTTP client used by REST and JSON-RPC transports. * - *

    The supplied client already applies TLS, protocol observation, and A2A error-envelope - * handling. Subclasses may decorate it to integrate an HTTP gateway (for example, URI rewriting - * or gateway-specific headers) while preserving those behaviors. The default implementation - * returns the client unchanged. + *

    The default implementation preserves the standard A2A request paths. A custom runtime may + * override this hook for gateway-specific transport concerns such as rewriting action path + * separators, while continuing to reuse this runtime's TLS, SSE, protocol-error, and lifecycle + * handling. * - * @param httpClient fully configured transport client - * @return the client to use; never {@code null} + * @param httpClient fully configured engine HTTP client + * @return the client to supply to the A2A Java SDK transport */ protected A2AHttpClient customizeHttpClient(A2AHttpClient httpClient) { return Objects.requireNonNull(httpClient, "httpClient"); 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 a04b09a4..7c63ae43 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 @@ -100,21 +100,23 @@ public static DefaultWorkflowEngineClient owning( private static ReceivedMessage negotiationResponse(SendMessageResult result) { for (ReceivedMessage received : result.getReceivedMessages()) { - boolean present = - received.taskMetadata().containsKey(A2ATExtension.NEGOTIATION_T.uri()) - || received.message() != null - && received.message().metadata().containsKey(A2ATExtension.NEGOTIATION_T.uri()) - || received.artifacts().stream() - .anyMatch( - a -> - a.metadata() != null - && a.metadata().containsKey(A2ATExtension.NEGOTIATION_T.uri())); - if (present) return received; + if (carriesNegotiationMetadata(received)) return received; } throw new IllegalArgumentException( "Unsupported INPUT_REQUIRED interaction: no Negotiation-T Propose"); } + private static boolean carriesNegotiationMetadata(ReceivedMessage received) { + return received.taskMetadata().containsKey(A2ATExtension.NEGOTIATION_T.uri()) + || received.message() != null + && received.message().metadata().containsKey(A2ATExtension.NEGOTIATION_T.uri()) + || received.artifacts().stream() + .anyMatch( + a -> + a.metadata() != null + && a.metadata().containsKey(A2ATExtension.NEGOTIATION_T.uri())); + } + private static NegotiationContext validateReply( NegotiationContext original, MessageContent reply) { if (!reply.extensions().contains(A2ATExtension.NEGOTIATION_T.uri()) @@ -147,9 +149,39 @@ public long callbackTimeoutSeconds() { return transport.sendTimeoutSeconds(); } + private static void validateTaskExtension(AgentCard card, MessageContent content) { + String taskExtension = A2ATExtension.TASK_T.uri(); + List declaredExtensions = A2ATransport.extractExtensionUris(card); + boolean activated = content.extensions().contains(taskExtension); + boolean supplied = content.metadata().containsKey(taskExtension); + if (activated || supplied) { + if (!activated || !supplied) { + throw new IllegalArgumentException( + "Task-T content must contain matching extension activation and metadata"); + } + if (!declaredExtensions.contains(taskExtension)) { + throw new IllegalArgumentException("Target agent does not declare Task-T: " + card.name()); + } + } + String negotiationExtension = A2ATExtension.NEGOTIATION_T.uri(); + if (content.extensions().contains(negotiationExtension) + && !declaredExtensions.contains(negotiationExtension)) { + throw new IllegalArgumentException( + "Target agent does not declare Negotiation-T: " + card.name()); + } + } + + private static Throwable unwrap(Throwable error) { + Throwable current = error; + while ((current instanceof CompletionException || current instanceof ExecutionException) + && current.getCause() != null) { + current = current.getCause(); + } + return current; + } + @Override - public CompletableFuture sendTask( - String agentName, MessageContent content) { + public CompletableFuture sendTask(String agentName, MessageContent content) { TaskRequest request = TaskRequest.builder() .agentName(agentName) @@ -267,20 +299,6 @@ else if (invocation.finishing.compareAndSet(false, true)) return invocation.completion; } - private static void validateTaskExtension(AgentCard card, MessageContent content) { - String taskExtension = A2ATExtension.TASK_T.uri(); - boolean activated = content.extensions().contains(taskExtension); - boolean supplied = content.metadata().containsKey(taskExtension); - if (!activated && !supplied) return; - if (!activated || !supplied) { - throw new IllegalArgumentException( - "Task-T content must contain matching extension activation and metadata"); - } - if (!A2ATransport.extractExtensionUris(card).contains(taskExtension)) { - throw new IllegalArgumentException("Target agent does not declare Task-T: " + card.name()); - } - } - private void completeExceptionallyAfterCleanup( AgentCard card, Invocation invocation, Throwable error) { if (!invocation.finishing.compareAndSet(false, true)) { @@ -292,7 +310,8 @@ private void completeExceptionallyAfterCleanup( } Throwable primary = unwrap(error); cancelRemoteTask(card, invocation, primary) - .whenComplete((ignored, cleanupError) -> invocation.completion.completeExceptionally(primary)); + .whenComplete( + (ignored, cleanupError) -> invocation.completion.completeExceptionally(primary)); } private CompletableFuture cancelRemoteTask( @@ -329,15 +348,6 @@ private CompletableFuture cancelRemoteTask( } } - private static Throwable unwrap(Throwable error) { - Throwable current = error; - while ((current instanceof CompletionException || current instanceof ExecutionException) - && current.getCause() != null) { - current = current.getCause(); - } - return current; - } - private CompletableFuture send( AgentCard card, Invocation invocation, MessageContent content, String taskId) { if (!invocation.active()) @@ -406,6 +416,18 @@ private CompletableFuture advance( "Unsupported non-final remote task state: " + result.getTaskState())); } if (!"TASK_STATE_INPUT_REQUIRED".equals(result.getTaskState())) { + if (result.getTask() == null + && result.getTaskState().isEmpty() + && result.getReceivedMessages().stream() + .anyMatch(DefaultWorkflowEngineClient::carriesNegotiationMetadata)) { + // A2A-T pre-task negotiation: the remote proposes on a bare message before creating any + // task, so there is no taskId to anchor on; correlate by contextId + negotiation id and + // continue with taskless follow-up sends instead of failing or treating it as output. + log.info( + "[EngineClient] Taskless Negotiation-T propose: agent={}", + invocation.task.getAgentName()); + return negotiate(card, invocation, result, null); + } emitAgentResponse(invocation, result); return CompletableFuture.completedFuture(result); } @@ -413,7 +435,22 @@ private CompletableFuture advance( return CompletableFuture.failedFuture( new IllegalArgumentException("INPUT_REQUIRED has no remote task identity")); } - String remoteTask = result.getTask().id(); + return negotiate(card, invocation, result, result.getTask().id()); + } + + /** + * Shared negotiation loop for both carriers: a task-anchored {@code INPUT_REQUIRED} status + * ({@code remoteTask} set) and a taskless bare-message Propose ({@code remoteTask == null}, + * pre-task A2A-T negotiation). The follow-up send reuses the remote task id when there is one + * and stays taskless otherwise. + */ + private CompletableFuture negotiate( + AgentCard card, Invocation invocation, SendMessageResult result, String remoteTask) { + if (!invocation.original.extensions().contains(A2ATExtension.NEGOTIATION_T.uri())) { + return CompletableFuture.failedFuture( + new IllegalArgumentException( + "Remote requested Negotiation-T although it was not activated by the host")); + } ReceivedMessage received = negotiationResponse(result); NegotiationContext context = A2atMessages.contextOf(received); if (context.performative() != NegotiationPerformative.PROPOSE || context.isExhausted()) { @@ -426,8 +463,15 @@ private CompletableFuture advance( return CompletableFuture.failedFuture( new IllegalArgumentException("Negotiation round regressed or maxRounds changed")); } - String key = remoteTask + ":" + context.id() + ":" + context.round(); - if (!invocation.answered.add(key)) { + if (!invocation.answered.add(answerKey(remoteTask, context))) { + if (remoteTask == null) { + return CompletableFuture.failedFuture( + new IllegalStateException( + "Duplicate taskless Negotiation-T Propose with no remote task to observe: " + + context.id() + + " round " + + context.round())); + } return observeTask(card, invocation).thenCompose(next -> advance(card, invocation, next)); } if (++invocation.exchanges > maxNegotiationExchanges) { @@ -498,6 +542,14 @@ private CompletableFuture advance( }); } + private static String answerKey(String remoteTask, NegotiationContext context) { + return (remoteTask == null ? "taskless" : remoteTask) + + ":" + + context.id() + + ":" + + context.round(); + } + /** A send ACK is not a task result. Observe without resending the business command. */ private CompletableFuture observeTask(AgentCard card, Invocation invocation) { if (invocation.remoteTaskId == null) @@ -666,6 +718,18 @@ public void close() { if (closeTransportOnClose) transport.close(); } + private static final class PendingPoll { + final CompletableFuture completion = new CompletableFuture<>(); + volatile ScheduledFuture scheduled; + + void cancel() { + ScheduledFuture current = scheduled; + if (current != null) current.cancel(false); + completion.completeExceptionally( + new java.util.concurrent.CancellationException("Task wait ended")); + } + } + private final class Invocation { final TaskRequest task; final MessageContent original; @@ -687,10 +751,7 @@ private final class Invocation { int exchanges; Invocation( - TaskRequest task, - MessageContent content, - ControlPoint callbacks, - EventCallback events) { + TaskRequest task, MessageContent content, ControlPoint callbacks, EventCallback events) { this.task = task; this.original = content; this.callbacks = callbacks; @@ -702,10 +763,7 @@ Duration remaining() { } boolean active() { - return !closed.get() - && !finishing.get() - && !completion.isDone() - && !remaining().isZero(); + return !closed.get() && !finishing.get() && !completion.isDone() && !remaining().isZero(); } void cancelPendingPoll() { @@ -713,16 +771,4 @@ void cancelPendingPoll() { if (pending != null) pending.cancel(); } } - - private static final class PendingPoll { - final CompletableFuture completion = new CompletableFuture<>(); - volatile ScheduledFuture scheduled; - - void cancel() { - ScheduledFuture current = scheduled; - if (current != null) current.cancel(false); - completion.completeExceptionally( - new java.util.concurrent.CancellationException("Task wait ended")); - } - } } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java index f8b83721..9a2ec312 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java @@ -29,8 +29,8 @@ /** * Configuration for DefaultWorkflowEngineClient. * - *

    Use the builder to create immutable transport, authentication, timeout, and - * protocol-selection settings for one engine client. + *

    Use the builder to create immutable transport, authentication, timeout, and protocol-selection + * settings for one engine client. */ @Getter public class WorkflowEngineClientConfig { diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/control/NegotiationStrategy.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/control/NegotiationStrategy.java index 4ea382e5..b1075788 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/control/NegotiationStrategy.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/control/NegotiationStrategy.java @@ -26,11 +26,12 @@ /** * Strategy for making typed negotiation decisions. * - *

    Single responsibility: when an agent returns INPUT_REQUIRED (Negotiation-T), produce the typed - * decision to send back. This is a separate concern from workflow orchestration (task - * dispatch, routing). Users who need custom negotiation logic (LLM-based decisions, DAG-predecessor - * forwarding, etc.) implement this interface and inject it into {@link DefaultControlPoint} rather - * than mixing negotiation policy into their ControlPoint class. + *

    Single responsibility: when an agent requests negotiation (Negotiation-T) — either as a + * task-anchored {@code INPUT_REQUIRED} status or as a taskless pre-task bare-message Propose — + * produce the typed decision to send back. This is a separate concern from workflow + * orchestration (task dispatch, routing). Users who need custom negotiation logic (LLM-based + * decisions, DAG-predecessor forwarding, etc.) implement this interface and inject it into {@link + * DefaultControlPoint} rather than mixing negotiation policy into their ControlPoint class. * *

    The SDK's {@link ControlPoint#onNegotiation} remains the entry point for the auto-negotiation * loop; {@link DefaultControlPoint} delegates to an injected strategy by default. Users may still diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java index 1c398739..66c4fed0 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java @@ -19,8 +19,8 @@ package dev.openan.workflow.engine.core; -import dev.openan.workflow.engine.client.WorkflowEngineClient; import dev.openan.workflow.engine.client.WireLog; +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.control.EventType; diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/MessageContent.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/MessageContent.java index 99753713..3edd041b 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/MessageContent.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/MessageContent.java @@ -66,6 +66,13 @@ public static MessageContent parts(List> parts) { return new MessageContent(parts, Map.of(), Set.of()); } + /** Returns a copy with one additional business-selected extension activation. */ + public MessageContent withExtension(String extensionUri) { + LinkedHashSet active = new LinkedHashSet<>(extensions); + active.add(Objects.requireNonNull(extensionUri, "extensionUri")); + return new MessageContent(parts, metadata, active); + } + static List> snapshotParts(List> parts) { List> snapshot = new ArrayList<>(parts.size()); for (Part part : parts) { diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java index 1aba6d04..d53d0d23 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java @@ -62,11 +62,7 @@ public static Workflow load(String baseUrl, String psopId, String accessToken, b /** Loads one workflow with explicit connection and response-read deadlines. */ public static Workflow load( - String baseUrl, - String psopId, - String accessToken, - boolean sslVerify, - Timeouts timeouts) + String baseUrl, String psopId, String accessToken, boolean sslVerify, Timeouts timeouts) throws Exception { StringBuilder urlBuilder = new StringBuilder(baseUrl).append("/api/v1/orchestrate/psop/").append(psopId); @@ -164,11 +160,7 @@ private static RuntimeException requestFailure(HttpResult response) { } private static HttpResult execute( - String method, - String url, - String jsonBody, - boolean sslVerify, - Timeouts timeouts) + String method, String url, String jsonBody, boolean sslVerify, Timeouts timeouts) throws Exception { java.util.Objects.requireNonNull(timeouts, "timeouts"); HttpURLConnection connection = (HttpURLConnection) URI.create(url).toURL().openConnection(); @@ -218,14 +210,6 @@ public record Timeouts(Duration connectTimeout, Duration readTimeout) { requirePositiveAndSupported(readTimeout, "readTimeout"); } - int connectTimeoutMillis() { - return Math.toIntExact(connectTimeout.toMillis()); - } - - int readTimeoutMillis() { - return Math.toIntExact(readTimeout.toMillis()); - } - private static void requirePositiveAndSupported(Duration value, String name) { if (value == null || value.isZero() || value.isNegative()) { throw new IllegalArgumentException(name + " must be positive"); @@ -243,6 +227,14 @@ private static void requirePositiveAndSupported(Duration value, String name) { throw new IllegalArgumentException(name + " exceeds HttpURLConnection limit"); } } + + int connectTimeoutMillis() { + return Math.toIntExact(connectTimeout.toMillis()); + } + + int readTimeoutMillis() { + return Math.toIntExact(readTimeout.toMillis()); + } } private record HttpResult(int statusCode, String body) {} 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 ce894f39..c4c549da 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 @@ -43,6 +43,8 @@ class A2ATransportHeaderTest { private static final String TASK_T_URI = "https://projects.tmforum.org/a2aproject/telecommunication/extensions/Task-T/v1"; + private static final String NEGOTIATION_T_URI = + "https://projects.tmforum.org/a2aproject/telecommunication/extensions/Negotiation-T/v1"; private static dev.openan.workflow.engine.model.MessageContent content( String text, Map metadata) { @@ -62,6 +64,7 @@ private static AgentCard agentCard() throws Exception { "capabilities": { "streaming": false, "extensions": [ + {"uri": "%s", "required": false}, {"uri": "%s", "required": false} ] }, @@ -78,7 +81,7 @@ private static AgentCard agentCard() throws Exception { ] } """ - .formatted(TASK_T_URI); + .formatted(TASK_T_URI, NEGOTIATION_T_URI); return new ObjectMapper() .registerModule(new AgentCardJacksonModule()) .readValue(json, AgentCard.class); @@ -145,6 +148,29 @@ void forwardsAuthorizationAndActiveExtensionHeadersToRuntime() throws Exception assertEquals(TASK_T_URI, capturedHeaders.get().get("A2A-Extensions")); } + @Test + void forwardsBusinessSelectedNegotiationActivationWithoutNegotiationMetadata() throws Exception { + AtomicReference> capturedHeaders = new AtomicReference<>(); + A2AJavaClientRuntime runtime = new CapturingRuntime(capturedHeaders); + AgentCard card = agentCard(); + var extensions = new java.util.LinkedHashSet<>(List.of(TASK_T_URI, NEGOTIATION_T_URI)); + var message = + new dev.openan.workflow.engine.model.MessageContent( + List.of(new org.a2aproject.sdk.spec.TextPart("diagnose")), + Map.of(TASK_T_URI, "structured task"), + extensions); + + try (A2ATransport transport = + new A2ATransport( + List.of(card), runtime, WorkflowEngineClientConfig.builder().build())) { + transport.send(card, card.name(), message, "context-1", null, null).join(); + } + + assertEquals( + TASK_T_URI + "," + NEGOTIATION_T_URI, + capturedHeaders.get().get("A2A-Extensions")); + } + @Test void notificationTransportFailureIsNotReportedAsSubscriptionSuccess() throws Exception { AgentCard card = agentCard(); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java index 59b4a61c..25297356 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java @@ -34,40 +34,93 @@ import org.junit.jupiter.api.Test; class DefaultWorkflowEngineClientNegotiationTest { - @Test - void publicClientContractExposesOnlyTheTaskOrientedSendName() { - Set methodNames = - Arrays.stream(WorkflowEngineClient.class.getMethods()) - .map(java.lang.reflect.Method::getName) - .collect(java.util.stream.Collectors.toSet()); - assertFalse(methodNames.contains("dispatch")); - assertFalse(methodNames.contains("sendMessage")); + static A2AJavaClientRuntime runtime(Function> send) { + AtomicReference lastRequest = new AtomicReference<>(); + return new A2AJavaClientRuntime() { + public Iterable sendMessage( + AgentCard card, + MessageSendParams params, + ClientCallContext context, + Consumer sink, + Consumer logs) { + lastRequest.set(params); + List events = send.apply(params); + if (sink != null) events.forEach(sink); + return events; + } + + public org.a2aproject.sdk.spec.Task cancelTask( + AgentCard card, String taskId, ClientCallContext context) { + MessageSendParams request = lastRequest.get(); + if (request == null) throw new IllegalStateException("No task has been sent"); + return response(request, TaskState.TASK_STATE_CANCELED, MessageContent.text("canceled")) + .getTask(); + } + + public void close() {} + }; } static AgentCard card() throws Exception { + return card(true); + } + + static AgentCard cardWithoutNegotiation() throws Exception { + return card(false); + } + + private static AgentCard card(boolean supportsNegotiation) throws Exception { return new com.fasterxml.jackson.databind.ObjectMapper() .registerModule(new AgentCardJacksonModule()) .readValue( """ {"name":"test","description":"test","version":"1", - "capabilities":{"streaming":true,"extensions":[]}, + "capabilities":{"streaming":true,"extensions":[%s]}, "defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"], "skills":[],"supportedInterfaces":[{"protocolBinding":"HTTP+JSON","protocolVersion":"1.0", "url":"http://localhost:1","tenant":""}]} - """, + """ + .formatted( + supportsNegotiation + ? "{\"uri\":\"" + + A2ATExtension.NEGOTIATION_T.uri() + + "\",\"required\":false}" + : ""), AgentCard.class); } + static MessageContent negotiationTask(String text) { + return MessageContent.text(text).withExtension(A2ATExtension.NEGOTIATION_T.uri()); + } + static MessageContent negotiation(String id, int round, NegotiationPerformative performative) { + return negotiation(id, round, round, performative); + } + + static MessageContent negotiation( + String id, int round, int maxRounds, NegotiationPerformative performative) { return A2atMessages.from( new MetadataContent( "Negotiation-T/information-negotiation/propose/v1", "opaque", A2ATExtension.NEGOTIATION_T.uri(), - new NegotiationContext(id, round, round, performative)), + new NegotiationContext(id, round, maxRounds, performative)), List.of(new TextPart("not interpreted"))); } + /** A taskless A2A-T response: a bare agent message with no task lifecycle attached. */ + static List messageResponse(MessageSendParams request, MessageContent content) { + Message message = + Message.builder() + .messageId(UUID.randomUUID().toString()) + .contextId(request.message().contextId()) + .role(Message.Role.ROLE_AGENT) + .parts(content.parts()) + .metadata(content.metadata()) + .build(); + return List.of(new MessageEvent(message)); + } + static TaskEvent response(MessageSendParams request, TaskState state, MessageContent content) { Message message = Message.builder() @@ -86,31 +139,14 @@ static TaskEvent response(MessageSendParams request, TaskState state, MessageCon .build()); } - static A2AJavaClientRuntime runtime(Function> send) { - AtomicReference lastRequest = new AtomicReference<>(); - return new A2AJavaClientRuntime() { - public Iterable sendMessage( - AgentCard card, - MessageSendParams params, - ClientCallContext context, - Consumer sink, - Consumer logs) { - lastRequest.set(params); - List events = send.apply(params); - if (sink != null) events.forEach(sink); - return events; - } - - public org.a2aproject.sdk.spec.Task cancelTask( - AgentCard card, String taskId, ClientCallContext context) { - MessageSendParams request = lastRequest.get(); - if (request == null) throw new IllegalStateException("No task has been sent"); - return response(request, TaskState.TASK_STATE_CANCELED, MessageContent.text("canceled")) - .getTask(); - } - - public void close() {} - }; + @Test + void publicClientContractExposesOnlyTheTaskOrientedSendName() { + Set methodNames = + Arrays.stream(WorkflowEngineClient.class.getMethods()) + .map(java.lang.reflect.Method::getName) + .collect(java.util.stream.Collectors.toSet()); + assertFalse(methodNames.contains("dispatch")); + assertFalse(methodNames.contains("sendMessage")); } @Test @@ -167,7 +203,7 @@ void standaloneTaskAcceptsPerCallNegotiationStrategy() throws Exception { client .sendTask( "test", - MessageContent.text("start"), + negotiationTask("start"), request -> { negotiations.incrementAndGet(); return CompletableFuture.completedFuture( @@ -215,13 +251,12 @@ public org.a2aproject.sdk.spec.Task cancelTask( public void close() {} }; try (var transport = - new A2ATransport( - List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { var client = new DefaultWorkflowEngineClient(transport); assertThrows( CompletionException.class, - () -> client.sendTask("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", negotiationTask("start")).join()); assertEquals(1, cancellations.get()); } } @@ -239,8 +274,7 @@ public Iterable sendMessage( Consumer sink, Consumer logs) { sent.set(params); - java.util.concurrent.locks.LockSupport.parkNanos( - TimeUnit.MILLISECONDS.toNanos(1_200)); + java.util.concurrent.locks.LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1_200)); return List.of( response( params, @@ -266,7 +300,7 @@ public void close() {} CompletionException error = assertThrows( CompletionException.class, - () -> client.sendTask("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", negotiationTask("start")).join()); assertInstanceOf(TimeoutException.class, error.getCause()); assertTrue(canceled.await(3, TimeUnit.SECONDS)); } @@ -303,14 +337,13 @@ public org.a2aproject.sdk.spec.Task cancelTask( public void close() {} }; try (var transport = - new A2ATransport( - List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { var client = new DefaultWorkflowEngineClient(transport); CompletionException error = assertThrows( CompletionException.class, - () -> client.sendTask("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", negotiationTask("start")).join()); assertTrue(error.getCause().getMessage().contains("TASK_STATE_AUTH_REQUIRED")); assertEquals(1, cancellations.get()); } @@ -346,8 +379,36 @@ void standaloneTaskValidatesTaskTExtensionOnlyWhenActivated() throws Exception { } @Test - void standaloneTaskPreservesStandardA2AErrorWithoutCancelingAnUncreatedTask() - throws Exception { + void standaloneTaskRejectsNegotiationActivationNotDeclaredByTarget() throws Exception { + AtomicInteger sends = new AtomicInteger(); + try (var transport = + new A2ATransport( + List.of(cardWithoutNegotiation()), + runtime( + params -> { + sends.incrementAndGet(); + return List.of( + response( + params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done"))); + }), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + MessageContent content = + new MessageContent( + List.of(new TextPart("negotiation-capable task")), + Map.of(), + Set.of(A2ATExtension.NEGOTIATION_T.uri())); + + CompletionException error = + assertThrows(CompletionException.class, () -> client.sendTask("test", content).join()); + + assertTrue(error.getCause().getMessage().contains("does not declare Negotiation-T")); + assertEquals(0, sends.get()); + } + } + + @Test + void standaloneTaskPreservesStandardA2AErrorWithoutCancelingAnUncreatedTask() throws Exception { AtomicInteger cancellations = new AtomicInteger(); RemoteA2AErrorException remoteError = RemoteA2AErrorException.fromPayload( @@ -376,14 +437,13 @@ public org.a2aproject.sdk.spec.Task cancelTask( public void close() {} }; try (var transport = - new A2ATransport( - List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { var client = new DefaultWorkflowEngineClient(transport); CompletionException error = assertThrows( CompletionException.class, - () -> client.sendTask("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", negotiationTask("start")).join()); RemoteA2AErrorException actual = RemoteA2AErrorException.findIn(error); assertNotNull(actual); assertEquals(429, actual.getHttpStatus()); @@ -425,7 +485,7 @@ void answersFinalRoundAndKeepsOriginalRemoteAssociation() throws Exception { .build()); assertEquals( "TASK_STATE_COMPLETED", - client.sendTask("test", MessageContent.text("start")).join().getTaskState()); + client.sendTask("test", negotiationTask("start")).join().getTaskState()); } assertEquals(2, sent.size()); assertNull(sent.get(0).message().taskId()); @@ -433,7 +493,7 @@ void answersFinalRoundAndKeepsOriginalRemoteAssociation() throws Exception { assertEquals(sent.get(0).message().contextId(), sent.get(1).message().contextId()); assertTrue(seen.get().previousExchanges().isEmpty()); assertFalse(seen.get().remainingWait().isZero()); - assertEquals(MessageContent.text("start"), seen.get().originalSubmission()); + assertEquals(negotiationTask("start"), seen.get().originalSubmission()); } @Test @@ -459,7 +519,7 @@ void terminalStateNeverRestartsOldNegotiation() throws Exception { return CompletableFuture.failedFuture(new AssertionError()); }) .build()); - client.sendTask("test", MessageContent.text("start")).join(); + client.sendTask("test", negotiationTask("start")).join(); } assertEquals(0, callbacks.get()); } @@ -492,7 +552,7 @@ void rejectsWrongContextNewProposeAndMissingCallbackWithoutSendingAgain() throws .build()); assertThrows( CompletionException.class, - () -> client.sendTask("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", negotiationTask("start")).join()); } assertEquals(1, sends.get()); } @@ -514,7 +574,198 @@ void plainInputRequiredIsExplicitlyUnsupported() throws Exception { var client = new DefaultWorkflowEngineClient(transport); assertThrows( CompletionException.class, - () -> client.sendTask("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", negotiationTask("start")).join()); + } + } + + @Test + void tasklessProposeEntersNegotiationAndFollowUpOmitsTaskId() throws Exception { + List sent = new CopyOnWriteArrayList<>(); + AtomicInteger negotiations = new AtomicInteger(); + var runtime = + runtime( + params -> { + sent.add(params); + return sent.size() == 1 + ? messageResponse( + params, negotiation("pre-task", 1, 5, NegotiationPerformative.PROPOSE)) + : messageResponse(params, MessageContent.text("诊断结果")); + }); + try (var transport = + new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + client.setControlPoint( + ControlPoint.builder() + .onNegotiation( + request -> { + negotiations.incrementAndGet(); + var context = A2atMessages.contextOf(request.received()); + return CompletableFuture.completedFuture( + new NegotiationReply.Send( + negotiation( + context.id(), + context.round(), + context.maxRounds(), + NegotiationPerformative.ACCEPT))); + }) + .build()); + SendMessageResult result = client.sendTask("test", negotiationTask("start")).join(); + assertEquals("", result.getTaskState()); + assertEquals("诊断结果", result.getText()); + } + assertEquals(2, sent.size()); + assertNull(sent.get(0).message().taskId()); + assertNull(sent.get(1).message().taskId()); + assertEquals(1, negotiations.get()); + assertEquals(sent.get(0).message().contextId(), sent.get(1).message().contextId()); + } + + @Test + void tasklessRoundLoopAdvancesRoundsAndCompletesWithTaskAnchoredResult() throws Exception { + List sent = new CopyOnWriteArrayList<>(); + List historySizes = new CopyOnWriteArrayList<>(); + var runtime = + runtime( + params -> { + sent.add(params); + return switch (sent.size()) { + case 1 -> messageResponse( + params, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE)); + case 2 -> messageResponse( + params, negotiation("pre", 2, 5, NegotiationPerformative.PROPOSE)); + default -> List.of( + response( + params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done"))); + }; + }); + try (var transport = + new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + client.setControlPoint( + ControlPoint.builder() + .onNegotiation( + request -> { + historySizes.add(request.previousExchanges().size()); + var context = A2atMessages.contextOf(request.received()); + return CompletableFuture.completedFuture( + new NegotiationReply.Send( + negotiation( + context.id(), + context.round(), + context.maxRounds(), + NegotiationPerformative.ACCEPT))); + }) + .build()); + SendMessageResult result = client.sendTask("test", negotiationTask("start")).join(); + assertEquals("TASK_STATE_COMPLETED", result.getTaskState()); + } + assertEquals(3, sent.size()); + assertNull(sent.get(1).message().taskId()); + assertNull(sent.get(2).message().taskId()); + assertEquals(List.of(0, 1), historySizes); + } + + @Test + void tasklessProposeWithoutHostActivationFailsLoudly() throws Exception { + AtomicInteger sends = new AtomicInteger(); + try (var transport = + new A2ATransport( + List.of(card()), + runtime( + p -> { + sends.incrementAndGet(); + return messageResponse( + p, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE)); + }), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + CompletionException error = + assertThrows( + CompletionException.class, + () -> client.sendTask("test", MessageContent.text("plain")).join()); + assertTrue(error.getCause().getMessage().contains("not activated by the host")); + assertEquals(1, sends.get()); + } + } + + @Test + void tasklessProposeWithMissingNegotiationContextFailsLoudly() throws Exception { + MessageContent malformed = + new MessageContent( + List.of(new TextPart("propose without context")), + Map.of(A2ATExtension.NEGOTIATION_T.uri(), "unstructured"), + Set.of(A2ATExtension.NEGOTIATION_T.uri())); + try (var transport = + new A2ATransport( + List.of(card()), + runtime(p -> messageResponse(p, malformed)), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + CompletionException error = + assertThrows( + CompletionException.class, + () -> client.sendTask("test", negotiationTask("start")).join()); + assertTrue(error.getCause().getMessage().contains("negotiationContext")); + } + } + + @Test + void duplicateTasklessProposeFailsExplicitlyWithoutRepeatingTheCallback() throws Exception { + AtomicInteger negotiations = new AtomicInteger(); + try (var transport = + new A2ATransport( + List.of(card()), + runtime( + p -> + messageResponse( + p, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE))), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + client.setControlPoint( + ControlPoint.builder() + .onNegotiation( + request -> { + negotiations.incrementAndGet(); + var context = A2atMessages.contextOf(request.received()); + return CompletableFuture.completedFuture( + new NegotiationReply.Send( + negotiation( + context.id(), + context.round(), + context.maxRounds(), + NegotiationPerformative.ACCEPT))); + }) + .build()); + CompletionException error = + assertThrows( + CompletionException.class, + () -> client.sendTask("test", negotiationTask("start")).join()); + assertTrue(error.getCause().getMessage().contains("Duplicate taskless")); + assertEquals(1, negotiations.get()); + } + } + + @Test + void tasklessMessageWithoutNegotiationMetadataRemainsNormalResponse() throws Exception { + AtomicInteger negotiations = new AtomicInteger(); + try (var transport = + new A2ATransport( + List.of(card()), + runtime(p -> messageResponse(p, MessageContent.text("plain final"))), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + client.setControlPoint( + ControlPoint.builder() + .onNegotiation( + q -> { + negotiations.incrementAndGet(); + return CompletableFuture.failedFuture(new AssertionError()); + }) + .build()); + SendMessageResult result = client.sendTask("test", negotiationTask("start")).join(); + assertEquals("", result.getTaskState()); + assertEquals("plain final", result.getText()); + assertEquals(0, negotiations.get()); } } @@ -543,7 +794,7 @@ void abortAcknowledgementIsNotDiagnosisSuccessAndPreservesRemoteState() throws E new NegotiationReply.Send( negotiation("city1", 1, NegotiationPerformative.ABORT)))) .build()); - var result = client.sendTask("test", MessageContent.text("start")).join(); + var result = client.sendTask("test", negotiationTask("start")).join(); assertEquals("TASK_STATE_COMPLETED", result.getTaskState()); assertEquals("negotiation.aborted", result.getFailureCode()); } @@ -606,7 +857,7 @@ public void close() {} assertEquals( "TASK_STATE_COMPLETED", client - .sendTask("test", MessageContent.text("start")) + .sendTask("test", negotiationTask("start")) .get(5, TimeUnit.SECONDS) .getTaskState()); assertEquals(2, sends.get()); @@ -644,7 +895,7 @@ void canceledOrTimedOutNegotiationNeverSendsLateReply() throws Exception { return answer; }) .build()); - var result = client.sendTask("test", MessageContent.text("start")); + var result = client.sendTask("test", negotiationTask("start")); assertTrue(entered.await(1, TimeUnit.SECONDS)); if (timeout) assertThrows(ExecutionException.class, () -> result.get(3, TimeUnit.SECONDS)); else assertTrue(result.cancel(true)); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java index 5caea7ae..601b583a 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java @@ -42,6 +42,7 @@ void rejectsFileAndInlineCredentialsTogether() { assertTrue(error.getMessage().contains("mutually exclusive")); } + @Test void notificationAcknowledgementDefaultsToFiveMinutesAndRemainsConfigurable() { assertEquals( diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java index 197bfdfc..e9eb32ab 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java @@ -52,15 +52,17 @@ void waitsForTheOtherAgentAndSkipsMergeWhileReportingCorrelatedFailure(int statu var agent1 = new CompletableFuture(); var agent2 = new CompletableFuture(); var observed = new java.util.concurrent.CopyOnWriteArrayList>(); - var client = new StubWorkflowEngineClient() { - @Override public CompletableFuture sendTask( - String agentName, - MessageContent content, - dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { - started.countDown(); - return agentName.equals("agent1") ? agent1 : agent2; - } - }; + var client = + new StubWorkflowEngineClient() { + @Override + public CompletableFuture sendTask( + String agentName, + MessageContent content, + dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { + started.countDown(); + return agentName.equals("agent1") ? agent1 : agent2; + } + }; var merges = new AtomicInteger(); var callbacks = ControlPoint.builder() .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("run task"))) @@ -129,16 +131,17 @@ void protocolFailureReachesWorkflowHistoryWithoutNegotiationOrSuccessfulOutputs( var remoteError = RemoteA2AErrorException.fromPayload(error(status)); var calls = new AtomicInteger(); var negotiations = new AtomicInteger(); - var client = new StubWorkflowEngineClient("remote-agent") { - @Override - public CompletableFuture sendTask( - String agentName, - MessageContent content, - dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { - calls.incrementAndGet(); - return CompletableFuture.failedFuture(new CompletionException(remoteError)); - } - }; + var client = + new StubWorkflowEngineClient("remote-agent") { + @Override + public CompletableFuture sendTask( + String agentName, + MessageContent content, + dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { + calls.incrementAndGet(); + return CompletableFuture.failedFuture(new CompletionException(remoteError)); + } + }; var step = WorkflowStep.builder().name("remote").stepType(StepType.ALL_SUCCESS) .subtasks(List.of(Task.builder().agent("remote-agent").description("execute").build())) .next(List.of()).build(); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java index 9fe86144..8fb98f3f 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java @@ -19,14 +19,13 @@ package dev.openan.workflow.engine.core; -import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertSame; -import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import dev.openan.workflow.engine.StubWorkflowEngineClient; diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/model/MessageContentTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/model/MessageContentTest.java index 5f5e2b62..3035dc54 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/model/MessageContentTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/model/MessageContentTest.java @@ -70,6 +70,20 @@ void snapshotsPrimitiveAndArrayDataWithoutFlattening() { List.of(List.of("one", Map.of("value", 2)), 3, "{not parsed}"), received.outputs()); } + @Test + void addsBusinessSelectedExtensionWithoutMutatingOriginalContent() { + MessageContent original = + new MessageContent(List.of(new TextPart("task")), Map.of("task", "body"), Set.of("task")); + + MessageContent negotiationCapable = original.withExtension("negotiation"); + + assertEquals(Set.of("task"), original.extensions()); + assertEquals(List.of("task", "negotiation"), List.copyOf(negotiationCapable.extensions())); + assertEquals(original.parts(), negotiationCapable.parts()); + assertEquals(original.metadata(), negotiationCapable.metadata()); + assertThrows(IllegalArgumentException.class, () -> original.withExtension("")); + } + @Test void keepsMetadataOnlyResponseAndAllArtifactFields() { List items = new ArrayList<>(List.of("a"));