org.apache.maven.plugins
maven-compiler-plugin
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/spring/SpringWorkbenchApplication.java b/samples/src/main/java/dev/openan/workflow/engine/examples/SpringWorkbenchApplication.java
similarity index 68%
rename from samples/src/main/java/dev/openan/workflow/engine/examples/spring/SpringWorkbenchApplication.java
rename to samples/src/main/java/dev/openan/workflow/engine/examples/SpringWorkbenchApplication.java
index a3796ce8..6a664658 100644
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/spring/SpringWorkbenchApplication.java
+++ b/samples/src/main/java/dev/openan/workflow/engine/examples/SpringWorkbenchApplication.java
@@ -17,22 +17,28 @@
* under the License.
*/
-package dev.openan.workflow.engine.examples.spring;
+package dev.openan.workflow.engine.examples;
+
+import dev.openan.workflow.engine.client.EnvFileLoader;
+import dev.openan.workflow.engine.examples.demo.SpringSpnDemo;
+import dev.openan.workflow.engine.examples.util.EnvResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
+import java.nio.file.Path;
+
/**
* Spring Boot Workbench Agent -- northbound A2A server.
*
* This is the Workbench Agent itself: a Spring Boot service that exposes A2A-T endpoints
* (message:send, message:stream) via the {@code spring-boot-starter} auto-configuration. The
- * business logic lives in {@link SpringWorkbenchExecutor} (implements {@code AgentExecutor}).
+ * business logic lives in {@link dev.openan.workflow.engine.examples.workbench.SpringWorkbenchExecutor} (implements {@code AgentExecutor}).
*
*
Demo orchestration (starting OMC agents, sending Task-T, shutting down) is handled separately
- * by {@link SpringSpnDemo}, mirroring the existing {@code SpnCrossCityDiagnosisDemo} pattern.
+ * by {@link dev.openan.workflow.engine.examples.demo.SpringSpnDemo}.
*
*
Can also be run standalone as a Spring Boot service (no demo logic).
*/
@@ -42,8 +48,16 @@ public class SpringWorkbenchApplication {
private static final Logger log = LoggerFactory.getLogger(SpringWorkbenchApplication.class);
public static void main(String[] args) {
- System.setProperty("jdk.internal.httpclient.disableHostnameVerification", "true");
+ loadDotEnv();
SpringApplication.run(SpringWorkbenchApplication.class, args);
log.info("[SpringWorkbench] A2A server started -- ready to receive A2A-T messages");
}
+
+ /** Loads local demo configuration before Spring resolves application properties. */
+ public static void loadDotEnv() {
+ String envPath = EnvResolver.resolveEnvPath();
+ if (envPath != null) {
+ EnvFileLoader.loadToSystemProperties(Path.of(envPath));
+ }
+ }
}
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/BaseAgentExecutor.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/BaseAgentExecutor.java
index 47293b95..e8874248 100644
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/BaseAgentExecutor.java
+++ b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/BaseAgentExecutor.java
@@ -45,7 +45,7 @@ protected static String extractText(Message message) {
return sb.toString();
}
- protected static Message buildStatusMessage(String contextId, String taskId, String text) {
+ public static Message buildStatusMessage(String contextId, String taskId, String text) {
return Message.builder()
.messageId(UUID.randomUUID().toString())
.contextId(contextId)
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/ExtensionPrePositioner.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/ExtensionPrePositioner.java
deleted file mode 100644
index 90d6c288..00000000
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/ExtensionPrePositioner.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License. You may obtain
- * a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-
-package dev.openan.workflow.engine.examples.agents;
-
-import dev.openan.workflow.engine.client.ExtensionSender;
-
-import org.a2aproject.sdk.spec.AgentCard;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-import java.util.function.Consumer;
-
-/**
- * Pre-positions Authorization-T and Notification-T to downstream agents.
- *
- *
Single responsibility: send the two pre-positioning control messages (whitelist policy +
- * result subscription) to each non-workbench agent before the workflow starts. This is a
- * separate concern from workflow execution and from negotiation -- the orchestrator calls this
- * once, then proceeds to run the PSOP workflow.
- */
-public class ExtensionPrePositioner {
-
- private static final Logger log = LoggerFactory.getLogger(ExtensionPrePositioner.class);
-
- private final String authInput;
- private final String notifInput;
-
- public ExtensionPrePositioner() {
- this.authInput =
- "任务类型新增授权，操作名称业务抢通，"
- + "操作类型光模块更换，"
- + "操作对象SPN专线业务，溢权策略OMC自动执行，"
- + "触发执行条件业务投诉诊断确认故障，"
- + "预期输出返回是否成功。";
- this.notifInput =
- "通知主题为service-recovery-execution-result，"
- + "订阅条件为业务抢通方案执行结果，"
- + "上报通知数据格式为TextPart。";
- }
-
- /**
- * Pre-position Authorization-T + Notification-T to every non-workbench agent.
- *
- *
When {@code notificationCallback} is non-null, subsequent events pushed by agents
- * through the Notification-T SSE stream (e.g. recovery results) are forwarded to the
- * callback. The callback receives a Map with keys: agent, text, metadata, state.
- */
- public void prePosition(
- ExtensionSender sender,
- List agentCards,
- Consumer> notificationCallback) {
- for (AgentCard card : agentCards) {
- String name = card.name();
- if (name.contains("Workbench")) {
- continue;
- }
- log.info("[PrePosition] Authorization-T to {}", name);
- sender.sendAuthorization(name, "下发授权放行策略", authInput).join();
- log.info("[PrePosition] Notification-T to {}", name);
- sender.sendNotification(
- name, "订阅业务抢通结果通知", notifInput,
- notificationCallback).join();
- }
- log.info("[PrePosition] Extension pre-positioning complete");
- }
-
- /** Pre-position without Notification-T event callback. */
- public void prePosition(ExtensionSender sender, List agentCards) {
- prePosition(sender, agentCards, null);
- }
-}
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 dc3b35f7..b9f0f002 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
@@ -20,11 +20,16 @@
package dev.openan.workflow.engine.examples.agents;
import net.openan.a2at.sdk.client.A2ATClient;
-import net.openan.a2at.sdk.negotiation.types.model.NegotiationType;
+import net.openan.a2at.sdk.core.model.FilledParamData;
+import net.openan.a2at.sdk.core.model.StandardTemplates;
+import net.openan.a2at.sdk.server.A2ATServer;
+import net.openan.a2at.sdk.core.exception.A2ATErrorCodes;
+import net.openan.a2at.sdk.core.validation.ContentValidationException;
import org.a2aproject.sdk.server.agentexecution.RequestContext;
import org.a2aproject.sdk.server.tasks.AgentEmitter;
import org.a2aproject.sdk.spec.A2AError;
+import org.a2aproject.sdk.spec.Message;
import org.a2aproject.sdk.spec.Part;
import org.a2aproject.sdk.spec.TaskState;
import org.a2aproject.sdk.spec.TaskStatus;
@@ -37,121 +42,221 @@
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import dev.openan.workflow.engine.examples.extension.AuthorizationPolicy;
+import dev.openan.workflow.engine.examples.extension.NotificationPolicy;
+import dev.openan.workflow.engine.examples.extension.PrePositionedExtensionHandler;
+import dev.openan.workflow.engine.examples.extension.SdkSlotSchemaLoader;
+import dev.openan.workflow.engine.client.A2ATExtension;
+import dev.openan.workflow.engine.examples.negotiation.NegotiationUtils;
/**
* Server-side negotiation base, mirroring the Python reference
* (orchestration-center/samples/agents/negotiation_base_agent.py).
*
* Every agent that declares the Negotiation-T extension MUST be able to receive and reply to
- * negotiation messages. This base implements that capability: on a new task it starts a fulfillment
- * negotiation and replies with INPUT_REQUIRED carrying the negotiation context in task metadata; on
- * a follow-up ([NEGOTIATION_RESOLUTION]) it re-executes the business and completes.
+ * negotiation messages. This base implements that capability: on an incomplete task it starts an information
+ * negotiation and replies with INPUT_REQUIRED carrying the negotiation context; on
+ * an 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 triggers on every new task (deterministic), so the demo exercises the full
- * Negotiation-T round-trip each run. When the A2A-T .env is absent or LLM is disabled, the
- * negotiation context falls back to a minimal in-process context and the business text falls back
- * to the subclass default.
+ *
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.
*/
public abstract class NegotiationBaseAgentExecutor extends BaseAgentExecutor {
private static final Logger log = LoggerFactory.getLogger(NegotiationBaseAgentExecutor.class);
// Pre-positioned extensions (Authorization-T / Notification-T) are handled
// by a dedicated handler, keeping this class focused on Negotiation-T.
- private final PrePositionedExtensionHandler prePositionedHandler =
- new PrePositionedExtensionHandler();
- private final BlockingQueue notificationQueue = new LinkedBlockingQueue<>();
+ private final PrePositionedExtensionHandler prePositionedHandler;
+ private final BlockingQueue notificationQueue = new LinkedBlockingQueue<>();
+ private final java.util.concurrent.ConcurrentMap pendingNegotiationTasks =
+ new java.util.concurrent.ConcurrentHashMap<>();
private volatile A2ATClient a2atClient;
- /** The pre-positioned Authorization-T whitelist policy text, or null. */
- protected final String getAuthorizationPolicy() {
+ protected NegotiationBaseAgentExecutor() {
+ this(new PrePositionedExtensionHandler());
+ }
+
+ protected NegotiationBaseAgentExecutor(PrePositionedExtensionHandler prePositionedHandler) {
+ this.prePositionedHandler = java.util.Objects.requireNonNull(prePositionedHandler);
+ }
+
+ /** The active SDK-validated Authorization-T whitelist, or null. */
+ protected final AuthorizationPolicy getAuthorizationPolicy() {
return prePositionedHandler.getAuthorizationPolicy();
}
- /** The pre-positioned Notification-T subscription text, or null. */
- protected final String getNotificationSubscription() {
+ /** The active Notification-T subscription, or null. */
+ protected final NotificationPolicy getNotificationSubscription() {
return prePositionedHandler.getNotificationSubscription();
}
- protected void pushNotificationResult(String result) {
- notificationQueue.offer(result);
+ /**
+ * Enqueues one business-recovery notification for the independent Notification-T stream.
+ * The artifact part remains the protocol-defined event summary; the full structured prompt is
+ * carried under the Notification-T metadata URI.
+ */
+ protected void pushRecoveryNotification(String artifactName, String content) {
+ if (artifactName == null || artifactName.isBlank()) {
+ throw new IllegalArgumentException("Notification artifact name must not be blank");
+ }
+ if (content == null || content.isBlank()) {
+ throw new IllegalArgumentException("Notification content must not be blank");
+ }
+ if (getNotificationSubscription() == null) {
+ log.warn(
+ "[{}] Dropping {} because no validated Notification-T subscription is active",
+ getClass().getSimpleName(),
+ artifactName);
+ return;
+ }
+ if (!notificationQueue.offer(new NotificationEvent(artifactName, content))) {
+ log.warn(
+ "[{}] Notification queue rejected {}",
+ getClass().getSimpleName(),
+ artifactName);
+ }
+ }
+
+ protected final void pushRecoveryPlan(String content) {
+ pushRecoveryNotification("recovery-plan", content);
+ }
+
+ protected final void pushRecoveryResult(String content) {
+ pushRecoveryNotification("recovery-result", content);
+ }
+
+ private record NotificationEvent(String artifactName, String content) {
+ private NotificationEvent {
+ java.util.Objects.requireNonNull(artifactName, "artifactName");
+ java.util.Objects.requireNonNull(content, "content");
+ }
}
/**
- * Resolve the A2A-T .env path; null disables A2ATClient (negotiation still works with fallback
- * context).
+ * Resolve the A2A-T .env path; null means Negotiation-T cannot be served.
*/
protected abstract String resolveEnvPath();
- private A2ATClient a2at() {
+ private synchronized A2ATClient a2at() {
if (a2atClient != null) {
return a2atClient;
}
- if (Boolean.getBoolean("a2at.llm.disabled")) {
- return null;
- }
String env = resolveEnvPath();
if (env == null || env.isBlank()) {
- return null;
+ throw new IllegalStateException("A2A-T SDK configuration is required for OMC content validation");
}
try {
- a2atClient = new A2ATClient(Path.of(env));
+ a2atClient =
+ dev.openan.workflow.engine.examples.util.A2ATInitialization.create(
+ () -> new A2ATClient(Path.of(env)));
log.info("[{}] A2ATClient ready for negotiation", getClass().getSimpleName());
return a2atClient;
} catch (Exception e) {
- log.warn(
- "[{}] A2ATClient init failed, negotiation will use fallback context: {}",
- getClass().getSimpleName(),
- e.getMessage());
- return null;
+ throw new IllegalStateException(
+ "A2A-T client initialization failed for " + getClass().getSimpleName(), e);
+ }
+ }
+
+ private volatile net.openan.a2at.sdk.server.A2ATServer a2atServer;
+
+ /** Server-side SDK facade; initialization failure is not a validation bypass. */
+ protected synchronized net.openan.a2at.sdk.server.A2ATServer a2atServer() {
+ if (a2atServer != null) {
+ return a2atServer;
+ }
+ String env = resolveEnvPath();
+ if (env == null || env.isBlank()) {
+ throw new IllegalStateException("A2A-T SDK configuration is required for OMC content validation");
+ }
+ try {
+ a2atServer =
+ dev.openan.workflow.engine.examples.util.A2ATInitialization.create(
+ () -> new net.openan.a2at.sdk.server.A2ATServer(Path.of(env)));
+ return a2atServer;
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "A2A-T server initialization failed for " + getClass().getSimpleName(), e);
}
}
+ /**
+ * The Task-T template this agent validates incoming task prompts against. Default: the
+ * private-line complaint scenario.
+ */
+ protected net.openan.a2at.sdk.core.model.TemplateUri taskTemplateUri() {
+ return StandardTemplates.PRIVATE_LINE_COMPLAINT;
+ }
+
+ /**
+ * Parameter JSON Schema for the Task-T validate-and-fill pipeline: the task parameters this
+ * agent requires. A blank/missing slot (or validation rejection) triggers Negotiation-T.
+ */
+ protected Map buildTaskParamSchema() {
+ return SdkSlotSchemaLoader.loadConfigured(taskTemplateUri());
+ }
+
@Override
public void execute(RequestContext ctx, AgentEmitter emitter) throws A2AError {
String taskId = ctx.getTaskId();
String contextId = ctx.getContextId();
+ long started = System.nanoTime();
String input = extractText(ctx.getMessage());
// Read Task-T prompt from message metadata if present (mirrors Python SDK)
if (ctx.getMessage() != null && ctx.getMessage().metadata() != null) {
Object taskTPrompt = ctx.getMessage().metadata().get(NegotiationUtils.TASK_PROMPT_KEY);
if (taskTPrompt instanceof String taskTText
- && taskTText.length() > input.length()) {
+ && !taskTText.isBlank()) {
log.info("[{}] Using Task-T prompt from message metadata", getClass().getSimpleName());
input = taskTText;
}
}
log.info(
- "[{}] Received task: taskId={}, text={} chars, followUp={}, prePositioned={}",
+ "[{}] TASK_START taskId={}, contextId={}, inputChars={}, followUp={}, "
+ + "prePositionedExtension={}",
getClass().getSimpleName(),
taskId,
+ contextId,
input.length(),
- NegotiationUtils.isFollowUpTask(input),
- PrePositionedExtensionHandler.detect(ctx) != null);
+ isNegotiationReply(ctx),
+ PrePositionedExtensionHandler.detect(ctx));
try {
String prePositionedExt = PrePositionedExtensionHandler.detect(ctx);
if (prePositionedExt != null) {
- if (prePositionedExt.contains("Notification")) {
+ if (A2ATExtension.NOTIFICATION_T.uri().equals(prePositionedExt)) {
handleNotificationSubscription(ctx, emitter);
} else {
prePositionedHandler.handle(
ctx, emitter, prePositionedExt, getClass().getSimpleName());
}
- } else if (NegotiationUtils.isFollowUpTask(input)) {
- handleFollowUp(ctx, emitter, input);
+ } else if (isNegotiationReply(ctx)) {
+ handleFollowUp(ctx, emitter);
} else {
- handleNewTask(ctx, emitter, input);
+ handleNewTask(ctx, emitter);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.info(
- "[{}] Notification-T stream interrupted (shutdown)",
- getClass().getSimpleName());
+ "[{}] TASK_INTERRUPTED taskId={}, contextId={}, elapsedMs={}, reason=shutdown",
+ getClass().getSimpleName(),
+ taskId,
+ contextId,
+ elapsedMillis(started));
} catch (Exception e) {
- log.error("[{}] execute failed: {}", getClass().getSimpleName(), e.getMessage(), e);
+ pendingNegotiationTasks.remove(taskId);
+ log.error(
+ "[{}] TASK_FAILED taskId={}, contextId={}, elapsedMs={}, errorType={}, message={}",
+ getClass().getSimpleName(),
+ taskId,
+ contextId,
+ elapsedMillis(started),
+ e.getClass().getSimpleName(),
+ e.getMessage(),
+ e);
emitter.fail(buildStatusMessage(contextId, taskId, "Failed: " + e.getMessage()));
}
}
@@ -167,86 +272,237 @@ private void handleNotificationSubscription(RequestContext ctx, AgentEmitter emi
String taskId = ctx.getTaskId();
String contextId = ctx.getContextId();
String agentTag = getClass().getSimpleName();
- String notifUri =
- "https://projects.tmforum.org/a2aproject/telecommunication/extensions/Notification-T/v1";
- log.info("[{}] Notification-T subscription received, keeping stream open", agentTag);
- List> ackParts = List.of(new TextPart("Subscribed to recovery results"));
+ String notifUri = A2ATExtension.NOTIFICATION_T.uri();
+ Map requestMetadata = ctx.getMessage().metadata();
+ Object promptValue = requestMetadata.get(notifUri);
+ String prompt = promptValue instanceof String text ? text : String.valueOf(promptValue);
+ Object templateUri = requestMetadata.get(
+ net.openan.a2at.sdk.core.model.MetadataContent.TEMPLATE_URI_METADATA_KEY);
+ try {
+ prePositionedHandler.acceptNotification(prompt, templateUri, agentTag);
+ } catch (Exception validationError) {
+ log.warn(
+ "[{}] Notification-T rejected: {}",
+ agentTag,
+ validationError.getMessage());
+ emitter.fail(buildStatusMessage(
+ contextId, taskId, "Notification-T validation failed"));
+ return;
+ }
+ log.info(
+ "[{}] NOTIFICATION_SUBSCRIBED taskId={}, contextId={}, action=keep-stream-open",
+ agentTag,
+ taskId,
+ contextId);
+ List> ackParts = List.of(new TextPart("订阅成功，启动业务抢通事件上报任务"));
+ Map subscriptionMetadata =
+ Map.of(notifUri, "## 订阅结果\n订阅结果：成功");
emitter.addArtifact(
- ackParts, "subscription", agentTag + " subscription", Map.of(), false, true);
+ ackParts,
+ "subscription",
+ agentTag + " subscription",
+ subscriptionMetadata,
+ false,
+ true);
emitStatus(
emitter,
TaskState.TASK_STATE_WORKING,
contextId,
taskId,
- "Notification-T subscribed, waiting for recovery results",
- Map.of());
+ "订阅成功，启动业务抢通事件上报任务",
+ subscriptionMetadata);
while (!Thread.currentThread().isInterrupted()) {
- String result = notificationQueue.poll(30, TimeUnit.SECONDS);
- if (result == null) {
+ long heartbeatSeconds = Math.max(
+ 1L, Long.getLong("a2at.notification.heartbeat.seconds", 30L));
+ NotificationEvent event = notificationQueue.poll(heartbeatSeconds, TimeUnit.SECONDS);
+ if (event == null) {
+ emitStatus(
+ emitter,
+ TaskState.TASK_STATE_WORKING,
+ contextId,
+ taskId,
+ "Notification-T heartbeat",
+ Map.of(
+ "notificationHeartbeat", true,
+ "timestamp", java.time.Instant.now().toString()));
continue;
}
- log.info("[{}] Pushing recovery result via Notification-T stream", agentTag);
+ log.info(
+ "[{}] NOTIFICATION_PUSH taskId={}, contextId={}, artifactName={}, contentChars={}",
+ agentTag,
+ taskId,
+ contextId,
+ event.artifactName(),
+ event.content().length());
Map notifMeta = new LinkedHashMap<>();
- notifMeta.put(notifUri, result);
- List> resultParts = List.of(new TextPart(result));
+ notifMeta.put(notifUri, event.content());
+ List> resultParts = List.of(new TextPart("业务抢通事件"));
emitter.addArtifact(
- resultParts, "recovery-result", "recovery-result", notifMeta, false, true);
+ resultParts,
+ event.artifactName(),
+ event.artifactName(),
+ notifMeta,
+ false,
+ true);
}
}
- /** New task: start negotiation and request input. Business runs on the follow-up. */
- private void handleNewTask(RequestContext ctx, AgentEmitter emitter, String input) {
- if (needsNegotiation(input)) {
- requestNegotiation(ctx, emitter, input);
+ /** New task: consume SDK-filled data, not the incoming protocol text. */
+ private void handleNewTask(RequestContext ctx, AgentEmitter emitter) {
+ requireTemplate(ctx.getMessage().metadata(), taskTemplateUri().uri());
+ String input = requiredText(ctx.getMessage().metadata(), A2ATExtension.TASK_T.uri());
+ Map data;
+ try {
+ FilledParamData filled = a2atServer().validateTaskPromptAndDataFilling(
+ input, buildTaskParamSchema(), taskTemplateUri().uri());
+ data = SpnTaskInput.selected(java.util.Objects.requireNonNull(filled, "Task validator returned null").data());
+ } catch (ContentValidationException error) {
+ if (!A2ATErrorCodes.VALIDATION_SEMANTIC_REJECTED.equals(error.getCode())
+ && !A2ATErrorCodes.SLOT_VALIDATION_ERROR.equals(error.getCode())) throw error;
+ // A rejected prompt supplies no trusted partial data. Ask for the complete business fields.
+ data = Map.of();
+ }
+ List missing = invalidTaskFields(data);
+ if (!missing.isEmpty()) {
+ requestNegotiation(ctx, emitter, data, missing);
} else {
- log.info(
- "[{}] Parameters sufficient, skipping negotiation", getClass().getSimpleName());
- runBusinessAndComplete(ctx, emitter, input);
+ log.info("[{}] Parameters sufficient, skipping negotiation", getClass().getSimpleName());
+ runBusinessAndComplete(ctx, emitter, new SpnTaskInput(data));
}
}
- /** Follow-up: clean markers, run business, complete with extension metadata. */
- private void handleFollowUp(RequestContext ctx, AgentEmitter emitter, String input) {
- String taskId = ctx.getTaskId();
- String contextId = ctx.getContextId();
- String cleanInput = NegotiationUtils.cleanupResolutionMarker(input);
- log.info("[{}] Follow-up received, re-executing business", getClass().getSimpleName());
- runBusinessAndComplete(ctx, emitter, cleanInput);
+ private boolean isNegotiationReply(RequestContext ctx) {
+ Map meta = ctx.getMessage() != null ? ctx.getMessage().metadata() : null;
+ return NegotiationUtils.hasNegotiationMetadata(meta) && !NegotiationUtils.hasTaskMetadata(meta);
}
- /** Send INPUT_REQUIRED to start a Negotiation-T round. */
- @SuppressWarnings("unchecked")
- private void requestNegotiation(RequestContext ctx, AgentEmitter emitter, String input) {
- String taskId = ctx.getTaskId();
- String contextId = ctx.getContextId();
- Map neg = startNegotiation(input, taskId, contextId);
- Map contextData =
- neg != null && neg.get(NegotiationUtils.NEGOTIATION_CONTEXT_KEY) instanceof Map
- ? (Map) neg.get(NegotiationUtils.NEGOTIATION_CONTEXT_KEY)
- : fallbackContext();
- String negText =
- neg != null && neg.get(NegotiationUtils.NEGOTIATION_TEXT_KEY) instanceof String
- ? (String) neg.get(NegotiationUtils.NEGOTIATION_TEXT_KEY)
- : defaultNegotiationText();
- String concern = defaultNegotiationConcern();
- Map metadata =
- NegotiationUtils.negotiationResponseMetadata(contextData, negText, concern);
- emitStatus(
- emitter,
- TaskState.TASK_STATE_INPUT_REQUIRED,
- contextId,
- taskId,
- "[NEGOTIATION_REQUEST] "
- + getClass().getSimpleName()
- + " needs clarification before completing this task.",
- metadata);
- log.info("[{}] Requested negotiation (INPUT_REQUIRED)", getClass().getSimpleName());
+ /** Replies are read from Negotiation-T metadata, validated by phase, then consumed once. */
+ private void handleFollowUp(RequestContext ctx, AgentEmitter emitter) {
+ Map metadata = ctx.getMessage().metadata();
+ var context = java.util.Objects.requireNonNull(parseNegotiationContext(metadata), "Missing negotiationContext");
+ PendingNegotiation pending = pendingNegotiationTasks.get(ctx.getTaskId());
+ if (pending == null || !java.util.Objects.equals(pending.a2aContextId(), ctx.getContextId())
+ || !pending.context().id().equals(context.id()) || pending.context().round() != context.round()
+ || pending.context().maxRounds() != context.maxRounds()) {
+ throw new IllegalArgumentException("No matching pending task/negotiation context");
+ }
+ String prompt = requiredText(metadata, A2ATExtension.NEGOTIATION_T.uri());
+ boolean abort = context.performative() == net.openan.a2at.sdk.core.model.NegotiationPerformative.ABORT;
+ requireTemplate(metadata, abort ? StandardTemplates.NEGOTIATION_ABORT.uri()
+ : StandardTemplates.INFORMATION_NEGOTIATION_ACCEPT_REJECT.uri());
+ if (!pendingNegotiationTasks.remove(ctx.getTaskId(), pending))
+ throw new IllegalStateException("Negotiation reply already consumed");
+ switch (context.performative()) {
+ case ACCEPT -> acceptReply(ctx, emitter, prompt, context, pending);
+ case REJECT -> {
+ var filled = a2at().validateRejectPromptAndDataFilling(prompt, context,
+ endingReasonSchema(), StandardTemplates.INFORMATION_NEGOTIATION_ACCEPT_REJECT.uri());
+ endNegotiation(ctx, emitter, "REJECT", filled);
+ }
+ case ABORT -> {
+ var filled = a2at().validateAbortPromptAndDataFilling(prompt, context,
+ endingReasonSchema(), StandardTemplates.NEGOTIATION_ABORT.uri());
+ endNegotiation(ctx, emitter, "ABORT", filled);
+ }
+ default -> throw new IllegalArgumentException("Expected Accept, Reject or Abort reply");
+ }
+ }
+
+ private void acceptReply(RequestContext ctx, AgentEmitter emitter, String prompt,
+ net.openan.a2at.sdk.core.model.NegotiationContext context, PendingNegotiation pending) {
+ FilledParamData filled = a2at().validateAcceptPromptAndDataFilling(prompt, context,
+ replySchema(pending.requestedFields()), StandardTemplates.INFORMATION_NEGOTIATION_ACCEPT_REJECT.uri());
+ Map merged = new LinkedHashMap<>(pending.validatedData());
+ for (String field : pending.requestedFields()) merged.put(field, requiredText(filled.data(), field));
+ if (!invalidTaskFields(merged).isEmpty())
+ throw new IllegalArgumentException("Negotiation Accept did not resolve required business fields");
+ SpnTaskInput input = new SpnTaskInput(merged);
+ log.info("[{}] NEGOTIATION_APPLIED taskId={}, port={}, fields={}",
+ getClass().getSimpleName(), ctx.getTaskId(), input.accessPort(), pending.requestedFields());
+ runBusinessAndComplete(ctx, emitter, input);
+ }
+
+ private void endNegotiation(RequestContext ctx, AgentEmitter emitter, String phase, FilledParamData filled) {
+ String reason = requiredText(filled.data(), "reason");
+ log.info("[{}] NEGOTIATION_ENDED taskId={}, performative={}, action=no-diagnosis",
+ getClass().getSimpleName(), ctx.getTaskId(), phase);
+ emitter.fail(buildStatusMessage(ctx.getContextId(), ctx.getTaskId(),
+ "Negotiation-T " + phase + ": " + reason));
+ }
+
+ private static Map endingReasonSchema() {
+ return Map.of("type", "object", "properties", Map.of("reason",
+ Map.of("type", "string", "minLength", 1, "description", "从拒绝说明或终止原因中提取实际原因，不臆造")),
+ "required", List.of("reason"));
+ }
+
+ private Map replySchema(List fields) {
+ Map properties = new LinkedHashMap<>();
+ for (String field : fields) properties.put(field, Map.of("type", "string", "minLength", 1,
+ "description", field.equals("任务对象") ? "当前地市的任务对象，必须包含实际接入端口名称"
+ : "完整投诉上下文，必须包含投诉分类及OSS侧事件流水号，保留其他已有信息"));
+ return Map.of("type", "object", "properties", properties, "required", fields);
+ }
+
+ private static String requiredText(Map metadata, String key) {
+ Object value = metadata == null ? null : metadata.get(key);
+ if (!(value instanceof String text) || text.isBlank() || "null".equalsIgnoreCase(text.strip()))
+ throw new IllegalArgumentException("Missing nonblank " + key);
+ return text;
+ }
+
+ private static void requireTemplate(Map metadata, String expected) {
+ if (!expected.equals(requiredText(metadata, net.openan.a2at.sdk.core.model.MetadataContent.TEMPLATE_URI_METADATA_KEY)))
+ throw new IllegalArgumentException("Unexpected templateUri; expected " + expected);
+ }
+
+ /** Ask for only unresolved business fields; both cities use the same SDK generation path. */
+ private void requestNegotiation(RequestContext ctx, AgentEmitter emitter,
+ Map data, List missing) {
+ var context = new net.openan.a2at.sdk.core.model.NegotiationContext(
+ java.util.UUID.randomUUID().toString(), 1,
+ net.openan.a2at.sdk.core.model.NegotiationContext.DEFAULT_MAX_ROUNDS,
+ net.openan.a2at.sdk.core.model.NegotiationPerformative.PROPOSE);
+ var items = missing.stream().map(field -> new net.openan.a2at.sdk.negotiation.content.NegotiationItem(
+ field, field.equals("任务对象") ? "请提供本地市实际接入端口名称，不能使用其他地市的端口"
+ : "请提供投诉分类及本次投诉的完整上下文，保留事件流水号")).toList();
+ var generated = a2at().generateNegotiationProposePromptFromData(
+ new net.openan.a2at.sdk.negotiation.content.NegotiationProposeData(context,
+ new net.openan.a2at.sdk.negotiation.content.InformationProposeContent(
+ items, "AND：所列字段均为本次诊断必需，必须全部提供")),
+ StandardTemplates.INFORMATION_NEGOTIATION_PROPOSE.uri());
+ PendingNegotiation pending = new PendingNegotiation(ctx.getContextId(), context, Map.copyOf(data), List.copyOf(missing));
+ if (pendingNegotiationTasks.putIfAbsent(ctx.getTaskId(), pending) != null)
+ throw new IllegalStateException("Task already waiting for negotiation");
+ emitter.requiresInput(Message.builder().messageId(java.util.UUID.randomUUID().toString())
+ .contextId(ctx.getContextId()).taskId(ctx.getTaskId()).role(Message.Role.ROLE_AGENT)
+ .parts(List.of(new TextPart("存在信息缺失，请补充信息")))
+ .metadata(generated.buildMetadataContent())
+ .extensions(List.of(A2ATExtension.NEGOTIATION_T.uri())).build());
+ log.info("[{}] Requested negotiation via INPUT_REQUIRED fields={}", getClass().getSimpleName(), missing);
+ }
+
+ private record PendingNegotiation(String a2aContextId,
+ net.openan.a2at.sdk.core.model.NegotiationContext context,
+ Map validatedData, List requestedFields) {}
+
+ /** Sample-specific semantic checks after SDK filling; no raw protocol text is inspected. */
+ protected List invalidTaskFields(Map data) {
+ return SpnTaskInput.invalidFields(data);
}
/** Run business logic, emit artifact, and complete the task. */
- private void runBusinessAndComplete(RequestContext ctx, AgentEmitter emitter, String input) {
+ private void runBusinessAndComplete(
+ RequestContext ctx, AgentEmitter emitter, SpnTaskInput input) {
String taskId = ctx.getTaskId();
String contextId = ctx.getContextId();
+ long started = System.nanoTime();
+ log.info(
+ "[{}] BUSINESS_START taskId={}, contextId={}, inputChars={}",
+ getClass().getSimpleName(),
+ taskId,
+ contextId,
+ input.diagnosisInput().length());
String response = executeBusiness(ctx, emitter, input);
Map metadata = buildResponseMetadata(ctx, response);
List> parts = List.of(new TextPart(response));
@@ -254,52 +510,13 @@ private void runBusinessAndComplete(RequestContext ctx, AgentEmitter emitter, St
emitStatus(
emitter, TaskState.TASK_STATE_COMPLETED, contextId, taskId, "Completed", metadata);
emitter.complete(buildStatusMessage(contextId, taskId, "Completed"));
- log.info("[{}] Task completed", getClass().getSimpleName());
- }
-
- /** Start a fulfillment negotiation via A2ATClient, or return a fallback payload. */
- @SuppressWarnings("unchecked")
- private Map startNegotiation(String input, String taskId, String contextId) {
- A2ATClient client = a2at();
- if (client == null) {
- return fallbackNegotiationPayload(input, taskId, contextId);
- }
- try {
- Map facts = new LinkedHashMap<>();
- facts.put("agent", getClass().getSimpleName());
- if (taskId != null) facts.put("task_id", taskId);
- if (contextId != null) facts.put("context_id", contextId);
- // contentText should be the short negotiation request text, not
- // the full task input. The full input is passed as a fact so the
- // negotiation context retains the original task context.
- facts.put("input", input);
- return client.startNegotiation(
- NegotiationType.FULFILLMENT, defaultNegotiationText(), facts);
- } catch (Exception e) {
- log.warn(
- "[{}] startNegotiation failed, using fallback: {}",
- getClass().getSimpleName(),
- e.getMessage());
- return fallbackNegotiationPayload(input, taskId, contextId);
- }
- }
-
- private Map fallbackContext() {
- Map ctx = new LinkedHashMap<>();
- ctx.put("negotiationType", "fulfillment");
- ctx.put("negotiationId", UUID.randomUUID().toString());
- ctx.put("round", 1);
- ctx.put("status", "in-progress");
- ctx.put("extra", Map.of());
- return ctx;
- }
-
- private Map fallbackNegotiationPayload(
- String input, String taskId, String contextId) {
- Map payload = new LinkedHashMap<>();
- payload.put(NegotiationUtils.NEGOTIATION_CONTEXT_KEY, fallbackContext());
- payload.put(NegotiationUtils.NEGOTIATION_TEXT_KEY, defaultNegotiationText());
- return payload;
+ log.info(
+ "[{}] BUSINESS_DONE taskId={}, contextId={}, responseChars={}, elapsedMs={}",
+ getClass().getSimpleName(),
+ taskId,
+ contextId,
+ response != null ? response.length() : 0,
+ elapsedMillis(started));
}
private void emitStatus(
@@ -321,23 +538,12 @@ private void emitStatus(
emitter.emitEvent(event);
}
- // ---- subclass extension points ----
-
- /**
- * Whether the incoming task parameters are incomplete/wrong and require a Negotiation-T round
- * before proceeding. Default: false (parameters are sufficient, skip negotiation and run
- * business directly). Override to return true when the agent needs clarification.
- */
- protected boolean needsNegotiation(String input) {
- return false;
- }
-
/**
* Run the agent's actual business logic; return the response text. May emit intermediate
* events.
*/
protected abstract String executeBusiness(
- RequestContext ctx, AgentEmitter emitter, String input);
+ RequestContext ctx, AgentEmitter emitter, SpnTaskInput input);
/** Build the task metadata for the completed task (e.g. Authorization-T / Notification-T). */
protected Map buildResponseMetadata(RequestContext ctx, String response) {
@@ -346,7 +552,7 @@ protected Map buildResponseMetadata(RequestContext ctx, String r
/** Short human-readable summary for the artifact parts. */
protected String buildResultSummary() {
- return "业务处理结果";
+ return "专线业务投诉诊断任务诊断结果消息";
}
/** Artifact display name. Default: subclass simple name + " result". */
@@ -354,18 +560,35 @@ protected String buildArtifactName() {
return getClass().getSimpleName() + " result";
}
- /** Default negotiation text shown when A2ATClient is unavailable. */
- protected String defaultNegotiationText() {
- return "Please confirm the task parameters so I can proceed.";
+ /**
+ * Parses the SDK-carried negotiation context for the validate APIs.
+ *
+ * The latest SDK serializes {@code id}, {@code round}, and {@code maxRounds} under the
+ * canonical {@code negotiationContext} metadata key. The content layer remains stateless;
+ * callers own and advance this value.
+ */
+ protected static net.openan.a2at.sdk.core.model.NegotiationContext parseNegotiationContext(
+ Map metadata) {
+ if (metadata == null) {
+ return null;
+ }
+ Object raw = metadata.get(
+ net.openan.a2at.sdk.core.model.MetadataContent.NEGOTIATION_CONTEXT_METADATA_KEY);
+ return raw instanceof Map contextMap
+ ? dev.openan.workflow.engine.client.A2atMessages.contextOf(
+ new dev.openan.workflow.engine.model.ReceivedMessage(
+ new dev.openan.workflow.engine.model.MessageContent(
+ List.of(), metadata, java.util.Set.of()), Map.of(), List.of()))
+ : null;
}
- /** Default negotiation concern. */
- protected String defaultNegotiationConcern() {
- return "needs clarification";
+ private static long elapsedMillis(long startedNanos) {
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos);
}
@Override
public void cancel(RequestContext ctx, AgentEmitter emitter) throws A2AError {
+ pendingNegotiationTasks.remove(ctx.getTaskId());
emitter.cancel();
}
}
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationStrategy.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationStrategy.java
deleted file mode 100644
index 0cfa4f52..00000000
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationStrategy.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License. You may obtain
- * a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-
-package dev.openan.workflow.engine.examples.agents;
-
-import dev.openan.workflow.engine.examples.LlmHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-
-/**
- * Negotiation clarification strategy for the SPN cross-city workbench.
- *
- * Single responsibility: when a downstream agent returns INPUT_REQUIRED (Negotiation-T),
- * generate a clarification text that supplements the missing parameters. Uses LLM when the A2A-T
- * .env is configured, with a deterministic fallback.
- *
- *
Extracted from {@code WorkbenchControlPoint.onNegotiation} so that the control point only
- * cares about workflow orchestration (task dispatch, routing), while negotiation policy evolves
- * independently.
- */
-public class NegotiationStrategy implements dev.openan.workflow.engine.control.NegotiationStrategy {
-
- private static final Logger log = LoggerFactory.getLogger(NegotiationStrategy.class);
-
- private final String a2atEnvPath;
-
- public NegotiationStrategy(String a2atEnvPath) {
- this.a2atEnvPath = a2atEnvPath;
- }
-
- /**
- * Generate a clarification for the given negotiation request.
- *
- * @param agentName the agent requesting negotiation
- * @param negotiationText the concern/question raised by the agent
- * @param receiveResult the raw negotiation context (may be null/empty)
- * @return clarification text to send back to the agent
- */
- public CompletableFuture resolve(
- String agentName, String negotiationText, Map receiveResult) {
- log.info("[NegotiationStrategy] agent={}: {}", agentName, negotiationText);
- String fallback = "根据工作台上下文，客户A城市1-城市2间SPN专线中断，" + "城市1OMC告警端口Down，光功率-28dBm。";
- String sys =
- "你是SPN跨城专线故障工作台的协商澄清专家。根据协商请求，"
- + "补充客户A城市1-城市2间SPN专线中断的上下文"
- + "（城市1OMC告警端口Down、光功率-28dBm）。中文。";
- String clarification = LlmHelper.text(a2atEnvPath, sys, negotiationText, fallback);
- return CompletableFuture.completedFuture(clarification);
- }
-}
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationUtils.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationUtils.java
deleted file mode 100644
index e581955a..00000000
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/NegotiationUtils.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- * Licensed under the Apache License, Version 2.0 (the License); you may
- * not use this file except in compliance with the License. You may obtain
- * a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-
-package dev.openan.workflow.engine.examples.agents;
-
-import dev.openan.workflow.engine.client.A2ATExtension;
-
-import net.openan.a2at.sdk.negotiation.runtime.NegotiationHandler;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-
-/**
- * Constants and helpers for the server-side negotiation flow, mirroring the Python reference in
- * orchestration-center/common/negotiation_utils.py.
- *
- * Markers and key names align with the a2a-t-sdk-java negotiation payload mapper
- * (NEGOTIATION_CONTEXT_KEY / NEGOTIATION_TEXT_KEY) and the engine-side NegotiationTHandler /
- * autoNegotiate follow-up format.
- */
-final class NegotiationUtils {
-
- static final String NEGOTIATION_RESOLUTION_MARKER = "[NEGOTIATION_RESOLUTION]";
- static final String NEGOTIATION_REQUEST_MARKER = "[NEGOTIATION_REQUEST]";
- static final String NEGOTIATION_CONTEXT_MARKER = "[NEGOTIATION_CONTEXT]";
- static final String NEGOTIATION_CONCERN_KEY = "negotiationConcern";
- static final String NEGOTIATION_CONTEXT_KEY = NegotiationHandler.NEGOTIATION_CONTEXT_KEY;
- static final String NEGOTIATION_TEXT_KEY = A2ATExtension.NEGOTIATION_T.uri();
- static final String TASK_PROMPT_KEY = A2ATExtension.TASK_T.uri();
-
- private NegotiationUtils() {}
-
- static boolean isFollowUpTask(String text) {
- return text != null && text.contains(NEGOTIATION_RESOLUTION_MARKER);
- }
-
- /** Extract the original task text from a [NEGOTIATION_RESOLUTION] follow-up message. */
- static String extractOriginalTask(String text) {
- if (text == null || !text.contains(NEGOTIATION_RESOLUTION_MARKER)) {
- return text != null ? text : "";
- }
- int idx = text.indexOf("Original Task:");
- if (idx < 0) {
- return "";
- }
- String after = text.substring(idx + "Original Task:".length()).trim();
- int end = after.indexOf("\n\nPlease re-execute the task based on the clarification above.");
- if (end >= 0) {
- after = after.substring(0, end).trim();
- }
- return after;
- }
-
- /** Strip negotiation markers to recover a clean task text for re-execution. */
- static String cleanupResolutionMarker(String text) {
- if (text == null) {
- return "";
- }
- if (!text.contains(NEGOTIATION_RESOLUTION_MARKER)) {
- return text;
- }
- String original = extractOriginalTask(text);
- return original.isEmpty() ? text : original;
- }
-
- /** Build the negotiation response metadata placed on an INPUT_REQUIRED task. */
- static Map negotiationResponseMetadata(
- Map contextData, String negotiationText, String concern) {
- Map metadata = new LinkedHashMap<>();
- if (contextData != null && !contextData.isEmpty()) {
- metadata.put(NEGOTIATION_CONTEXT_KEY, contextData);
- }
- if (negotiationText != null && !negotiationText.isEmpty()) {
- metadata.put(NEGOTIATION_TEXT_KEY, negotiationText);
- }
- if (concern != null && !concern.isEmpty()) {
- metadata.put(NEGOTIATION_CONCERN_KEY, concern);
- }
- return metadata;
- }
-
- /** Build the [NEGOTIATION_RESOLUTION] follow-up text the client resends. */
- static String buildResolutionMessage(String originalTask, String resolutionText) {
- return NEGOTIATION_RESOLUTION_MARKER
- + "\n"
- + "The engine has reviewed your negotiation request and provides the following clarification:\n\n"
- + resolutionText
- + "\n\n---\nOriginal Task:\n"
- + originalTask
- + "\n\nPlease re-execute the task based on the clarification above.";
- }
-}
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/PrePositionedExtensionHandler.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/PrePositionedExtensionHandler.java
deleted file mode 100644
index 9eaa7150..00000000
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/PrePositionedExtensionHandler.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License. You may obtain
- * a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-package dev.openan.workflow.engine.examples.agents;
-
-import org.a2aproject.sdk.server.agentexecution.RequestContext;
-import org.a2aproject.sdk.server.tasks.AgentEmitter;
-import org.a2aproject.sdk.spec.Message;
-import org.a2aproject.sdk.spec.Part;
-import org.a2aproject.sdk.spec.TaskState;
-import org.a2aproject.sdk.spec.TaskStatus;
-import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
-import org.a2aproject.sdk.spec.TextPart;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * Handles pre-positioned Authorization-T / Notification-T messages on the agent server side.
- *
- * This is a separate concern from Negotiation-T. Pre-positioned extensions are sent by
- * the orchestrator before the workflow starts (via {@code
- * WorkflowEngineClient.sendExtensionMessage}) to establish whitelists and subscriptions. When an
- * agent receives one, it must:
- *
- *
- * - Store the payload for later use during business execution
- *
- Acknowledge receipt with a short artifact
- *
- Complete the task immediately (no negotiation, no business logic)
- *
- *
- * Extracted from {@code NegotiationBaseAgentExecutor} so that the latter only carries
- * Negotiation-T responsibility.
- */
-public class PrePositionedExtensionHandler {
- private static final Logger log = LoggerFactory.getLogger(PrePositionedExtensionHandler.class);
- private volatile String authorizationPolicy;
- private volatile String notificationSubscription;
-
- /**
- * Detect whether the incoming message carries an Authorization-T or Notification-T extension in
- * its metadata. Returns the extension keyword ("Authorization-T" or "Notification-T"), or null
- * for a normal task.
- */
- public static String detect(RequestContext ctx) {
- Message msg = ctx.getMessage();
- if (msg == null || msg.metadata() == null) {
- return null;
- }
- for (String key : msg.metadata().keySet()) {
- if (key.contains("Authorization-T")) {
- return "Authorization-T";
- }
- if (key.contains("Notification-T")) {
- return "Notification-T";
- }
- }
- return null;
- }
-
- private static void emitStatus(
- AgentEmitter emitter,
- TaskState state,
- String contextId,
- String taskId,
- String text,
- Map metadata) {
- TaskStatus status =
- new TaskStatus(
- state, BaseAgentExecutor.buildStatusMessage(contextId, taskId, text), null);
- TaskStatusUpdateEvent event =
- TaskStatusUpdateEvent.builder()
- .taskId(taskId)
- .contextId(contextId)
- .status(status)
- .metadata(metadata)
- .build();
- emitter.emitEvent(event);
- }
-
- /** The pre-positioned Authorization-T whitelist policy text, or null. */
- public String getAuthorizationPolicy() {
- return authorizationPolicy;
- }
-
- /** The pre-positioned Notification-T subscription text, or null. */
- public String getNotificationSubscription() {
- return notificationSubscription;
- }
-
- /**
- * Handle a pre-positioned message: store the payload, emit an ACK artifact, and complete the
- * task.
- *
- * @param agentTag short agent class name for logging
- */
- public void handle(
- RequestContext ctx, AgentEmitter emitter, String extKeyword, String agentTag) {
- String taskId = ctx.getTaskId();
- String contextId = ctx.getContextId();
- String payloadText =
- ctx.getMessage().metadata().entrySet().stream()
- .filter(e -> e.getKey().contains(extKeyword))
- .map(Map.Entry::getValue)
- .findFirst()
- .map(v -> v instanceof String s ? s : String.valueOf(v))
- .orElse("");
- if (extKeyword.contains("Authorization")) {
- authorizationPolicy = payloadText;
- } else if (extKeyword.contains("Notification")) {
- notificationSubscription = payloadText;
- }
- log.info(
- "[{}] Pre-positioned {} received, payload length={}",
- agentTag,
- extKeyword,
- payloadText.length());
- String ackText = extKeyword + " pre-positioning acknowledged";
- List> parts = List.of(new TextPart(ackText));
- emitter.addArtifact(parts, "result", agentTag + " ack", Map.of(), false, true);
- emitStatus(
- emitter,
- TaskState.TASK_STATE_COMPLETED,
- contextId,
- taskId,
- extKeyword + " pre-positioned successfully",
- Map.of());
- emitter.complete(BaseAgentExecutor.buildStatusMessage(contextId, taskId, "Completed"));
- log.info("[{}] {} pre-positioning completed", agentTag, extKeyword);
- }
-}
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity1Executor.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity1Executor.java
index cebb38ff..62cdf79b 100644
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity1Executor.java
+++ b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity1Executor.java
@@ -19,60 +19,59 @@
package dev.openan.workflow.engine.examples.agents;
-import dev.openan.workflow.engine.examples.LlmHelper;
+import dev.openan.workflow.engine.examples.extension.AuthorizationPolicy;
+import dev.openan.workflow.engine.examples.extension.PrePositionedExtensionHandler;
+import dev.openan.workflow.engine.examples.util.LlmHelper;
import org.a2aproject.sdk.server.agentexecution.RequestContext;
import org.a2aproject.sdk.server.tasks.AgentEmitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.LocalDate;
+import java.time.ZoneId;
import java.util.LinkedHashMap;
import java.util.Map;
+import dev.openan.workflow.engine.examples.negotiation.NegotiationUtils;
+import dev.openan.workflow.engine.examples.util.EnvResolver;
/**
* SPN Domain Agent for City1 (City1 OMC).
*
- * Server-side negotiation-capable (extends {@link NegotiationBaseAgentExecutor}): on a new task
+ *
Server-side negotiation-capable (extends {@link NegotiationBaseAgentExecutor}): on an incomplete task
* it replies INPUT_REQUIRED to start a Negotiation-T round, and on the follow-up it runs the
- * diagnosis/recovery business. City1 side has a FAULT (port Down, optical power -28dBm).
- * Diagnosis/recovery text is LLM-generated (deepseek) when the A2A-T .env is configured, else
- * deterministic. Authorization-T and Notification-T are pre-positioned before the workflow starts,
- * so this agent no longer injects them in response metadata.
+ * diagnosis business. City1 side has a FAULT (port Down, optical power -28dBm). Diagnosis text is
+ * LLM-generated when the A2A-T .env is configured, otherwise deterministic. The recovery event
+ * payload is formatted deterministically so required protocol fields and identifiers cannot be
+ * hallucinated. Authorization-T and Notification-T use independent channels.
*/
public class SpnDomainAgentCity1Executor extends NegotiationBaseAgentExecutor {
private static final Logger log = LoggerFactory.getLogger(SpnDomainAgentCity1Executor.class);
+ // Diagnosis result placed in artifact.metadata[Task-T URI], mirroring spec case 7.1.
private static final String FAULT_DIAGNOSIS_RESULT =
- "诊断结果：城市1城市OMC诊断结果 - 端口Down，光功率-28dBm(低于阈值)，存在故障。\n"
- + "修复方案：更换城市1OMC端口光模块，恢复端口Down状态。此修复方案需授权后执行（授权已预置）。\n"
- + "故障根因：城市1OMC端口光模块故障。";
+ "1. 诊断结果：成功\n"
+ + "2. 诊断结果详情：P33206-YWHJ-业务汇聚机房1(990)的12-TPJ1EM8F-4端口出现无收光问题，"
+ + "判断原因为对端设备掉电或端口关闭\n"
+ + "3. 修复建议：恢复供电后重新启动网元，或者重新开启端口";
- private static final String RECOVERY_RESULT = "城市1OMC端口光模块已更换，端口恢复Up，专线业务恢复正常。";
+ private static final String ACCESS_PORT = "P781-珠江新城-PTN7900-23-TPA1EG24-17";
+ private static final String RECOVERY_DETAIL =
+ "将时延越限和丢包越限的受影响隧道调优到新的可用路径，保留原路径用于回退。";
- private static String llmDiagnosisResult(String input, String fallback) {
- String env = EnvResolver.resolveEnvPath();
- String sys = "你是SPN领域城市1OMC故障诊断专家。根据输入诊断信息，输出诊断结果、修复方案和故障根因，提及城市1。简洁专业，中文。";
- String user = "输入：\n" + input + "\n\n已知：城市1OMC端口port-7=DOWN，光功率-28dBm（阈值-20dBm）。请输出诊断结论。";
- return LlmHelper.text(env, sys, user, fallback);
- }
-
- private static String llmRecoveryResult(String input, String fallback) {
- String env = EnvResolver.resolveEnvPath();
- String sys = "你是SPN领域城市1OMC抢通执行专家。用一句话报告抢通成功结果，提及城市1OMC端口恢复Up、专线业务恢复。中文。";
- String user = "输入：\n" + input + "\n\n已更换光模块，端口恢复Up。请输出抢通结果。";
- return LlmHelper.text(env, sys, user, fallback);
+ public SpnDomainAgentCity1Executor() {
+ super();
}
- private static void sleepBriefly() {
- try {
- Thread.sleep(300);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
+ public SpnDomainAgentCity1Executor(PrePositionedExtensionHandler extensionHandler) {
+ super(extensionHandler);
}
- @Override
- protected boolean needsNegotiation(String input) {
- return true;
+ private static String llmDiagnosisResult(String input, String fallback) {
+ String env = EnvResolver.resolveEnvPath();
+ String sys = "你是SPN领域OMC故障诊断专家。按如下结构输出：1. 诊断结果（成功/失败）；"
+ + "2. 诊断结果详情（故障根因点资源对象、端口、现象）；3. 修复建议。简洁专业，中文。";
+ String user = "输入：\n" + input + "\n\n已知：P33206-YWHJ-业务汇聚机房1(990)的12-TPJ1EM8F-4端口出现无收光问题。请输出诊断结果。";
+ return LlmHelper.text(env, sys, user, fallback);
}
@Override
@@ -81,43 +80,79 @@ protected String resolveEnvPath() {
}
@Override
- protected String executeBusiness(RequestContext ctx, AgentEmitter emitter, String input) {
- String diagnosisResult = llmDiagnosisResult(input, FAULT_DIAGNOSIS_RESULT);
- String recoveryResult = selfTriggerRecovery(ctx, diagnosisResult);
- return diagnosisResult + "\n\n" + recoveryResult;
+ protected String executeBusiness(RequestContext ctx, AgentEmitter emitter, SpnTaskInput input) {
+ String diagnosisResult = llmDiagnosisResult(input.diagnosisInput(), FAULT_DIAGNOSIS_RESULT);
+ publishRecoveryLifecycle(ctx, input);
+ // Task-T returns diagnosis only. Recovery plan/result belong exclusively to the
+ // independently established Notification-T subscription.
+ return diagnosisResult;
}
/**
- * After diagnosis, check the pre-positioned Authorization-T whitelist policy. If the repair
- * action matches the whitelist, execute recovery and return the result (reported as
- * Notification-T metadata by the base class). If not in whitelist, return a refusal message.
+ * After diagnosis, publish the recovery plan. If it matches the active Authorization-T
+ * whitelist, execute it automatically and publish the final result as a second event.
*/
- private String selfTriggerRecovery(RequestContext ctx, String diagnosisResult) {
- String policy = getAuthorizationPolicy();
+ private void publishRecoveryLifecycle(RequestContext ctx, SpnTaskInput input) {
+ AuthorizationPolicy policy = getAuthorizationPolicy();
boolean inWhitelist =
policy != null
- && !policy.isEmpty()
- && (policy.contains("业务抢通")
- || policy.contains("光模块")
- || policy.contains("授权"));
+ && policy.authorizes(
+ "业务投诉诊断",
+ "业务抢通",
+ "隧道调优",
+ LocalDate.now(ZoneId.of("Asia/Shanghai")));
+ String taskId = ctx.getTaskId() != null ? ctx.getTaskId() : "unknown-task";
+ pushRecoveryPlan(formatRecoveryEvent("未启动", taskId, inWhitelist, null, null, input));
if (inWhitelist) {
log.info("[SPN-Domain-Agent] Fault in whitelist, self-triggering recovery");
- String recoveryResult = llmRecoveryResult(diagnosisResult, RECOVERY_RESULT);
+ String recoveryResult =
+ formatRecoveryEvent(
+ "已结束",
+ taskId,
+ true,
+ "成功",
+ java.time.Instant.now().toString(), input);
log.info(
"[SPN-Domain-Agent] Recovery result reported via Notification-T: {}",
recoveryResult);
- pushNotificationResult(recoveryResult);
- return recoveryResult;
+ pushRecoveryResult(recoveryResult);
+ return;
+ }
+ log.info(
+ "[SPN-Domain-Agent] Recovery plan reported but automatic execution was not started because it is outside the active whitelist");
+ }
+
+ private static String formatRecoveryEvent(
+ String state,
+ String taskId,
+ boolean authorized,
+ String executionResult,
+ String finishedAt, SpnTaskInput input) {
+ StringBuilder event =
+ new StringBuilder("### 业务抢通事件\n")
+ .append("1. 业务抢通方案执行状态：").append(state).append('\n')
+ .append("2. 投诉诊断任务流水号：").append(taskId).append('\n')
+ .append("3. OSS侧事件流水号：").append(input.incidentId()).append('\n')
+ .append("4. 接入端口名称：").append(input.accessPort()).append('\n')
+ .append("5. 是否已授权OMC自动抢通：").append(authorized ? "是" : "否").append('\n')
+ .append("6. 业务抢通方案名称：隧道调优\n")
+ .append("7. 业务抢通方案详情：").append(RECOVERY_DETAIL);
+ if (finishedAt != null) {
+ event.append('\n').append("8. 业务抢通方案执行结束时间：").append(finishedAt);
+ }
+ if (executionResult != null) {
+ event.append('\n').append("9. 业务抢通方案执行结果：").append(executionResult);
}
- log.info("[SPN-Domain-Agent] Fault not in whitelist, refusing recovery");
- String refusalResult = "操作不在白名单内，拒绝执行抢通。";
- pushNotificationResult(refusalResult);
- return refusalResult;
+ return event.toString();
}
@Override
- protected String defaultNegotiationText() {
- return "城市1OMC诊断需要确认客户专线故障的详细端口信息后再执行，请补充。";
+ protected java.util.List invalidTaskFields(Map data) {
+ var invalid = new java.util.ArrayList<>(super.invalidTaskFields(data));
+ // Sample inventory: City1 must not diagnose a foreign or unknown access port.
+ if (!ACCESS_PORT.equals(SpnTaskInput.field(data, "任务对象", "接入端口名称"))
+ && !invalid.contains("任务对象")) invalid.add("任务对象");
+ return invalid;
}
@Override
@@ -129,7 +164,7 @@ protected Map buildResponseMetadata(RequestContext ctx, String r
@Override
protected String buildResultSummary() {
- return "SPN专线故障诊断结果";
+ return "专线业务投诉诊断任务诊断结果消息";
}
@Override
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity2Executor.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity2Executor.java
index 5eecf2c8..b9e4c7fd 100644
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity2Executor.java
+++ b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnDomainAgentCity2Executor.java
@@ -19,7 +19,8 @@
package dev.openan.workflow.engine.examples.agents;
-import dev.openan.workflow.engine.examples.LlmHelper;
+import dev.openan.workflow.engine.examples.extension.PrePositionedExtensionHandler;
+import dev.openan.workflow.engine.examples.util.LlmHelper;
import org.a2aproject.sdk.server.agentexecution.RequestContext;
import org.a2aproject.sdk.server.tasks.AgentEmitter;
import org.slf4j.Logger;
@@ -28,37 +29,51 @@
import java.util.LinkedHashMap;
import java.util.Map;
+import dev.openan.workflow.engine.examples.negotiation.NegotiationUtils;
+import dev.openan.workflow.engine.examples.util.EnvResolver;
/**
* SPN Domain Agent for City2 (City2 OMC).
*
* Server-side negotiation-capable (extends {@link NegotiationBaseAgentExecutor}). City2 side is
- * NORMAL. Diagnosis/recovery text is LLM-generated when the A2A-T .env is configured, else
- * deterministic. Authorization-T and Notification-T are pre-positioned before the workflow starts.
+ * NORMAL. Diagnosis text is LLM-generated when the A2A-T .env is configured, else deterministic.
+ * Authorization-T and Notification-T are initiated on independent channels.
*/
public class SpnDomainAgentCity2Executor extends NegotiationBaseAgentExecutor {
private static final Logger log = LoggerFactory.getLogger(SpnDomainAgentCity2Executor.class);
+ // Diagnosis result placed in artifact.metadata[Task-T URI], mirroring spec case 7.1 (normal).
private static final String NORMAL_DIAGNOSIS_RESULT =
- "诊断结果：城市2城市OMC诊断结果 - 端口状态正常，光功率-17dBm(正常范围)，无异常告警。\n"
- + "修复建议：城市2城市无需修复，故障不在此地市。\n"
- + "故障根因：无根因(此地市正常)。";
+ "1. 诊断结果：成功\n"
+ + "2. 诊断结果详情：城市2OMC端口状态正常，光功率-17dBm(正常范围)，无异常告警，故障不在此地市\n"
+ + "3. 修复建议：城市2无需修复";
- private static final String RECOVERY_RESULT = "城市2侧OMC抢通完成，业务恢复正常。";
+ public SpnDomainAgentCity2Executor() {
+ super();
+ }
- private static String llmDiagnosisResult(String input, String fallback) {
- String env = EnvResolver.resolveEnvPath();
- String sys =
- "你是SPN领域城市2OMC故障诊断专家。根据输入诊断信息，城市2侧端口正常、光功率-17dBm(正常范围)、无异常告警。输出诊断结果：城市2无需修复、故障不在此地市。简洁专业，中文。";
- String user = "输入：\n" + input + "\n\n城市2OMC端口port-3=UP，光功率-17dBm，无告警。请输出诊断结论。";
- return LlmHelper.text(env, sys, user, fallback);
+ public SpnDomainAgentCity2Executor(PrePositionedExtensionHandler extensionHandler) {
+ super(extensionHandler);
}
- private static void sleepBriefly() {
- try {
- Thread.sleep(300);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ private static String llmDiagnosisResult(String input, String fallback) {
+ String env = EnvResolver.resolveEnvPath();
+ String sys = "你是SPN领域OMC故障诊断专家。诊断结果表示诊断任务是否成功完成，不表示是否发现故障。"
+ + "本例必须输出“1. 诊断结果：成功”；城市2本地端口正常、无故障，不得写成诊断失败，"
+ + "不得把城市2的任务对象称为城市1。继续输出2.诊断结果详情、3.修复建议，限300个中文字符，确保句子完整。";
+ String user = "输入：\n" + input
+ + "\n\n城市2 OMC 已核验本次任务对象：端口状态UP，光功率-17dBm，无告警。"
+ + "结论是城市2侧诊断执行成功且未发现本地故障。请按指定结构输出。";
+ String result = LlmHelper.text(env, sys, user, fallback);
+ String normalized = result.replace("*", "").replace("#", "");
+ boolean successfulDiagnosis =
+ normalized.matches("(?s).*诊断结果\\s*[：:]?\\s*成功.*")
+ && (normalized.contains("正常") || normalized.contains("未发现"));
+ if (!successfulDiagnosis) {
+ log.warn(
+ "[SPN-Domain-Agent-City2] LLM output contradicted the verified normal state; using deterministic result");
+ return fallback;
}
+ return result;
}
@Override
@@ -67,16 +82,20 @@ protected String resolveEnvPath() {
}
@Override
- protected String executeBusiness(RequestContext ctx, AgentEmitter emitter, String input) {
- String result = llmDiagnosisResult(input, NORMAL_DIAGNOSIS_RESULT);
+ protected String executeBusiness(RequestContext ctx, AgentEmitter emitter, SpnTaskInput input) {
+ String result = llmDiagnosisResult(input.diagnosisInput(), NORMAL_DIAGNOSIS_RESULT);
log.info(
"[SPN-Domain-Agent-City2] Diagnosis complete (City2), no fault, no recovery needed");
return result;
}
@Override
- protected String defaultNegotiationText() {
- return "城市2OMC诊断需要确认客户专线故障是否涉及城市2侧端口，请补充。";
+ protected java.util.List invalidTaskFields(Map data) {
+ var invalid = new java.util.ArrayList<>(super.invalidTaskFields(data));
+ // Sample inventory belongs to City2; negotiation must never reuse City1's input.
+ if (!"P882-珠江新城-PTN7900-23-TPA1EG24-11".equals(SpnTaskInput.field(data, "任务对象", "接入端口名称"))
+ && !invalid.contains("任务对象")) invalid.add("任务对象");
+ return invalid;
}
@Override
@@ -88,7 +107,7 @@ protected Map buildResponseMetadata(RequestContext ctx, String r
@Override
protected String buildResultSummary() {
- return "SPN专线故障诊断结果";
+ return "专线业务投诉诊断任务诊断结果消息";
}
@Override
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnTaskInput.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnTaskInput.java
new file mode 100644
index 00000000..304eb39b
--- /dev/null
+++ b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/SpnTaskInput.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright (c) 2026 Huawei Technologies Co., Ltd.
+ * All Rights Reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package dev.openan.workflow.engine.examples.agents;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+/** Sample business input, constructed only from SDK-validated task or negotiation fields. */
+public record SpnTaskInput(Map parameters) {
+ public SpnTaskInput {
+ parameters = Map.copyOf(parameters);
+ List invalid = invalidFields(parameters);
+ if (!invalid.isEmpty()) throw new IllegalArgumentException("Incomplete complaint parameters: " + invalid);
+ }
+
+ /** Fields requiring clarification; this is SPN business validation, not an engine constraint. */
+ public static List invalidFields(Map parameters) {
+ List fields = new ArrayList<>();
+ if (field(parameters, "任务对象", "接入端口名称").isBlank()) fields.add("任务对象");
+ if (field(parameters, "任务上下文", "投诉分类").isBlank()
+ || field(parameters, "任务上下文", "OSS侧事件流水号").isBlank()) fields.add("任务上下文");
+ return fields;
+ }
+
+ /** The access port belonging to this validated city task. */
+ public String accessPort() { return field(parameters, "任务对象", "接入端口名称"); }
+
+ /** The original business incident identifier, not the A2A task identifier. */
+ public String incidentId() { return field(parameters, "任务上下文", "OSS侧事件流水号"); }
+
+ /** Business-only diagnostic input; never concatenate raw protocol or unvalidated replies here. */
+ public String diagnosisInput() {
+ return "已校验投诉参数\n任务对象：" + parameters.get("任务对象")
+ + "\n任务上下文：" + parameters.get("任务上下文");
+ }
+
+ static String field(Map parameters, String section, String name) {
+ Object raw = parameters.get(section);
+ if (!(raw instanceof String text)) return "";
+ var match = Pattern.compile(Pattern.quote(name) + "[：:]\\s*[\\\"“]?([^；;\\\"”\\r\\n]+)").matcher(text);
+ String value = match.find() ? match.group(1).strip() : "";
+ return "null".equalsIgnoreCase(value) ? "" : value;
+ }
+
+ static Map selected(Map values) {
+ Map result = new LinkedHashMap<>();
+ for (String field : List.of("任务对象", "任务上下文")) {
+ Object value = values.get(field);
+ if (value instanceof String text) result.put(field, text);
+ }
+ return Map.copyOf(result);
+ }
+}
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/TransportWorkbenchAgentExecutor.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/TransportWorkbenchAgentExecutor.java
deleted file mode 100644
index 06f47d0a..00000000
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/TransportWorkbenchAgentExecutor.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License. You may obtain
- * a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-
-package dev.openan.workflow.engine.examples.agents;
-
-import org.a2aproject.sdk.server.agentexecution.RequestContext;
-import org.a2aproject.sdk.server.tasks.AgentEmitter;
-import org.a2aproject.sdk.spec.A2AError;
-import org.a2aproject.sdk.spec.Part;
-import org.a2aproject.sdk.spec.TextPart;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Transport Workbench Agent - the orchestrator.
- *
- * Single responsibility: agent server I/O. Receives a Task-T from the upper layer,
- * delegates the full orchestration pipeline to {@link WorkbenchOrchestrator}, and returns the
- * result. This class does NOT contain workflow logic, pre-positioning, or negotiation strategy --
- * those live in dedicated classes.
- */
-public class TransportWorkbenchAgentExecutor extends BaseAgentExecutor {
- private static final Logger log =
- LoggerFactory.getLogger(TransportWorkbenchAgentExecutor.class);
-
- private final String orchUrl;
- private final String credentialsPath;
- private final boolean sslVerify;
- private final String a2atEnvPath;
-
- public TransportWorkbenchAgentExecutor(
- String registryUrl, String orchUrl, String credentialsPath, boolean sslVerify) {
- this(registryUrl, orchUrl, credentialsPath, sslVerify, null);
- }
-
- public TransportWorkbenchAgentExecutor(
- String registryUrl,
- String orchUrl,
- String credentialsPath,
- boolean sslVerify,
- String a2atEnvPath) {
- this.orchUrl = orchUrl;
- this.credentialsPath = credentialsPath;
- this.sslVerify = sslVerify;
- this.a2atEnvPath = a2atEnvPath;
- }
-
- @Override
- public void execute(RequestContext ctx, AgentEmitter emitter) throws A2AError {
- String taskId = ctx.getTaskId();
- String contextId = ctx.getContextId();
- String input = extractText(ctx.getMessage());
- log.info("[Workbench] Received task: taskId={}, text={} chars", taskId, input.length());
- emitter.submit(buildStatusMessage(contextId, taskId, "Task received"));
- emitter.startWork(buildStatusMessage(contextId, taskId, "Processing"));
- try {
- String result =
- new WorkbenchOrchestrator(orchUrl, credentialsPath, sslVerify, a2atEnvPath)
- .run(input);
- Map metadata = new LinkedHashMap<>();
- metadata.put(NegotiationUtils.TASK_PROMPT_KEY, result);
- List> parts = List.of(new TextPart("跨城故障协同诊断汇总结果"));
- emitter.addArtifact(
- parts, "result", "cross-city-diagnosis-summary", metadata, false, true);
- emitter.complete(buildStatusMessage(contextId, taskId, "Completed"));
- log.info("[Workbench] Task completed");
- } catch (Exception e) {
- log.error("[Workbench] Failed: {}", e.getMessage(), e);
- emitter.fail(buildStatusMessage(contextId, taskId, "Failed: " + e.getMessage()));
- }
- }
-
- @Override
- public void cancel(RequestContext ctx, AgentEmitter emitter) throws A2AError {
- emitter.cancel();
- }
-}
diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/WorkbenchControlPoint.java b/samples/src/main/java/dev/openan/workflow/engine/examples/agents/WorkbenchControlPoint.java
deleted file mode 100644
index 49a5b5a4..00000000
--- a/samples/src/main/java/dev/openan/workflow/engine/examples/agents/WorkbenchControlPoint.java
+++ /dev/null
@@ -1,191 +0,0 @@
-/*
- * Copyright (c) 2026 Huawei Technologies Co., Ltd.
- * All Rights Reserved.
- *
- * SPDX-License-Identifier: Apache-2.0
- *
- * Licensed under the Apache License, Version 2.0 (the License); you may
- * not use this file except in compliance with the License. You may obtain
- * a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an AS IS BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-
-package dev.openan.workflow.engine.examples.agents;
-
-import dev.openan.workflow.engine.examples.LlmHelper;
-import dev.openan.workflow.engine.client.WorkflowEngineClient;
-import dev.openan.workflow.engine.control.DefaultControlPoint;
-import dev.openan.workflow.engine.model.JumpCondition;
-import dev.openan.workflow.engine.model.RouteDecision;
-import dev.openan.workflow.engine.model.TaskRequest;
-import dev.openan.workflow.engine.model.TaskResponse;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-
-/**
- * ControlPoint for the SPN cross-city diagnosis workflow.
- *
- *