diff --git a/config/checkstyle/checkstyle-suppressions.xml b/config/checkstyle/checkstyle-suppressions.xml index c598199..2a015bb 100644 --- a/config/checkstyle/checkstyle-suppressions.xml +++ b/config/checkstyle/checkstyle-suppressions.xml @@ -19,4 +19,19 @@ + + + + + + + + + diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositioner.java b/samples/src/main/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositioner.java index a851f4e..447cb3c 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositioner.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositioner.java @@ -11,7 +11,7 @@ * 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 + * 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. @@ -23,8 +23,12 @@ import dev.openan.workflow.engine.client.NotificationSubscription; import dev.openan.workflow.engine.examples.demo.SpnCasePrompts; import dev.openan.workflow.engine.model.SendMessageResult; +import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import net.openan.a2at.sdk.core.model.StandardTemplates; @@ -38,11 +42,17 @@ *

Single responsibility: send a one-shot whitelist operation and open a long-lived result * subscription for each non-workbench agent on separate channels. Individual workflow tasks do not * own these operations. + * + *

Agents are pre-positioned in parallel: each agent's Authorization-T + Notification-T sequence + * runs on its own worker, because every operation is LLM-bound (prompt generation plus + * agent-side validation) and the sequences of different agents are independent. Per-agent failures + * are counted and skipped exactly as before; they never abort other agents. */ public class ExtensionPrePositioner { private static final Logger log = LoggerFactory.getLogger(ExtensionPrePositioner.class); private static final int AUTHORIZATION_VALIDATION_ATTEMPTS = 3; + private static final int MAX_PARALLEL_AGENTS = 4; private final net.openan.a2at.sdk.client.A2ATClient contentClient; private final Map authData; @@ -92,6 +102,12 @@ private static void requireState( } } + /** Per-agent result of one parallel pre-positioning run. */ + private record AgentOutcome( + List subscriptions, + int authorizationFailures, + int notificationFailures) {} + /** * Pre-position Authorization-T + Notification-T to every non-workbench agent. * @@ -123,104 +139,152 @@ public List prePosition( notificationCallback, Consumer subscriptionOpened) { long allStarted = System.nanoTime(); - int targetCount = 0; - for (AgentCard card : agentCards) { - if (!card.name().contains("Workbench")) { - targetCount++; - } - } + List targets = + agentCards.stream().filter(card -> !card.name().contains("Workbench")).toList(); log.info( - "[PrePosition] START targetAgents={}, notificationCallback={}", - targetCount, + "[PrePosition] START targetAgents={}, parallel={}, notificationCallback={}", + targets.size(), + targets.size() > 1, notificationCallback != null); - java.util.ArrayList subscriptions = new java.util.ArrayList<>(); + if (targets.isEmpty()) { + log.info( + "[PrePosition] DONE targetAgents=0, activeSubscriptions=0, " + + "authorizationFailures=0, notificationFailures=0, elapsedMs={}", + elapsedMillis(allStarted)); + return List.of(); + } + + ExecutorService executor = + Executors.newFixedThreadPool( + Math.min(targets.size(), MAX_PARALLEL_AGENTS), + runnable -> { + Thread thread = new Thread(runnable, "preposition"); + thread.setDaemon(true); + return thread; + }); + try { + List> futures = + targets.stream() + .map( + card -> + CompletableFuture.supplyAsync( + () -> + prePositionAgent( + card, + authorizationSender, + notificationSender, + notificationCallback, + subscriptionOpened), + executor)) + .toList(); + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); + + ArrayList subscriptions = new ArrayList<>(); + int authorizationFailures = 0; + int notificationFailures = 0; + for (CompletableFuture future : futures) { + AgentOutcome outcome = future.join(); + subscriptions.addAll(outcome.subscriptions()); + authorizationFailures += outcome.authorizationFailures(); + notificationFailures += outcome.notificationFailures(); + } + log.info( + "[PrePosition] DONE targetAgents={}, activeSubscriptions={}, " + + "authorizationFailures={}, notificationFailures={}, elapsedMs={}", + targets.size(), + subscriptions.size(), + authorizationFailures, + notificationFailures, + elapsedMillis(allStarted)); + return List.copyOf(subscriptions); + } finally { + executor.shutdownNow(); + } + } + + /** Runs one agent's Authorization-T + Notification-T sequence; failures are counted, not thrown. */ + private AgentOutcome prePositionAgent( + AgentCard card, + ExtensionSender authorizationSender, + ExtensionSender notificationSender, + java.util.function.BiConsumer< + NotificationSubscription, dev.openan.workflow.engine.model.ReceivedMessage> + notificationCallback, + Consumer subscriptionOpened) { + String name = card.name(); + long agentStarted = System.nanoTime(); + ArrayList subscriptions = new ArrayList<>(); int authorizationFailures = 0; int notificationFailures = 0; - for (AgentCard card : agentCards) { - String name = card.name(); - if (name.contains("Workbench")) { - continue; - } - long agentStarted = System.nanoTime(); - long operationStarted = System.nanoTime(); - try { - log.info("[PrePosition] SEND extension=Authorization-T, agent={}", name); - SendMessageResult authResult = sendAuthorization(authorizationSender, name); - log.info( - "[PrePosition] ACK extension=Authorization-T, agent={}, state={}, " - + "responseChars={}, elapsedMs={}", - name, - authResult.getTaskState(), - authResult.getText() != null ? authResult.getText().length() : 0, - elapsedMillis(operationStarted)); - requireState(authResult, "Authorization-T", name, "TASK_STATE_COMPLETED"); - } catch (RuntimeException e) { - authorizationFailures++; - log.warn( - "[PrePosition] OPERATION_FAILED extension=Authorization-T, agent={}, " - + "elapsedMs={}, errorType={}, message={}, action=continue", - name, - elapsedMillis(operationStarted), - e.getClass().getSimpleName(), - e.getMessage()); - } + long operationStarted = System.nanoTime(); + try { + log.info("[PrePosition] SEND extension=Authorization-T, agent={}", name); + SendMessageResult authResult = sendAuthorization(authorizationSender, name); + log.info( + "[PrePosition] ACK extension=Authorization-T, agent={}, state={}, " + + "responseChars={}, elapsedMs={}", + name, + authResult.getTaskState(), + authResult.getText() != null ? authResult.getText().length() : 0, + elapsedMillis(operationStarted)); + requireState(authResult, "Authorization-T", name, "TASK_STATE_COMPLETED"); + } catch (RuntimeException e) { + authorizationFailures++; + log.warn( + "[PrePosition] OPERATION_FAILED extension=Authorization-T, agent={}, " + + "elapsedMs={}, errorType={}, message={}, action=continue", + name, + elapsedMillis(operationStarted), + e.getClass().getSimpleName(), + e.getMessage()); + } - operationStarted = System.nanoTime(); - NotificationSubscription subscription = null; - try { - log.info("[PrePosition] SEND extension=Notification-T, agent={}", name); - var notificationContent = - contentClient.generateNotificationPromptFromDataWithSchema( - notifData, notifSchema, StandardTemplates.SERVICE_RECOVERY.uri()); - subscription = - notificationSender.openNotification( - name, - dev.openan.workflow.engine.client.A2atMessages.from( - notificationContent, List.of(new org.a2aproject.sdk.spec.TextPart("订阅业务抢通事件"))), - notificationCallback == null ? (handle, event) -> {} : notificationCallback); - if (subscriptionOpened != null && subscription.isActive()) { - subscriptionOpened.accept(subscription); - } - var notificationResult = subscription.acknowledgement().join(); - requireState( - notificationResult, - "Notification-T", - name, - "TASK_STATE_WORKING", - "TASK_STATE_COMPLETED"); - subscriptions.add(subscription); - log.info( - "[PrePosition] ACK extension=Notification-T, agent={}, state={}, " - + "responseChars={}, elapsedMs={}", - name, - notificationResult.getTaskState(), - notificationResult.getText() != null ? notificationResult.getText().length() : 0, - elapsedMillis(operationStarted)); - log.info( - "[PrePosition] AGENT_DONE agent={}, elapsedMs={}", name, elapsedMillis(agentStarted)); - } catch (RuntimeException e) { - notificationFailures++; - if (subscription != null) { - subscription.close(); - } - log.warn( - "[PrePosition] OPERATION_FAILED extension=Notification-T, agent={}, " - + "elapsedMs={}, errorType={}, message={}, action=continue", - name, - elapsedMillis(operationStarted), - e.getClass().getSimpleName(), - e.getMessage()); + operationStarted = System.nanoTime(); + NotificationSubscription subscription = null; + try { + log.info("[PrePosition] SEND extension=Notification-T, agent={}", name); + var notificationContent = + contentClient.generateNotificationPromptFromDataWithSchema( + notifData, notifSchema, StandardTemplates.SERVICE_RECOVERY.uri()); + subscription = + notificationSender.openNotification( + name, + dev.openan.workflow.engine.client.A2atMessages.from( + notificationContent, List.of(new org.a2aproject.sdk.spec.TextPart("订阅业务抢通事件"))), + notificationCallback == null ? (handle, event) -> {} : notificationCallback); + if (subscriptionOpened != null && subscription.isActive()) { + subscriptionOpened.accept(subscription); } + var notificationResult = subscription.acknowledgement().join(); + requireState( + notificationResult, + "Notification-T", + name, + "TASK_STATE_WORKING", + "TASK_STATE_COMPLETED"); + subscriptions.add(subscription); + log.info( + "[PrePosition] ACK extension=Notification-T, agent={}, state={}, " + + "responseChars={}, elapsedMs={}", + name, + notificationResult.getTaskState(), + notificationResult.getText() != null ? notificationResult.getText().length() : 0, + elapsedMillis(operationStarted)); + log.info("[PrePosition] AGENT_DONE agent={}, elapsedMs={}", name, elapsedMillis(agentStarted)); + } catch (RuntimeException e) { + notificationFailures++; + if (subscription != null) { + subscription.close(); + } + log.warn( + "[PrePosition] OPERATION_FAILED extension=Notification-T, agent={}, " + + "elapsedMs={}, errorType={}, message={}, action=continue", + name, + elapsedMillis(operationStarted), + e.getClass().getSimpleName(), + e.getMessage()); } - log.info( - "[PrePosition] DONE targetAgents={}, activeSubscriptions={}, " - + "authorizationFailures={}, notificationFailures={}, elapsedMs={}", - targetCount, - subscriptions.size(), - authorizationFailures, - notificationFailures, - elapsedMillis(allStarted)); - return List.copyOf(subscriptions); + return new AgentOutcome(subscriptions, authorizationFailures, notificationFailures); } private SendMessageResult sendAuthorization(ExtensionSender sender, String agentName) { diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositionerTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositionerTest.java index 7d53fa4..5a57758 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositionerTest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/extension/ExtensionPrePositionerTest.java @@ -47,6 +47,14 @@ private static org.a2aproject.sdk.spec.AgentCard city1Card() { .orElseThrow(); } + private static org.a2aproject.sdk.spec.AgentCard city2Card() { + return new WorkbenchAgentCatalog() + .load().stream() + .filter(card -> "SPN Domain Agent City2".equals(card.name())) + .findFirst() + .orElseThrow(); + } + private static NotificationSubscription newSubscription(String agentName) throws Exception { Constructor constructor = NotificationSubscription.class.getDeclaredConstructor( @@ -170,6 +178,56 @@ void authorizationFailureDoesNotBlockSubscriptionOrWorkflowStartup() throws Exce assertSame(subscription, registeredSubscription.get()); } + @Test + void agentsArePositionedInParallel() throws Exception { + // Both agents must be inside openNotification concurrently: each call waits on the + // barrier, so a serial implementation can never satisfy it and fails via the + // resulting IllegalStateException (counted as a notification failure). + CountDownLatch bothEntered = new CountDownLatch(2); + NotificationSubscription city1Subscription = newSubscription("SPN Domain Agent City1"); + NotificationSubscription city2Subscription = newSubscription("SPN Domain Agent City2"); + ExtensionSender authorizationSender = + sender( + (method, args) -> { + if ("sendAuthorization".equals(method.getName())) { + return CompletableFuture.completedFuture( + SendMessageResult.builder() + .taskState("TASK_STATE_COMPLETED") + .text("authorized") + .build()); + } + throw new UnsupportedOperationException(method.getName()); + }); + ExtensionSender notificationSender = + sender( + (method, args) -> { + if ("openNotification".equals(method.getName())) { + bothEntered.countDown(); + if (!bothEntered.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException( + "agents are not pre-positioned in parallel: only one entered openNotification"); + } + NotificationSubscription sub = + "SPN Domain Agent City1".equals(args[0]) + ? city1Subscription + : city2Subscription; + acknowledge(sub, "TASK_STATE_WORKING"); + return sub; + } + throw new UnsupportedOperationException(method.getName()); + }); + + List subscriptions = + new ExtensionPrePositioner() + .prePosition( + authorizationSender, + notificationSender, + List.of(city1Card(), city2Card()), + (handle, received) -> {}); + + assertEquals(2, subscriptions.size()); + } + @FunctionalInterface private interface Invocation { Object invoke(Method method, Object[] args) throws Throwable;