From b81355bd9c1d257d358c19a265c21ca943f5a822 Mon Sep 17 00:00:00 2001 From: konlue <2362586639@qq.com> Date: Wed, 4 Nov 2026 13:17:05 +0800 Subject: [PATCH 1/2] feat(agent): control model retry attempts with a deadline-aware policy (issue #313) The retry sequence of a model call was not bounded by its run. The retryability of a failure was guessed from the text of the exception message, the deadline of the run was ignored, and a cancelled run kept receiving provider calls because the loop observed thread interruption only. The decision now belongs to a neutral contract in argi-agent-api (io.github.agentic.ai.agent.retry) that classifies the failure by type and cause chain, bounds the number of attempts and observes the deadline and the cancellation of the run. The run guard reaches the interceptor through an optional ExecutionGuard on ModelRequest and is propagated by the independent ReAct engine and by the framework node, so a cancelled or expired run ends the backoff instead of dispatching another call. The existing configuration keeps its behaviour and every existing builder option keeps its default value; the type based classifier, the retryable categories, the reserved deadline time and the clock are opt in. --- .../ai/agent/interceptor/ModelRequest.java | 43 ++- .../ai/agent/retry/ModelCallFailure.java | 43 +++ .../agent/retry/ModelCallFailureCategory.java | 51 ++++ .../retry/ModelCallFailureClassifier.java | 33 +-- .../ai/agent/retry/ModelRetryController.java | 226 ++++++++++++++++ .../agentic/ai/agent/retry/RetryDecision.java | 121 +++++++++ .../agentic/ai/agent/retry/RetryPolicy.java | 78 ++++++ .../ai/agent/retry/RetryStopReason.java | 25 +- .../TypeBasedModelCallFailureClassifier.java | 75 ++++++ .../agent/retry/ModelRetryControllerTest.java | 223 ++++++++++++++++ .../agentic/ai/agent/react/ReActAgent.java | 6 +- .../react/ReActAgentExecutionGuardTest.java | 146 ++++++++++ .../modelretry/ModelRetryInterceptor.java | 251 ++++++++++++++---- .../orchestration/node/AgentLlmNode.java | 3 +- .../ModelRetryInterceptorControlledTest.java | 223 ++++++++++++++++ .../node/AgentLlmNodeExecutionGuardTest.java | 116 ++++++++ docs/architecture/controlled-model-retry.md | 107 ++++++++ 17 files changed, 1696 insertions(+), 74 deletions(-) create mode 100644 argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailure.java create mode 100644 argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureCategory.java rename argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculatorTest.java => argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureClassifier.java (50%) create mode 100644 argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelRetryController.java create mode 100644 argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryDecision.java create mode 100644 argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryPolicy.java rename argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculator.java => argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryStopReason.java (56%) create mode 100644 argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/TypeBasedModelCallFailureClassifier.java create mode 100644 argi-agent/argi-agent-api/src/test/java/io/github/agentic/ai/agent/retry/ModelRetryControllerTest.java create mode 100644 argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentExecutionGuardTest.java create mode 100644 argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java create mode 100644 argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNodeExecutionGuardTest.java create mode 100644 docs/architecture/controlled-model-retry.md diff --git a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/interceptor/ModelRequest.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/interceptor/ModelRequest.java index 0403b63ca..d0d346828 100644 --- a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/interceptor/ModelRequest.java +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/interceptor/ModelRequest.java @@ -20,6 +20,7 @@ import io.github.agentic.ai.api.chat.ChatRole; import io.github.agentic.ai.api.chat.GenerationOptions; import io.github.agentic.ai.api.tool.ToolRegistry; +import io.github.agentic.ai.runtime.execution.ExecutionGuard; import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; import java.util.ArrayList; @@ -62,6 +63,12 @@ public class ModelRequest { // context map. private final TrustedExecutionContext trustedExecutionContext; + // The guard of the run that owns this request. It carries the liveness and the + // deadline of the run to the interceptors, so that a retry sequence can be + // bounded by the run itself. It never leaves the process: the request attributes + // are passed to the provider and must stay free of runtime objects. + private final ExecutionGuard executionGuard; + public ModelRequest(ChatMessage systemMessage, List messages, GenerationOptions options, ChatClient chatClient, ToolRegistry toolRegistry, List tools, boolean toolSelectionSpecified, ToolRegistry dynamicToolRegistry, Map toolDescriptions, Map context, @@ -74,6 +81,15 @@ public ModelRequest(ChatMessage systemMessage, List messages, Gener ChatClient chatClient, ToolRegistry toolRegistry, List tools, boolean toolSelectionSpecified, ToolRegistry dynamicToolRegistry, Map toolDescriptions, Map context, Map attributes, TrustedExecutionContext trustedExecutionContext) { + this(systemMessage, messages, options, chatClient, toolRegistry, tools, toolSelectionSpecified, + dynamicToolRegistry, toolDescriptions, context, attributes, trustedExecutionContext, null); + } + + public ModelRequest(ChatMessage systemMessage, List messages, GenerationOptions options, + ChatClient chatClient, ToolRegistry toolRegistry, List tools, boolean toolSelectionSpecified, + ToolRegistry dynamicToolRegistry, Map toolDescriptions, Map context, + Map attributes, TrustedExecutionContext trustedExecutionContext, + ExecutionGuard executionGuard) { if (systemMessage != null && systemMessage.role() != ChatRole.SYSTEM) { throw new IllegalArgumentException("systemMessage must have the SYSTEM role"); } @@ -90,6 +106,7 @@ public ModelRequest(ChatMessage systemMessage, List messages, Gener this.context = new HashMap<>(context); this.attributes = Map.copyOf(attributes); this.trustedExecutionContext = trustedExecutionContext; + this.executionGuard = executionGuard; } public static Builder builder() { @@ -107,7 +124,8 @@ public static Builder builder(ModelRequest request) { .toolDescriptions(request.toolDescriptions) .context(request.context) .attributes(request.attributes) - .trustedExecutionContext(request.trustedExecutionContext); + .trustedExecutionContext(request.trustedExecutionContext) + .executionGuard(request.executionGuard); copy.toolSelectionSpecified = request.toolSelectionSpecified; return copy; } @@ -167,6 +185,16 @@ public Optional getTrustedExecutionContext() { return Optional.ofNullable(trustedExecutionContext); } + /** + * The guard of the run that issued this model call, when the run carries one. The + * guard reports the liveness and the deadline of the run, so that an interceptor + * that repeats the call can stop the sequence when the run is cancelled or out of + * time. It is never passed to the provider. + */ + public Optional getExecutionGuard() { + return Optional.ofNullable(executionGuard); + } + public Map getToolDescriptions() { return toolDescriptions; } @@ -209,6 +237,8 @@ public static class Builder { private TrustedExecutionContext trustedExecutionContext; + private ExecutionGuard executionGuard; + public Builder systemMessage(ChatMessage systemMessage) { this.systemMessage = systemMessage; return this; @@ -277,10 +307,19 @@ public Builder trustedExecutionContext(TrustedExecutionContext trustedExecutionC return this; } + /** + * Sets the guard of the run that owns this model call. The guard is propagated + * by the engine and stays inside the process. + */ + public Builder executionGuard(ExecutionGuard executionGuard) { + this.executionGuard = executionGuard; + return this; + } + public ModelRequest build() { return new ModelRequest(systemMessage, messages, options, chatClient, toolRegistry, tools, toolSelectionSpecified, dynamicToolRegistry, toolDescriptions, context, attributes, - trustedExecutionContext); + trustedExecutionContext, executionGuard); } } diff --git a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailure.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailure.java new file mode 100644 index 000000000..f130e28f2 --- /dev/null +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailure.java @@ -0,0 +1,43 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.retry; + +import java.util.Objects; +import java.util.Set; + +/** + * A classified model call failure. + * + * @param category the classification of the failure + * @param cause the original failure, may be {@code null} when the failure was not + * classified from a throwable + */ +public record ModelCallFailure(ModelCallFailureCategory category, Throwable cause) { + + public ModelCallFailure { + Objects.requireNonNull(category, "category must not be null"); + } + + /** + * Whether the failure may be retried under the given policy. + * @param retryableCategories the categories the policy allows to retry + * @return {@code true} when the category is retryable + */ + public boolean isRetryable(Set retryableCategories) { + return retryableCategories.contains(category); + } + +} diff --git a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureCategory.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureCategory.java new file mode 100644 index 000000000..ef80a8a7e --- /dev/null +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureCategory.java @@ -0,0 +1,51 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.retry; + +/** + * Classification of a failed model call. + * + *

+ * The category is derived from the failure itself, never from the text of an + * exception message: a permanent rejection that happens to mention a timeout must not + * be retried. + */ +public enum ModelCallFailureCategory { + + /** A temporary failure that a later attempt may still overcome. */ + TRANSIENT, + + /** The provider asked the caller to slow down. */ + RATE_LIMITED, + + /** The request itself was rejected; another attempt cannot change the outcome. */ + PERMANENT, + + /** The call was cancelled or interrupted and must never be retried. */ + CANCELLED, + + /** The failure could not be classified and is therefore not retried by default. */ + UNKNOWN; + + /** + * Whether this category is retryable without further configuration. + * @return {@code true} for {@link #TRANSIENT} and {@link #RATE_LIMITED} + */ + public boolean isRetryableByDefault() { + return this == TRANSIENT || this == RATE_LIMITED; + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculatorTest.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureClassifier.java similarity index 50% rename from argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculatorTest.java rename to argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureClassifier.java index c6a56837a..a18c847f8 100644 --- a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculatorTest.java +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelCallFailureClassifier.java @@ -13,23 +13,24 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.github.agentic.ai.agent.orchestration.interceptor.modelretry; +package io.github.agentic.ai.agent.retry; -import org.junit.jupiter.api.Test; - -import static org.junit.jupiter.api.Assertions.assertEquals; - -class RetryDelayCalculatorTest { - - @Test - void exponentialDelayIsCappedWithoutReadingTheClock() { - assertEquals(150, RetryDelayCalculator.nextDelay(100, 3.0, 150)); - assertEquals(150, RetryDelayCalculator.nextDelay(150, 3.0, 150)); - } +/** + * Classifies a failed model call into a {@link ModelCallFailureCategory}. + * + *

+ * Implementations must classify from the failure type and its cause chain. Matching + * on the text of an exception message is not a classification: the same message can + * describe a permanent rejection or a transient outage. + */ +@FunctionalInterface +public interface ModelCallFailureClassifier { - @Test - void zeroDelayRemainsZero() { - assertEquals(0, RetryDelayCalculator.nextDelay(0, 2.0, 30_000)); - } + /** + * Classifies a failure. + * @param failure the failure raised by the model call + * @return the classified failure, never {@code null} + */ + ModelCallFailure classify(Throwable failure); } diff --git a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelRetryController.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelRetryController.java new file mode 100644 index 000000000..de6352d44 --- /dev/null +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/ModelRetryController.java @@ -0,0 +1,226 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.retry; + +import io.github.agentic.ai.runtime.execution.ExecutionGuard; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; + +/** + * Decides whether a model call may start another attempt. + * + *

+ * One controller belongs to one model call. It owns the attempt counter, the + * classification of the failures it observes and the deadline of the run, so that a + * retry sequence is bounded on every axis at once: the number of attempts, the + * retryability of the failure and the remaining time. + * + *

+ * The controller never waits. The caller owns the backoff, and must re-check the + * controller before every attempt, because a cancellation or a deadline can stop the + * run while the caller is waiting. + */ +public final class ModelRetryController { + + private final RetryPolicy policy; + + private final ModelCallFailureClassifier classifier; + + private final ExecutionGuard guard; + + private final Clock clock; + + private final Instant deadline; + + private int attempts; + + private volatile RetryStopReason stopReason; + + /** + * Creates a controller. + * @param policy the bounded retry configuration + * @param classifier the failure classifier + * @param guard the run guard, may be {@code null} for an uncontrolled run + * @param clock the clock used for the deadline checks + * @param deadline the deadline of the run, may be {@code null} when the run has + * none + */ + public ModelRetryController(RetryPolicy policy, ModelCallFailureClassifier classifier, ExecutionGuard guard, + Clock clock, Instant deadline) { + this.policy = Objects.requireNonNull(policy, "policy must not be null"); + this.classifier = Objects.requireNonNull(classifier, "classifier must not be null"); + this.guard = guard; + this.clock = Objects.requireNonNull(clock, "clock must not be null"); + this.deadline = deadline; + } + + /** + * Creates a controller that takes its deadline from the guard. + * @param policy the bounded retry configuration + * @param classifier the failure classifier + * @param guard the run guard, may be {@code null} for an uncontrolled run + * @param clock the clock used for the deadline checks + * @return the controller + */ + public static ModelRetryController forGuard(RetryPolicy policy, ModelCallFailureClassifier classifier, + ExecutionGuard guard, Clock clock) { + Instant deadline = guard == null ? null : guard.deadline().orElse(null); + return new ModelRetryController(policy, classifier, guard, clock, deadline); + } + + /** + * Checks whether the next attempt may start and counts it. + * @return {@link RetryDecision#allow()} when the attempt may start, otherwise the + * stop decision + */ + public RetryDecision beforeAttempt() { + if (stopReason != null) { + return RetryDecision.stop(stopReason); + } + if (guard != null) { + // A stopped run is not a retryable failure: the guard owns the reason. + guard.assertActive(); + } + RetryDecision deadlineDecision = checkDeadline(); + if (deadlineDecision.isStop()) { + return deadlineDecision; + } + attempts++; + return RetryDecision.allow(); + } + + /** + * Checks whether the deadline still allows another attempt, without counting an + * attempt. + * @return {@link RetryDecision#allow()} when the attempt may start, otherwise the + * stop decision + */ + public RetryDecision checkAttemptAllowed() { + if (stopReason != null) { + return RetryDecision.stop(stopReason); + } + if (guard != null) { + guard.assertActive(); + } + return checkDeadline(); + } + + /** + * Decides what happens after a failed attempt. + * @param failure the failure raised by the attempt + * @return the retry decision with its backoff, or the stop decision + */ + public RetryDecision onFailure(Throwable failure) { + return onFailure(failure, attempts); + } + + /** + * Decides what happens after a failed attempt of a sequence that counts its + * attempts itself, such as a streaming response whose subscriptions each own their + * attempt budget. + * @param failure the failure raised by the attempt + * @param attemptsMade the number of attempts already made by that sequence + * @return the retry decision with its backoff, or the stop decision + */ + public RetryDecision onFailure(Throwable failure, int attemptsMade) { + if (stopReason != null) { + return RetryDecision.stop(stopReason); + } + ModelCallFailure classified = classifier.classify(failure); + if (classified == null) { + classified = new ModelCallFailure(ModelCallFailureCategory.UNKNOWN, failure); + } + if (classified.category() == ModelCallFailureCategory.CANCELLED) { + return stop(RetryStopReason.CANCELLED); + } + if (!classified.isRetryable(policy.retryableCategories())) { + return stop(RetryStopReason.NON_RETRYABLE); + } + if (attemptsMade >= policy.maxAttempts()) { + return stop(RetryStopReason.ATTEMPTS_EXHAUSTED); + } + long delay = policy.delayMillisAfter(attemptsMade); + if (deadline != null && !clock.instant().plusMillis(delay).isBefore(deadline)) { + return stop(RetryStopReason.DEADLINE_EXCEEDED); + } + return RetryDecision.retryAfter(Duration.ofMillis(delay)); + } + + /** + * The number of attempts that were allowed to start. + * @return the attempt count + */ + public int attempts() { + return attempts; + } + + /** + * Why the sequence stopped. + * @return the stop reason, empty while the sequence may continue + */ + public Optional stopReason() { + return Optional.ofNullable(stopReason); + } + + /** + * The policy of this controller. + * @return the retry policy + */ + public RetryPolicy policy() { + return policy; + } + + /** + * The deadline of the run. + * @return the deadline, empty when the run has none + */ + public Optional deadline() { + return Optional.ofNullable(deadline); + } + + /** + * The guard of the run. + * @return the guard, empty for an uncontrolled run + */ + public Optional guard() { + return Optional.ofNullable(guard); + } + + private RetryDecision checkDeadline() { + if (deadline == null) { + return RetryDecision.allow(); + } + Instant now = clock.instant(); + if (!now.isBefore(deadline)) { + return stop(RetryStopReason.DEADLINE_EXCEEDED); + } + long remaining = Duration.between(now, deadline).toMillis(); + if (remaining <= policy.deadlineReserveMillis()) { + return stop(RetryStopReason.DEADLINE_EXCEEDED); + } + return RetryDecision.allow(); + } + + private RetryDecision stop(RetryStopReason reason) { + this.stopReason = reason; + return RetryDecision.stop(reason); + } + +} diff --git a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryDecision.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryDecision.java new file mode 100644 index 000000000..5eb68b179 --- /dev/null +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryDecision.java @@ -0,0 +1,121 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.retry; + +import java.time.Duration; +import java.util.Objects; +import java.util.Optional; + +/** + * The outcome of a single retry decision: start the attempt, wait and retry, or stop + * with a stable reason. + */ +public final class RetryDecision { + + private static final RetryDecision ALLOWED = new RetryDecision(null, null); + + private final Duration delay; + + private final RetryStopReason stopReason; + + private RetryDecision(Duration delay, RetryStopReason stopReason) { + this.delay = delay; + this.stopReason = stopReason; + } + + /** + * The attempt may start. + * @return the shared allow decision + */ + public static RetryDecision allow() { + return ALLOWED; + } + + /** + * The attempt may start again after the given delay. + * @param delay the backoff to wait before the next attempt + * @return a retry decision + */ + public static RetryDecision retryAfter(Duration delay) { + Objects.requireNonNull(delay, "delay must not be null"); + if (delay.isNegative()) { + throw new IllegalArgumentException("delay must not be negative"); + } + return new RetryDecision(delay, null); + } + + /** + * The sequence stops. + * @param reason why the sequence stopped + * @return a stop decision + */ + public static RetryDecision stop(RetryStopReason reason) { + Objects.requireNonNull(reason, "reason must not be null"); + return new RetryDecision(null, reason); + } + + /** + * Whether the attempt may start. + * @return {@code true} when neither a delay nor a stop reason is present + */ + public boolean isAllowed() { + return delay == null && stopReason == null; + } + + /** + * Whether another attempt follows after a delay. + * @return {@code true} when a delay is present + */ + public boolean isRetry() { + return delay != null; + } + + /** + * Whether the sequence stopped. + * @return {@code true} when a stop reason is present + */ + public boolean isStop() { + return stopReason != null; + } + + /** + * The backoff before the next attempt. + * @return the delay, empty when the decision is not a retry + */ + public Optional delay() { + return Optional.ofNullable(delay); + } + + /** + * Why the sequence stopped. + * @return the stop reason, empty when the sequence did not stop + */ + public Optional stopReason() { + return Optional.ofNullable(stopReason); + } + + @Override + public String toString() { + if (isRetry()) { + return "RetryDecision[retryAfter=" + delay + "]"; + } + if (isStop()) { + return "RetryDecision[stop=" + stopReason + "]"; + } + return "RetryDecision[allow]"; + } + +} diff --git a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryPolicy.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryPolicy.java new file mode 100644 index 000000000..fbab15001 --- /dev/null +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryPolicy.java @@ -0,0 +1,78 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.retry; + +import java.util.Objects; +import java.util.Set; + +/** + * The bounded configuration of a retry sequence. + * + * @param maxAttempts the total number of attempts, the first call included + * @param initialDelayMillis the backoff before the second attempt + * @param maxDelayMillis the upper bound of a single backoff + * @param backoffMultiplier the factor applied to every following backoff + * @param retryableCategories the failure categories that may be retried + * @param deadlineReserveMillis the time that must remain before the deadline to + * start another attempt + */ +public record RetryPolicy(int maxAttempts, long initialDelayMillis, long maxDelayMillis, double backoffMultiplier, + Set retryableCategories, long deadlineReserveMillis) { + + public RetryPolicy { + if (maxAttempts < 1) { + throw new IllegalArgumentException("maxAttempts must be greater than or equal to 1"); + } + if (initialDelayMillis < 0) { + throw new IllegalArgumentException("initialDelayMillis must be greater than or equal to 0"); + } + if (maxDelayMillis < 0) { + throw new IllegalArgumentException("maxDelayMillis must be greater than or equal to 0"); + } + if (!Double.isFinite(backoffMultiplier) || backoffMultiplier < 1.0) { + throw new IllegalArgumentException("backoffMultiplier must be greater than or equal to 1.0"); + } + if (deadlineReserveMillis < 0) { + throw new IllegalArgumentException("deadlineReserveMillis must be greater than or equal to 0"); + } + Objects.requireNonNull(retryableCategories, "retryableCategories must not be null"); + retryableCategories = Set.copyOf(retryableCategories); + } + + /** + * The default policy: three attempts, one second initial backoff, thirty seconds + * upper bound, and the retryable categories that are safe by default. + * @return the default policy + */ + public static RetryPolicy defaults() { + return new RetryPolicy(3, 1000L, 30000L, 2.0, + Set.of(ModelCallFailureCategory.TRANSIENT, ModelCallFailureCategory.RATE_LIMITED), 0L); + } + + /** + * The backoff that follows the given number of attempts. + * @param attemptsMade the number of attempts already made, at least one + * @return the delay in milliseconds before the next attempt + */ + public long delayMillisAfter(int attemptsMade) { + long delay = initialDelayMillis; + for (int attempt = 1; attempt < attemptsMade; attempt++) { + delay = Math.min((long) (delay * backoffMultiplier), maxDelayMillis); + } + return delay; + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculator.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryStopReason.java similarity index 56% rename from argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculator.java rename to argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryStopReason.java index bfcd410d2..cb84cc011 100644 --- a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/RetryDelayCalculator.java +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/RetryStopReason.java @@ -13,15 +13,26 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.github.agentic.ai.agent.orchestration.interceptor.modelretry; +package io.github.agentic.ai.agent.retry; -final class RetryDelayCalculator { +/** + * Why a retry sequence stopped instead of starting another attempt. + */ +public enum RetryStopReason { + + /** The configured number of attempts was used up. */ + ATTEMPTS_EXHAUSTED, + + /** The failure is not retryable under the active policy. */ + NON_RETRYABLE, + + /** The run deadline does not allow another attempt. */ + DEADLINE_EXCEEDED, - private RetryDelayCalculator() { - } + /** The attempt was cancelled. */ + CANCELLED, - static long nextDelay(long currentDelay, double multiplier, long maxDelay) { - return Math.min((long) (currentDelay * multiplier), maxDelay); - } + /** The execution guard stopped the run. */ + EXECUTION_STOPPED } diff --git a/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/TypeBasedModelCallFailureClassifier.java b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/TypeBasedModelCallFailureClassifier.java new file mode 100644 index 000000000..903756dad --- /dev/null +++ b/argi-agent/argi-agent-api/src/main/java/io/github/agentic/ai/agent/retry/TypeBasedModelCallFailureClassifier.java @@ -0,0 +1,75 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.retry; + +import java.io.IOException; +import java.net.SocketTimeoutException; +import java.util.concurrent.CancellationException; +import java.util.concurrent.TimeoutException; +import java.util.function.Predicate; + +/** + * Classifies model call failures from the failure type and its cause chain. + * + *

+ * The classifier never inspects the text of an exception message. A provider that + * reports its failures through its own exception types can supply a dedicated + * {@link ModelCallFailureClassifier} instead; this implementation covers the failures + * that are recognisable without provider specific types. + */ +public final class TypeBasedModelCallFailureClassifier implements ModelCallFailureClassifier { + + /** Guards against a self referencing cause chain. */ + private static final int MAX_CAUSE_DEPTH = 32; + + @Override + public ModelCallFailure classify(Throwable failure) { + if (failure == null) { + return new ModelCallFailure(ModelCallFailureCategory.UNKNOWN, null); + } + if (matches(failure, TypeBasedModelCallFailureClassifier::isCancellation)) { + return new ModelCallFailure(ModelCallFailureCategory.CANCELLED, failure); + } + if (matches(failure, TypeBasedModelCallFailureClassifier::isTransient)) { + return new ModelCallFailure(ModelCallFailureCategory.TRANSIENT, failure); + } + return new ModelCallFailure(ModelCallFailureCategory.UNKNOWN, failure); + } + + private boolean matches(Throwable failure, Predicate predicate) { + Throwable cause = failure; + for (int depth = 0; cause != null && depth < MAX_CAUSE_DEPTH; depth++) { + if (predicate.test(cause)) { + return true; + } + Throwable next = cause.getCause(); + if (next == cause) { + break; + } + cause = next; + } + return false; + } + + private static boolean isCancellation(Throwable cause) { + return cause instanceof CancellationException || cause instanceof InterruptedException; + } + + private static boolean isTransient(Throwable cause) { + return cause instanceof SocketTimeoutException || cause instanceof TimeoutException || cause instanceof IOException; + } + +} diff --git a/argi-agent/argi-agent-api/src/test/java/io/github/agentic/ai/agent/retry/ModelRetryControllerTest.java b/argi-agent/argi-agent-api/src/test/java/io/github/agentic/ai/agent/retry/ModelRetryControllerTest.java new file mode 100644 index 000000000..cf9e3dd89 --- /dev/null +++ b/argi-agent/argi-agent-api/src/test/java/io/github/agentic/ai/agent/retry/ModelRetryControllerTest.java @@ -0,0 +1,223 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.retry; + +import io.github.agentic.ai.runtime.execution.DelegationScope; +import io.github.agentic.ai.runtime.execution.ExecutionBudget; +import io.github.agentic.ai.runtime.execution.ExecutionControl; +import io.github.agentic.ai.runtime.execution.ExecutionStoppedException; +import io.github.agentic.ai.runtime.execution.PrincipalReference; +import io.github.agentic.ai.runtime.execution.TenantReference; +import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.net.SocketTimeoutException; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.Set; +import java.util.concurrent.CancellationException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * 验证重试裁决：尝试次数上界、失败分类、deadline 门控与取消透传。 + */ +class ModelRetryControllerTest { + + private static final Instant NOW = Instant.parse("2026-01-01T00:00:00Z"); + + private static final Clock CLOCK = Clock.fixed(NOW, ZoneOffset.UTC); + + /** 验证总尝试次数含首次，达到上界后不再继续。 */ + @Test + void theAttemptCountIsBoundedByThePolicy() { + ModelRetryController controller = controller(policy(3), null, null); + + for (int attempt = 1; attempt <= 3; attempt++) { + assertThat(controller.beforeAttempt().isAllowed()).isTrue(); + RetryDecision decision = controller.onFailure(new IOException("temporary outage")); + if (attempt < 3) { + assertThat(decision.isRetry()).isTrue(); + } + else { + assertThat(decision.stopReason()).contains(RetryStopReason.ATTEMPTS_EXHAUSTED); + } + } + + assertThat(controller.attempts()).isEqualTo(3); + } + + /** 验证不可重试的失败在第一次就终止序列。 */ + @Test + void aPermanentFailureStopsTheSequenceImmediately() { + ModelRetryController controller = controller(policy(3), + failure -> new ModelCallFailure(ModelCallFailureCategory.PERMANENT, failure), null); + + assertThat(controller.beforeAttempt().isAllowed()).isTrue(); + RetryDecision decision = controller.onFailure(new IllegalStateException("rejected")); + + assertThat(decision.stopReason()).contains(RetryStopReason.NON_RETRYABLE); + assertThat(controller.attempts()).isEqualTo(1); + } + + /** 验证退避按倍数增长且不超过配置上限。 */ + @Test + void theBackoffGrowsAndStaysBounded() { + ModelRetryController controller = controller(new RetryPolicy(5, 100L, 250L, 2.0, retryable(), 0L), null, null); + + controller.beforeAttempt(); + assertThat(controller.onFailure(new IOException("temporary")).delay()).contains(Duration.ofMillis(100L)); + controller.beforeAttempt(); + assertThat(controller.onFailure(new IOException("temporary")).delay()).contains(Duration.ofMillis(200L)); + controller.beforeAttempt(); + assertThat(controller.onFailure(new IOException("temporary")).delay()).contains(Duration.ofMillis(250L)); + } + + /** 验证零退避保持为零，不引入额外等待。 */ + @Test + void aZeroDelayStaysZero() { + ModelRetryController controller = controller(new RetryPolicy(3, 0L, 30000L, 2.0, retryable(), 0L), null, null); + + controller.beforeAttempt(); + + assertThat(controller.onFailure(new IOException("temporary")).delay()).contains(Duration.ZERO); + } + + /** 验证 deadline 已过时不再允许发起下一次尝试。 */ + @Test + void anExpiredDeadlineStopsTheSequence() { + ModelRetryController controller = controller(policy(3), null, NOW.minusSeconds(1)); + + RetryDecision decision = controller.beforeAttempt(); + + assertThat(decision.stopReason()).contains(RetryStopReason.DEADLINE_EXCEEDED); + assertThat(controller.attempts()).isZero(); + } + + /** 验证剩余时间不足以再发起一次尝试时停止序列。 */ + @Test + void aDeadlineWithoutRoomForAnotherAttemptStopsTheSequence() { + ModelRetryController controller = controller(new RetryPolicy(3, 100L, 1000L, 2.0, retryable(), 500L), null, + NOW.plusMillis(100)); + + assertThat(controller.beforeAttempt().stopReason()).contains(RetryStopReason.DEADLINE_EXCEEDED); + } + + /** 验证退避时长会越过 deadline 时直接停止，而不是空等到超时。 */ + @Test + void aBackoffThatWouldOutliveTheDeadlineStopsTheSequence() { + ModelRetryController controller = controller(new RetryPolicy(3, 1000L, 1000L, 2.0, retryable(), 0L), null, + NOW.plusMillis(100)); + + controller.beforeAttempt(); + RetryDecision decision = controller.onFailure(new IOException("temporary")); + + assertThat(decision.stopReason()).contains(RetryStopReason.DEADLINE_EXCEEDED); + } + + /** 验证运行被取消时守卫的停止信号原样透传。 */ + @Test + void aCancelledRunStopsTheSequence() { + ExecutionControl control = control(NOW.plusSeconds(60)); + control.cancel(); + ModelRetryController controller = new ModelRetryController(policy(3), new TypeBasedModelCallFailureClassifier(), + control, CLOCK, NOW.plusSeconds(60)); + + assertThatThrownBy(controller::beforeAttempt).isInstanceOf(ExecutionStoppedException.class) + .satisfies(failure -> assertThat(((ExecutionStoppedException) failure).reason()) + .isEqualTo(ExecutionStoppedException.Reason.CANCELLED)); + } + + /** 验证守卫自身的 deadline 已过时同样阻止新的尝试。 */ + @Test + void anExpiredGuardStopsTheSequence() { + ExecutionControl control = control(NOW.minusSeconds(1)); + ModelRetryController controller = new ModelRetryController(policy(3), new TypeBasedModelCallFailureClassifier(), + control, CLOCK, NOW.minusSeconds(1)); + + assertThatThrownBy(controller::beforeAttempt).isInstanceOf(ExecutionStoppedException.class) + .satisfies(failure -> assertThat(((ExecutionStoppedException) failure).reason()) + .isEqualTo(ExecutionStoppedException.Reason.DEADLINE_EXCEEDED)); + } + + /** 验证序列一旦停止就保持停止，不会再放行新的尝试。 */ + @Test + void aStoppedSequenceStaysStopped() { + ModelRetryController controller = controller(policy(3), + failure -> new ModelCallFailure(ModelCallFailureCategory.PERMANENT, failure), null); + + controller.beforeAttempt(); + assertThat(controller.onFailure(new IllegalStateException("rejected")).stopReason()) + .contains(RetryStopReason.NON_RETRYABLE); + assertThat(controller.beforeAttempt().stopReason()).contains(RetryStopReason.NON_RETRYABLE); + assertThat(controller.attempts()).isEqualTo(1); + } + + /** 验证分类只看失败类型，不看异常正文：正文像超时的永久错误仍然不可重试。 */ + @Test + void theClassificationDoesNotReadTheFailureMessage() { + TypeBasedModelCallFailureClassifier classifier = new TypeBasedModelCallFailureClassifier(); + + ModelCallFailure failure = classifier.classify(new IllegalStateException("I/O error: connection timeout")); + + assertThat(failure.category()).isEqualTo(ModelCallFailureCategory.UNKNOWN); + assertThat(failure.isRetryable(retryable())).isFalse(); + } + + /** 验证取消与网络超时按类型归类。 */ + @Test + void cancellationsAndTimeoutsAreClassifiedByType() { + TypeBasedModelCallFailureClassifier classifier = new TypeBasedModelCallFailureClassifier(); + + assertThat(classifier.classify(new CancellationException("cancelled")).category()) + .isEqualTo(ModelCallFailureCategory.CANCELLED); + assertThat(classifier.classify(new SocketTimeoutException("read timeout")).category()) + .isEqualTo(ModelCallFailureCategory.TRANSIENT); + assertThat(classifier.classify(new IllegalStateException("connection lost")).category()) + .isEqualTo(ModelCallFailureCategory.UNKNOWN); + } + + private ModelRetryController controller(RetryPolicy policy, ModelCallFailureClassifier classifier, Instant deadline) { + return new ModelRetryController(policy, + classifier == null ? new TypeBasedModelCallFailureClassifier() : classifier, null, CLOCK, deadline); + } + + private static RetryPolicy policy(int maxAttempts) { + return new RetryPolicy(maxAttempts, 100L, 1000L, 2.0, retryable(), 0L); + } + + private static Set retryable() { + return Set.of(ModelCallFailureCategory.TRANSIENT, ModelCallFailureCategory.RATE_LIMITED); + } + + private static ExecutionControl control(Instant deadline) { + TrustedExecutionContext trusted = TrustedExecutionContext.builder() + .tenant(new TenantReference("test", "tenant")) + .principal(new PrincipalReference("test", "principal")) + .runId("run") + .correlationId("correlation") + .deadline(deadline) + .budget(ExecutionBudget.unlimited()) + .delegationScope(DelegationScope.none()) + .build(); + return new ExecutionControl(trusted, CLOCK); + } + +} diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java index c8a711b9c..b0092637b 100644 --- a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java +++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java @@ -398,7 +398,11 @@ private ChatResponse modelTurn(List messages, Map s AtomicReference effectiveTools) { context.control().assertActive(); ModelRequest request = ModelRequest.builder().messages(messages).toolRegistry(availableTools) - .trustedExecutionContext(context.trusted()).context(state).attributes(state).build(); + .trustedExecutionContext(context.trusted()) + .executionGuard(context.control()) + .context(state) + .attributes(state) + .build(); // The effective tool set is only known at the end of the interceptor chain, // where selection semantics are resolved; execution in the same turn must use // exactly the set the model was shown (issue #268: declared and executed agree). diff --git a/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentExecutionGuardTest.java b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentExecutionGuardTest.java new file mode 100644 index 000000000..bc3988a64 --- /dev/null +++ b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentExecutionGuardTest.java @@ -0,0 +1,146 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.react; + +import io.github.agentic.ai.agent.AgentEvent; +import io.github.agentic.ai.agent.AgentExecutionContext; +import io.github.agentic.ai.agent.AgentRequest; +import io.github.agentic.ai.agent.interceptor.ModelCallHandler; +import io.github.agentic.ai.agent.interceptor.ModelInterceptor; +import io.github.agentic.ai.agent.interceptor.ModelRequest; +import io.github.agentic.ai.agent.interceptor.ModelResponse; +import io.github.agentic.ai.api.chat.ChatClient; +import io.github.agentic.ai.api.chat.ChatClientCapabilities; +import io.github.agentic.ai.api.chat.ChatEvent; +import io.github.agentic.ai.api.chat.ChatMessage; +import io.github.agentic.ai.api.chat.ChatRequest; +import io.github.agentic.ai.api.chat.ChatResponse; +import io.github.agentic.ai.api.tool.ToolRegistry; +import io.github.agentic.ai.runtime.execution.DelegationScope; +import io.github.agentic.ai.runtime.execution.ExecutionBudget; +import io.github.agentic.ai.runtime.execution.ExecutionControl; +import io.github.agentic.ai.runtime.execution.PrincipalReference; +import io.github.agentic.ai.runtime.execution.TenantReference; +import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.time.Clock; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Flow; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * 验证中立 ReAct 把运行守卫传给模型请求，使重试序列能被运行本身终止。 + */ +@Timeout(20) +class ReActAgentExecutionGuardTest { + + /** 验证进入拦截器链的模型请求携带当前运行的执行守卫。 */ + @Test + void theModelRequestCarriesTheGuardOfTheRun() throws Exception { + AtomicReference captured = new AtomicReference<>(); + ModelInterceptor interceptor = new ModelInterceptor() { + @Override + public String getName() { + return "capture-guard"; + } + + @Override + public ModelResponse interceptModel(ModelRequest request, ModelCallHandler handler) { + captured.set((ExecutionControl) request.getExecutionGuard().orElse(null)); + return handler.call(request); + } + }; + ReActAgent agent = ReActAgent.builder(new SingleTurnClient()) + .tools(new ToolRegistry(List.of())) + .interceptors(List.of(interceptor)) + .build(); + + try (AgentExecutionContext context = context()) { + CountDownLatch completed = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + agent.stream(AgentRequest.messages(List.of(ChatMessage.user("hello"))), context) + .subscribe(new Flow.Subscriber<>() { + @Override + public void onSubscribe(Flow.Subscription subscription) { + subscription.request(Long.MAX_VALUE); + } + + @Override + public void onNext(AgentEvent event) { + } + + @Override + public void onError(Throwable throwable) { + failure.set(throwable); + completed.countDown(); + } + + @Override + public void onComplete() { + completed.countDown(); + } + }); + + assertThat(completed.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(failure.get()).isNull(); + assertThat(captured.get()).isSameAs(context.control()); + } + } + + private AgentExecutionContext context() { + TrustedExecutionContext trusted = TrustedExecutionContext.builder() + .tenant(new TenantReference("test", "tenant")) + .principal(new PrincipalReference("test", "principal")) + .runId("guard-test") + .correlationId("guard-correlation") + .budget(ExecutionBudget.unlimited()) + .delegationScope(DelegationScope.none()) + .build(); + return new AgentExecutionContext(trusted, Clock.systemUTC()); + } + + /** 返回一次普通回答的本地模型桩。 */ + private static final class SingleTurnClient implements ChatClient { + + @Override + public CompletionStage call(ChatRequest request) { + return CompletableFuture + .completedFuture(new ChatResponse("id", "model", List.of(ChatMessage.assistant("done")), List.of(), null, + null, Map.of())); + } + + @Override + public Flow.Publisher stream(ChatRequest request) { + throw new AssertionError("Expected the non-streaming model path"); + } + + @Override + public ChatClientCapabilities capabilities() { + return ChatClientCapabilities.basic(); + } + + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java index f859b319b..fe2a8bc7c 100644 --- a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java @@ -19,13 +19,25 @@ import io.github.agentic.ai.agent.interceptor.ModelInterceptor; import io.github.agentic.ai.agent.interceptor.ModelRequest; import io.github.agentic.ai.agent.interceptor.ModelResponse; - +import io.github.agentic.ai.agent.retry.ModelCallFailure; +import io.github.agentic.ai.agent.retry.ModelCallFailureCategory; +import io.github.agentic.ai.agent.retry.ModelCallFailureClassifier; +import io.github.agentic.ai.agent.retry.ModelRetryController; +import io.github.agentic.ai.agent.retry.RetryDecision; +import io.github.agentic.ai.agent.retry.RetryPolicy; +import io.github.agentic.ai.agent.retry.RetryStopReason; import io.github.agentic.ai.api.chat.ChatEvent; +import io.github.agentic.ai.runtime.execution.ExecutionGuard; +import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.time.Clock; import java.time.Duration; +import java.time.Instant; +import java.util.EnumSet; import java.util.Objects; +import java.util.Set; import java.util.concurrent.CancellationException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Predicate; @@ -44,11 +56,20 @@ * .backoffMultiplier(2.0) * .build(); * + * + *

+ * The retry sequence is bounded on three axes at once: the number of attempts, the + * retryability of the observed failure and the deadline of the run. When the model + * request carries the guard of its run, a cancelled or expired run stops the sequence + * instead of dispatching another provider call. */ public class ModelRetryInterceptor extends ModelInterceptor { private static final Logger log = LoggerFactory.getLogger(ModelRetryInterceptor.class); + /** The longest a single backoff slice blocks before the run control is checked again. */ + private static final long BACKOFF_SLICE_MILLIS = 100L; + private final int maxAttempts; private final long initialDelay; @@ -59,12 +80,24 @@ public class ModelRetryInterceptor extends ModelInterceptor { private final Predicate retryableExceptionPredicate; + private final ModelCallFailureClassifier failureClassifier; + + private final Set retryableCategories; + + private final long deadlineReserveMillis; + + private final Clock clock; + private ModelRetryInterceptor(Builder builder) { this.maxAttempts = builder.maxAttempts; this.initialDelay = builder.initialDelay; this.maxDelay = builder.maxDelay; this.backoffMultiplier = builder.backoffMultiplier; this.retryableExceptionPredicate = builder.retryableExceptionPredicate; + this.failureClassifier = builder.failureClassifier; + this.retryableCategories = Set.copyOf(builder.retryableCategories); + this.deadlineReserveMillis = builder.deadlineReserveMillis; + this.clock = builder.clock; } public static Builder builder() { @@ -73,12 +106,16 @@ public static Builder builder() { @Override public ModelResponse interceptModel(ModelRequest request, ModelCallHandler handler) { + ModelRetryController controller = controller(request); Exception lastException = null; - long currentDelay = initialDelay; for (int attempt = 1; attempt <= maxAttempts; attempt++) { + checkInterrupted(); + RetryDecision start = controller.beforeAttempt(); + if (start.isStop()) { + throw terminalFailure(start.stopReason().orElse(RetryStopReason.EXECUTION_STOPPED), lastException); + } try { - checkInterrupted(); if (attempt > 1) { log.info("Retry model call, on the {}th attempt (out of {} attempts).", attempt, maxAttempts); } @@ -88,7 +125,7 @@ public ModelResponse interceptModel(ModelRequest request, ModelCallHandler handl if (modelResponse.isStreaming()) { return ModelResponse .stream(JdkFlowAdapter.publisherToFlowPublisher(withStreamingRetry(request, handler, - JdkFlowAdapter.flowPublisherToFlux(modelResponse.getStream()), attempt, currentDelay))); + JdkFlowAdapter.flowPublisherToFlux(modelResponse.getStream()), attempt, controller))); } // Assistant content is data, never an out-of-band error signal. if (attempt > 1) { @@ -105,23 +142,21 @@ public ModelResponse interceptModel(ModelRequest request, ModelCallHandler handl lastException = e; log.warn("Model call failed (attempted {}/{}): {}", attempt, maxAttempts, e.getMessage()); - if (attempt >= maxAttempts) { - log.error("The maximum number of retries has been reached {}, and the model call has failed.", - maxAttempts); - throw new RuntimeException("Model call failed, maximum number of retries reached.", lastException); - } - - // Determine if an exception can be retried. - if (!retryableExceptionPredicate.test(e)) { - log.warn("Exceptions cannot be retried and are thrown immediately: {}", e.getMessage()); - throw new RuntimeException("Model call failed (non-retryable exception)", e); + RetryDecision decision = controller.onFailure(e); + if (!decision.isRetry()) { + RetryStopReason reason = decision.stopReason().orElse(RetryStopReason.NON_RETRYABLE); + if (reason == RetryStopReason.ATTEMPTS_EXHAUSTED) { + log.error("The maximum number of retries has been reached {}, and the model call has failed.", + maxAttempts); + } + else if (reason == RetryStopReason.NON_RETRYABLE) { + log.warn("Exceptions cannot be retried and are thrown immediately: {}", e.getMessage()); + } + throw terminalFailure(reason, e); } // Wait and try again - pause(currentDelay); - - // Calculate the next delay time (exponential backoff) - currentDelay = RetryDelayCalculator.nextDelay(currentDelay, backoffMultiplier, maxDelay); + pause(decision.delay().orElseThrow().toMillis(), controller); } } @@ -130,8 +165,13 @@ public ModelResponse interceptModel(ModelRequest request, ModelCallHandler handl lastException); } - private Flux withStreamingRetry(ModelRequest request, ModelCallHandler handler, - Flux responseFlux, int attempt, long currentDelay) { + /** + * Wraps a streaming response. Every subscription of the response owns its attempt + * budget, so that a repeated subscription is not charged for the attempts of an + * earlier one. + */ + private Flux withStreamingRetry(ModelRequest request, ModelCallHandler handler, Flux responseFlux, + int attempt, ModelRetryController controller) { return Flux.defer(() -> { CancellationException cancellation = cancellation(null); if (cancellation != null) { @@ -143,13 +183,13 @@ private Flux withStreamingRetry(ModelRequest request, ModelCallHandle // Retrying after partial output would duplicate chunks downstream. return Flux.error(error); } - return retryStreamingModelCall(request, handler, attempt, currentDelay, error); + return retryStreamingModelCall(request, handler, attempt, controller, error); }); }); } private Flux retryStreamingModelCall(ModelRequest request, ModelCallHandler handler, int failedAttempt, - long currentDelay, Throwable error) { + ModelRetryController controller, Throwable error) { if (error instanceof Error) { return Flux.error(error); } @@ -158,43 +198,64 @@ private Flux retryStreamingModelCall(ModelRequest request, ModelCallH return Flux.error(cancellation); } Exception exception = toException(error); - log.warn("Streaming model call failed (attempted {}/{}): {}", failedAttempt, maxAttempts, - exception.getMessage()); + log.warn("Streaming model call failed (attempted {}/{}): {}", failedAttempt, maxAttempts, exception.getMessage()); - if (failedAttempt >= maxAttempts) { - log.error("The maximum number of retries has been reached {}, and the streaming model call has failed.", - maxAttempts); - return Flux.error(new RuntimeException("Model call failed, maximum number of retries reached.", exception)); - } - - if (!retryableExceptionPredicate.test(exception)) { - log.warn("Exceptions cannot be retried and are thrown immediately: {}", exception.getMessage()); - return Flux.error(new RuntimeException("Model call failed (non-retryable exception)", exception)); + RetryDecision decision = controller.onFailure(exception, failedAttempt); + if (!decision.isRetry()) { + if (decision.stopReason().orElse(RetryStopReason.NON_RETRYABLE) == RetryStopReason.NON_RETRYABLE) { + log.warn("Exceptions cannot be retried and are thrown immediately: {}", exception.getMessage()); + } + return Flux.error(terminalFailure(decision.stopReason().orElse(RetryStopReason.NON_RETRYABLE), exception)); } int nextAttempt = failedAttempt + 1; - long nextDelay = nextDelay(currentDelay); - return delay(currentDelay).thenMany(Flux.defer(() -> { + long nextDelay = decision.delay().orElseThrow().toMillis(); + return delay(nextDelay).thenMany(Flux.defer(() -> { log.info("Retry model call, on the {}th attempt (out of {} attempts).", nextAttempt, maxAttempts); try { checkInterrupted(); + RetryDecision start = controller.checkAttemptAllowed(); + if (start.isStop()) { + return Flux + .error(terminalFailure(start.stopReason().orElse(RetryStopReason.EXECUTION_STOPPED), exception)); + } ModelResponse retryResponse = handler.call(request); - return toStreamingFlux(request, handler, retryResponse, nextAttempt, nextDelay); + return toStreamingFlux(request, handler, retryResponse, nextAttempt, controller); } catch (Exception retryError) { - return retryStreamingModelCall(request, handler, nextAttempt, nextDelay, retryError); + return retryStreamingModelCall(request, handler, nextAttempt, controller, retryError); } })); } - private void pause(long delay) { - checkInterrupted(); + /** + * Waits for the backoff. A run that carries a guard waits in slices, so that a + * cancellation or an expired deadline stops the wait instead of being noticed + * only when the next attempt would start. + */ + private void pause(long delay, ModelRetryController controller) { if (delay <= 0) { return; } + log.info("Retry after {} ms", delay); + ExecutionGuard guard = controller.guard().orElse(null); + if (guard == null) { + sleep(delay); + return; + } + long remaining = delay; + while (remaining > 0) { + guard.assertActive(); + long slice = Math.min(remaining, BACKOFF_SLICE_MILLIS); + sleep(slice); + remaining -= slice; + } + } + + private void sleep(long millis) { + checkInterrupted(); try { - log.info("Retry after {} ms", delay); - Thread.sleep(delay); + Thread.sleep(millis); } catch (InterruptedException failure) { Thread.currentThread().interrupt(); @@ -230,10 +291,10 @@ private CancellationException cancellation(Throwable failure) { } private Flux toStreamingFlux(ModelRequest request, ModelCallHandler handler, ModelResponse modelResponse, - int attempt, long currentDelay) { + int attempt, ModelRetryController controller) { if (modelResponse.isStreaming()) { return withStreamingRetry(request, handler, JdkFlowAdapter.flowPublisherToFlux(modelResponse.getStream()), - attempt, currentDelay); + attempt, controller); } return Flux.just(ChatEvent.response(modelResponse.getChatResponse())); } @@ -245,10 +306,6 @@ private Flux delay(long currentDelay) { return Flux.empty().delaySubscription(Duration.ofMillis(currentDelay)); } - private long nextDelay(long currentDelay) { - return RetryDelayCalculator.nextDelay(currentDelay, backoffMultiplier, maxDelay); - } - private Exception toException(Throwable error) { if (error instanceof Exception exception) { return exception; @@ -256,6 +313,53 @@ private Exception toException(Throwable error) { return new RuntimeException(error); } + /** + * Builds the controller of one model call. The deadline of the run comes from its + * guard, and falls back to the trusted execution context of the request. + */ + private ModelRetryController controller(ModelRequest request) { + ExecutionGuard guard = request.getExecutionGuard().orElse(null); + Instant deadline = guard == null ? null : guard.deadline().orElse(null); + if (deadline == null) { + deadline = request.getTrustedExecutionContext() + .flatMap(TrustedExecutionContext::deadline) + .orElse(null); + } + RetryPolicy policy = new RetryPolicy(maxAttempts, initialDelay, maxDelay, backoffMultiplier, retryableCategories, + deadlineReserveMillis); + return new ModelRetryController(policy, this::classify, guard, clock, deadline); + } + + /** + * Classifies a failed attempt. Without an explicit classifier the legacy + * predicate decides, so that an existing configuration keeps its behaviour. An + * explicit classifier classifies by failure type instead of by message text. + */ + private ModelCallFailure classify(Throwable failure) { + if (failure instanceof CancellationException || failure instanceof InterruptedException) { + return new ModelCallFailure(ModelCallFailureCategory.CANCELLED, failure); + } + if (failureClassifier != null) { + ModelCallFailure classified = failureClassifier.classify(failure); + return classified != null ? classified : new ModelCallFailure(ModelCallFailureCategory.UNKNOWN, failure); + } + boolean retryable = failure instanceof Exception && retryableExceptionPredicate.test((Exception) failure); + return new ModelCallFailure(retryable ? ModelCallFailureCategory.TRANSIENT : ModelCallFailureCategory.PERMANENT, + failure); + } + + private RuntimeException terminalFailure(RetryStopReason reason, Throwable cause) { + return switch (reason) { + case ATTEMPTS_EXHAUSTED -> new RuntimeException("Model call failed, maximum number of retries reached.", + cause); + case NON_RETRYABLE -> new RuntimeException("Model call failed (non-retryable exception)", cause); + case DEADLINE_EXCEEDED -> new RuntimeException( + "Model call failed, the run deadline does not allow another attempt", cause); + case CANCELLED -> new RuntimeException("Model call failed, the attempt was cancelled", cause); + case EXECUTION_STOPPED -> new RuntimeException("Model call failed, the run stopped", cause); + }; + } + @Override public String getName() { return "ModelRetry"; @@ -273,6 +377,15 @@ public static class Builder { private Predicate retryableExceptionPredicate = Builder::isRetryableException; + private ModelCallFailureClassifier failureClassifier; + + private Set retryableCategories = EnumSet.of(ModelCallFailureCategory.TRANSIENT, + ModelCallFailureCategory.RATE_LIMITED); + + private long deadlineReserveMillis; + + private Clock clock = Clock.systemUTC(); + /** * Set the maximum number of retries (including the first call). * @param maxAttempts The maximum number of attempts must be >= 1. @@ -330,6 +443,50 @@ public Builder retryableExceptionPredicate(Predicate predicate) { return this; } + /** + * Classifies the failures by their type instead of by the text of their + * message. Without an explicit classifier the retryable exception predicate + * decides, which keeps the behaviour of an existing configuration. + * @param failureClassifier The classifier of a failed attempt + */ + public Builder failureClassifier(ModelCallFailureClassifier failureClassifier) { + this.failureClassifier = Objects.requireNonNull(failureClassifier, "failureClassifier must not be null"); + return this; + } + + /** + * Sets the failure categories that may be retried. + * @param retryableCategories The retryable categories, must not be null + */ + public Builder retryableCategories(Set retryableCategories) { + this.retryableCategories = Objects + .requireNonNull(retryableCategories, "retryableCategories must not be null"); + return this; + } + + /** + * Sets the time that must remain before the deadline of the run to start + * another attempt. + * @param deadlineReserve The reserved time, must not be negative + */ + public Builder deadlineReserve(Duration deadlineReserve) { + Objects.requireNonNull(deadlineReserve, "deadlineReserve must not be null"); + if (deadlineReserve.isNegative()) { + throw new IllegalArgumentException("deadlineReserve must be greater than or equal to 0."); + } + this.deadlineReserveMillis = deadlineReserve.toMillis(); + return this; + } + + /** + * Sets the clock used for the deadline checks. + * @param clock The clock, must not be null + */ + public Builder clock(Clock clock) { + this.clock = Objects.requireNonNull(clock, "clock must not be null"); + return this; + } + public ModelRetryInterceptor build() { return new ModelRetryInterceptor(this); } diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNode.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNode.java index 50d877ed4..26eeca067 100644 --- a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNode.java +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNode.java @@ -190,7 +190,8 @@ public Map apply(OverAllState state, RunnableConfig config) thro .options(this.generationOptions) .toolRegistry(configuredRegistry()) .context(contextMap) - .trustedExecutionContext(config.trustedExecutionContext().orElse(null)); + .trustedExecutionContext(config.trustedExecutionContext().orElse(null)) + .executionGuard(config.executionGuard().orElse(null)); // Extract tool names and descriptions from toolCallbacks and pass them to // ModelRequest diff --git a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java new file mode 100644 index 000000000..e4c357aa3 --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java @@ -0,0 +1,223 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.orchestration.interceptors; + +import io.github.agentic.ai.agent.interceptor.ModelRequest; +import io.github.agentic.ai.agent.interceptor.ModelResponse; +import io.github.agentic.ai.agent.orchestration.interceptor.modelretry.ModelRetryInterceptor; +import io.github.agentic.ai.agent.retry.TypeBasedModelCallFailureClassifier; +import io.github.agentic.ai.api.chat.ChatMessage; +import io.github.agentic.ai.api.chat.ChatResponse; +import io.github.agentic.ai.api.chat.Usage; +import io.github.agentic.ai.runtime.execution.DelegationScope; +import io.github.agentic.ai.runtime.execution.ExecutionBudget; +import io.github.agentic.ai.runtime.execution.ExecutionControl; +import io.github.agentic.ai.runtime.execution.ExecutionStoppedException; +import io.github.agentic.ai.runtime.execution.PrincipalReference; +import io.github.agentic.ai.runtime.execution.TenantReference; +import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * 验证受控重试：尝试次数上界、失败分类、取消与 deadline 门控、用量不重计。 + */ +@Timeout(20) +class ModelRetryInterceptorControlledTest { + + private static final Instant NOW = Instant.parse("2026-01-01T00:00:00Z"); + + private static final Clock CLOCK = Clock.fixed(NOW, ZoneOffset.UTC); + + /** 验证连续失败时的调用次数恰好等于配置的总尝试次数。 */ + @Test + void theNumberOfCallsIsBoundedByTheConfiguredAttempts() { + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder().maxAttempts(3).initialDelay(0).build(); + AtomicInteger attempts = new AtomicInteger(); + + assertThatThrownBy(() -> interceptor.interceptModel(ModelRequest.builder().build(), request -> { + attempts.incrementAndGet(); + throw new IllegalStateException("I/O error: connection timeout"); + })).isInstanceOf(RuntimeException.class) + .hasMessageContaining("maximum number of retries reached"); + + assertThat(attempts).hasValue(3); + } + + /** 验证显式分类器按失败类型判定，正文像超时的永久错误也只调用一次。 */ + @Test + void anExplicitClassifierDecidesWithoutReadingTheFailureMessage() { + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder() + .maxAttempts(3) + .initialDelay(0) + .failureClassifier(new TypeBasedModelCallFailureClassifier()) + .build(); + AtomicInteger attempts = new AtomicInteger(); + + assertThatThrownBy(() -> interceptor.interceptModel(ModelRequest.builder().build(), request -> { + attempts.incrementAndGet(); + throw new IllegalStateException("I/O error: connection timeout"); + })).isInstanceOf(RuntimeException.class) + .hasMessageContaining("non-retryable exception"); + + assertThat(attempts).hasValue(1); + } + + /** 验证未显式配置分类器时，既有谓词仍按原语义决定是否重试。 */ + @Test + void theDefaultConfigurationKeepsTheLegacyPredicate() { + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder().maxAttempts(3).initialDelay(0).build(); + AtomicInteger attempts = new AtomicInteger(); + + ModelResponse response = interceptor.interceptModel(ModelRequest.builder().build(), request -> { + if (attempts.incrementAndGet() == 1) { + throw new IllegalStateException("I/O error on POST request"); + } + return ModelResponse.of(ChatMessage.assistant("recovered")); + }); + + assertThat(attempts).hasValue(2); + assertThat(response.getMessage().text()).isEqualTo("recovered"); + } + + /** 验证退避期间取消运行后不再派发新的模型调用，停止信号原样透传。 */ + @Test + void aCancelledRunStopsTheBackoffWithoutAnotherCall() throws Exception { + ExecutionControl control = control(NOW.plusSeconds(60)); + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder() + .maxAttempts(3) + .initialDelay(30000) + .clock(CLOCK) + .build(); + AtomicInteger attempts = new AtomicInteger(); + CountDownLatch called = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + + Thread worker = new Thread(() -> { + try { + interceptor.interceptModel(request(control), request -> { + attempts.incrementAndGet(); + called.countDown(); + throw new IllegalStateException("I/O error: connection lost"); + }); + } + catch (Throwable thrown) { + failure.set(thrown); + } + }); + worker.start(); + try { + boolean started = called.await(5, TimeUnit.SECONDS); + control.cancel(); + worker.join(10000); + + assertThat(failure.get()).as("the worker must stop with the signal of the guard") + .isInstanceOf(ExecutionStoppedException.class); + assertThat(started).as("the model call must have started").isTrue(); + assertThat(attempts).hasValue(1); + } + finally { + worker.interrupt(); + worker.join(2000); + } + } + + /** 验证 deadline 不足以再发起一次尝试时，序列在第一次失败后停止。 */ + @Test + void aDeadlineThatCannotFitAnotherAttemptStopsTheSequence() { + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder() + .maxAttempts(3) + .initialDelay(5000) + .clock(CLOCK) + .build(); + AtomicInteger attempts = new AtomicInteger(); + + assertThatThrownBy(() -> interceptor.interceptModel(requestWithDeadline(NOW.plusMillis(100)), request -> { + attempts.incrementAndGet(); + throw new IllegalStateException("I/O error: connection timeout"); + })).isInstanceOf(RuntimeException.class) + .hasMessageContaining("deadline"); + + assertThat(attempts).hasValue(1); + } + + /** 验证返回给引擎的用量只来自成功的那一次尝试。 */ + @Test + void theUsageOfTheSuccessfulAttemptIsReportedOnce() { + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder().maxAttempts(3).initialDelay(0).build(); + AtomicInteger attempts = new AtomicInteger(); + Usage usage = new Usage(7L, 3L, 10L); + + ModelResponse response = interceptor.interceptModel(ModelRequest.builder().build(), request -> { + if (attempts.incrementAndGet() == 1) { + throw new IllegalStateException("I/O error on POST request"); + } + return ModelResponse + .of(new ChatResponse("id", "model", List.of(ChatMessage.assistant("done")), List.of(), null, usage, + Map.of())); + }); + + assertThat(attempts).hasValue(2); + assertThat(response.getChatResponse().usage()).isEqualTo(usage); + } + + /** 验证复制请求时守卫不会丢失，否则上层语义会被静默降级。 */ + @Test + void aCopiedRequestKeepsTheGuard() { + ExecutionControl control = control(NOW.plusSeconds(60)); + ModelRequest request = ModelRequest.builder().executionGuard(control).build(); + + assertThat(ModelRequest.builder(request).build().getExecutionGuard()).contains(control); + } + + private ModelRequest request(ExecutionControl control) { + return ModelRequest.builder().executionGuard(control).build(); + } + + private ModelRequest requestWithDeadline(Instant deadline) { + return ModelRequest.builder().trustedExecutionContext(trusted(deadline)).build(); + } + + private ExecutionControl control(Instant deadline) { + return new ExecutionControl(trusted(deadline), CLOCK); + } + + private TrustedExecutionContext trusted(Instant deadline) { + return TrustedExecutionContext.builder() + .tenant(new TenantReference("test", "tenant")) + .principal(new PrincipalReference("test", "principal")) + .runId("run") + .correlationId("correlation") + .deadline(deadline) + .budget(ExecutionBudget.unlimited()) + .delegationScope(DelegationScope.none()) + .build(); + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNodeExecutionGuardTest.java b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNodeExecutionGuardTest.java new file mode 100644 index 000000000..e53e5795b --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/node/AgentLlmNodeExecutionGuardTest.java @@ -0,0 +1,116 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.orchestration.node; + +import io.github.agentic.ai.agent.interceptor.ModelCallHandler; +import io.github.agentic.ai.agent.interceptor.ModelInterceptor; +import io.github.agentic.ai.agent.interceptor.ModelRequest; +import io.github.agentic.ai.agent.interceptor.ModelResponse; +import io.github.agentic.ai.api.chat.ChatMessage; +import io.github.agentic.ai.runtime.execution.DelegationScope; +import io.github.agentic.ai.runtime.execution.ExecutionBudget; +import io.github.agentic.ai.runtime.execution.ExecutionControl; +import io.github.agentic.ai.runtime.execution.PrincipalReference; +import io.github.agentic.ai.runtime.execution.TenantReference; +import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; +import io.github.agentic.ai.workflow.graph.OverAllState; +import io.github.agentic.ai.workflow.graph.RunnableConfig; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.springframework.ai.chat.client.ChatClient; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.chat.model.ChatResponse; +import org.springframework.ai.chat.model.Generation; +import org.springframework.ai.chat.prompt.ChatOptions; +import org.springframework.ai.chat.prompt.Prompt; + +import java.time.Clock; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * 验证框架节点把运行守卫传给模型请求，使重试序列能被运行本身终止。 + */ +@Timeout(20) +class AgentLlmNodeExecutionGuardTest { + + /** 验证进入模型调用链的模型请求携带当前运行的执行守卫。 */ + @Test + void theModelRequestCarriesTheGuardOfTheRun() throws Exception { + ChatModel model = mock(ChatModel.class); + when(model.getOptions()).thenReturn(ChatOptions.builder().build()); + when(model.call(any(Prompt.class))).thenReturn(response("done")); + AtomicReference captured = new AtomicReference<>(); + ModelInterceptor interceptor = new ModelInterceptor() { + @Override + public String getName() { + return "capture-guard"; + } + + @Override + public ModelResponse interceptModel(ModelRequest request, ModelCallHandler handler) { + captured.set((ExecutionControl) request.getExecutionGuard().orElse(null)); + return handler.call(request); + } + }; + AgentLlmNode node = AgentLlmNode.builder() + .agentName("test") + .chatClient(new io.github.agentic.ai.adapter.springai.SpringAiChatClientAdapter(ChatClient.create(model))) + .modelInterceptors(List.of(interceptor)) + .build(); + ExecutionControl control = control(); + + apply(node, control); + + assertThat(captured.get()).isSameAs(control); + } + + private static Object apply(AgentLlmNode node, ExecutionControl control) throws Exception { + OverAllState state = new OverAllState(Map.of("messages", new ArrayList<>(List.of(ChatMessage.user("test"))))); + RunnableConfig config = RunnableConfig.builder() + .addMetadata("_stream_", false) + .executionGuard(control) + .build(); + return node.apply(state, config).get("messages"); + } + + private static ExecutionControl control() { + TrustedExecutionContext trusted = TrustedExecutionContext.builder() + .tenant(new TenantReference("test", "tenant")) + .principal(new PrincipalReference("test", "principal")) + .runId("guard-test") + .correlationId("guard-correlation") + .deadline(Instant.now().plusSeconds(60)) + .budget(ExecutionBudget.unlimited()) + .delegationScope(DelegationScope.none()) + .build(); + return new ExecutionControl(trusted, Clock.systemUTC()); + } + + private static ChatResponse response(String text) { + return new ChatResponse(List.of(new Generation(new AssistantMessage(text)))); + } + +} diff --git a/docs/architecture/controlled-model-retry.md b/docs/architecture/controlled-model-retry.md new file mode 100644 index 000000000..f887dd8f7 --- /dev/null +++ b/docs/architecture/controlled-model-retry.md @@ -0,0 +1,107 @@ +# Controlled model retry + +This document describes the retry contract that the agent runtime uses to repeat a +model call, and the configuration that keeps a retry sequence bounded. + +## Why the retry used to be uncontrolled + +`ModelRetryInterceptor` repeated a failed model call with three properties that made +the sequence hard to reason about: + +- **The retryability of a failure was guessed from its text.** The default predicate + matched substrings such as `connection`, `timeout` or `i/o error`, so a permanent + rejection that happened to mention a timeout was repeated as well, while a transient + failure with an unfamiliar message was not. +- **The run deadline was ignored.** The runtime owns a deadline for every run, but the + retry loop started the next provider call unconditionally, even when the deadline of + the run had already passed. +- **A cancelled run kept the sequence alive.** The loop observed thread interruption + only, while a run cancellation is delivered through the execution guard of the run. + A cancelled run therefore received another provider call after the cancellation. + +## The contract + +The neutral contract lives in `io.github.agentic.ai.agent.retry` in the `argi-agent-api` +module, so that the independent ReAct engine and the graph based framework share it +without either of them depending on the other. + +| Type | Responsibility | +|---|---| +| `ModelCallFailureCategory` | The classification of a failure: `TRANSIENT`, `RATE_LIMITED`, `PERMANENT`, `CANCELLED`, `UNKNOWN`. | +| `ModelCallFailure` | A classified failure and its cause. | +| `ModelCallFailureClassifier` | Classifies a failure; implementations classify from the failure type and its cause chain. | +| `TypeBasedModelCallFailureClassifier` | The default classifier; it reads types only and never the text of a message. | +| `RetryPolicy` | The bounded configuration: attempts, backoff, retryable categories and the reserved deadline time. | +| `RetryStopReason` | Why a sequence stopped: `ATTEMPTS_EXHAUSTED`, `NON_RETRYABLE`, `DEADLINE_EXCEEDED`, `CANCELLED`, `EXECUTION_STOPPED`. | +| `RetryDecision` | The outcome of one decision: allowed, retry after a delay, or stopped with a reason. | +| `ModelRetryController` | The state machine of one model call; it owns the attempt counter and the deadline. | + +`ModelRetryController` never waits. The caller owns the backoff and has to ask the +controller again before every attempt, because a cancellation or a deadline can stop +the run while the caller waits. + +## The three axes of a retry sequence + +1. **The number of attempts.** `RetryPolicy.maxAttempts` counts the first call, so + `maxAttempts(3)` performs at most three provider calls. +2. **The retryability of the failure.** The classifier maps the failure to a category + and the policy decides which categories may be repeated. `UNKNOWN` is not retryable + by default: a failure that cannot be classified is not repeated. +3. **The remaining time.** The effective deadline comes from the execution guard of the + run and falls back to the trusted execution context of the request. When the deadline + has passed, when less than `deadlineReserve` is left, or when the next backoff would + outlive the deadline, the sequence stops before another call is dispatched. + +## The guard of the run + +`ModelRequest` carries an optional `ExecutionGuard`. The engines propagate the guard +of the run they execute: + +- the independent ReAct engine passes the control of its execution context; +- the graph based framework passes the guard of the run configuration. + +The guard never reaches a provider: it is a typed field of the request and is not part +of the request attributes that the adapters forward to the model. + +With a guard present, the interceptor waits for the backoff in slices and checks the +guard before every slice, so a cancelled or expired run ends the wait instead of +dispatching another call. A stopped run is reported with the reason of the guard +(`CANCELLED`, `DEADLINE_EXCEEDED` or `OWNERSHIP_LOST`) and is never reported as an +exhausted retry. + +## Defaults and migration + +The existing configuration keeps its behaviour. The retryable exception predicate stays +the default decision, so an application that configured the interceptor before stays +unchanged. + +| Configuration | Decision | Recommended for | +|---|---|---| +| `.retryableExceptionPredicate(predicate)` (default) | The predicate decides, as before. | Existing applications. | +| `.failureClassifier(classifier)` | The classifier decides by failure type; the predicate is not consulted. | New applications, or applications that replace the predicate with a type based rule. | +| `.retryableCategories(categories)` | Restricts the categories that may be repeated. | Applications that want to repeat a category that is not retryable by default. | +| `.deadlineReserve(duration)` | The time that must remain before the deadline of the run to start another attempt. | Applications whose provider calls need room to report their result. | + +The type based classifier is the recommended default: + +```java +ModelRetryInterceptor.builder() + .maxAttempts(3) + .initialDelay(1000) + .backoffMultiplier(2.0) + .failureClassifier(new TypeBasedModelCallFailureClassifier()) + .deadlineReserve(Duration.ofMillis(500)) + .build(); +``` + +A provider that reports its failures through its own exception types can supply a +dedicated classifier instead, which keeps the decision independent of the text of a +message. + +## Scope + +This contract covers the repetition of one model call. It does not switch models, it +does not replay tool calls, it does not own a persistent retry budget and it does not +change the boundary of a streaming response that already produced output. The +assembly of the retry behaviour as a lifecycle strategy, together with the other model +policies, belongs to the lifecycle chain and to the model policy work. From f81fa2b10705ed0e7cc378cb9c29caa5999d2574 Mon Sep 17 00:00:00 2001 From: konlue Date: Wed, 4 Nov 2026 23:58:55 +0800 Subject: [PATCH 2/2] fix(agent): isolate streaming retry state per subscription and observe run stops during backoff (issue #313 review) --- .../modelretry/ModelRetryInterceptor.java | 35 ++++++++++---- .../ModelRetryInterceptorControlledTest.java | 46 +++++++++++++++++++ .../ModelRetryInterceptorTest.java | 23 ++++++++++ 3 files changed, 94 insertions(+), 10 deletions(-) diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java index fe2a8bc7c..280151224 100644 --- a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/interceptor/modelretry/ModelRetryInterceptor.java @@ -125,7 +125,7 @@ public ModelResponse interceptModel(ModelRequest request, ModelCallHandler handl if (modelResponse.isStreaming()) { return ModelResponse .stream(JdkFlowAdapter.publisherToFlowPublisher(withStreamingRetry(request, handler, - JdkFlowAdapter.flowPublisherToFlux(modelResponse.getStream()), attempt, controller))); + JdkFlowAdapter.flowPublisherToFlux(modelResponse.getStream()), attempt))); } // Assistant content is data, never an out-of-band error signal. if (attempt > 1) { @@ -167,16 +167,17 @@ else if (reason == RetryStopReason.NON_RETRYABLE) { /** * Wraps a streaming response. Every subscription of the response owns its attempt - * budget, so that a repeated subscription is not charged for the attempts of an - * earlier one. + * budget: the retry state is created per subscription, so a repeated subscription + * is not charged for the attempts of an earlier one. */ private Flux withStreamingRetry(ModelRequest request, ModelCallHandler handler, Flux responseFlux, - int attempt, ModelRetryController controller) { + int attempt) { return Flux.defer(() -> { CancellationException cancellation = cancellation(null); if (cancellation != null) { return Flux.error(cancellation); } + ModelRetryController controller = controller(request); AtomicBoolean chunkEmitted = new AtomicBoolean(false); return responseFlux.doOnNext(response -> chunkEmitted.set(true)).onErrorResume(error -> { if (chunkEmitted.get()) { @@ -210,7 +211,7 @@ private Flux retryStreamingModelCall(ModelRequest request, ModelCallH int nextAttempt = failedAttempt + 1; long nextDelay = decision.delay().orElseThrow().toMillis(); - return delay(nextDelay).thenMany(Flux.defer(() -> { + return delay(nextDelay, controller).thenMany(Flux.defer(() -> { log.info("Retry model call, on the {}th attempt (out of {} attempts).", nextAttempt, maxAttempts); try { checkInterrupted(); @@ -220,7 +221,7 @@ private Flux retryStreamingModelCall(ModelRequest request, ModelCallH .error(terminalFailure(start.stopReason().orElse(RetryStopReason.EXECUTION_STOPPED), exception)); } ModelResponse retryResponse = handler.call(request); - return toStreamingFlux(request, handler, retryResponse, nextAttempt, controller); + return toStreamingFlux(request, handler, retryResponse, nextAttempt); } catch (Exception retryError) { return retryStreamingModelCall(request, handler, nextAttempt, controller, retryError); @@ -291,19 +292,33 @@ private CancellationException cancellation(Throwable failure) { } private Flux toStreamingFlux(ModelRequest request, ModelCallHandler handler, ModelResponse modelResponse, - int attempt, ModelRetryController controller) { + int attempt) { if (modelResponse.isStreaming()) { return withStreamingRetry(request, handler, JdkFlowAdapter.flowPublisherToFlux(modelResponse.getStream()), - attempt, controller); + attempt); } return Flux.just(ChatEvent.response(modelResponse.getChatResponse())); } - private Flux delay(long currentDelay) { + /** + * Waits for the backoff of a streaming retry. A run that carries a guard waits in + * slices, mirroring the synchronous {@link #pause(long, ModelRetryController)}: a + * cancellation or an expired deadline terminates the wait instead of being noticed + * when the next attempt would start. Without a guard the wait is a plain delayed + * completion, because there is no run control to observe. + */ + private Flux delay(long currentDelay, ModelRetryController controller) { if (currentDelay <= 0) { return Flux.empty(); } - return Flux.empty().delaySubscription(Duration.ofMillis(currentDelay)); + ExecutionGuard guard = controller.guard().orElse(null); + if (guard == null) { + return Flux.empty().delaySubscription(Duration.ofMillis(currentDelay)); + } + long slices = (currentDelay + BACKOFF_SLICE_MILLIS - 1) / BACKOFF_SLICE_MILLIS; + return Flux.interval(Duration.ZERO, Duration.ofMillis(BACKOFF_SLICE_MILLIS)).take(slices) + .doOnNext(tick -> guard.assertActive()) + .thenMany(Flux.empty()); } private Exception toException(Throwable error) { diff --git a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java index e4c357aa3..76fda5522 100644 --- a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java +++ b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java @@ -15,10 +15,12 @@ */ package io.github.agentic.ai.agent.orchestration.interceptors; +import io.github.agentic.ai.agent.interceptor.ModelCallHandler; import io.github.agentic.ai.agent.interceptor.ModelRequest; import io.github.agentic.ai.agent.interceptor.ModelResponse; import io.github.agentic.ai.agent.orchestration.interceptor.modelretry.ModelRetryInterceptor; import io.github.agentic.ai.agent.retry.TypeBasedModelCallFailureClassifier; +import io.github.agentic.ai.api.chat.ChatEvent; import io.github.agentic.ai.api.chat.ChatMessage; import io.github.agentic.ai.api.chat.ChatResponse; import io.github.agentic.ai.api.chat.Usage; @@ -31,6 +33,8 @@ import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +import reactor.adapter.JdkFlowAdapter; +import reactor.core.publisher.Flux; import java.time.Clock; import java.time.Instant; @@ -196,6 +200,48 @@ void aCopiedRequestKeepsTheGuard() { assertThat(ModelRequest.builder(request).build().getExecutionGuard()).contains(control); } + /** + * 验证流式重试的响应式退避按切片观察运行状态：取消运行后等待在下一个切片终止， + * 而不是等满整个配置的退避时长。 + */ + @Test + void aStreamingBackoffTerminatesWhenTheRunIsCancelled() throws Exception { + ExecutionControl control = control(NOW.plusSeconds(60)); + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder() + .maxAttempts(2) + .initialDelay(30000) + .clock(CLOCK) + .build(); + AtomicInteger attempts = new AtomicInteger(); + ModelCallHandler handler = request -> { + attempts.incrementAndGet(); + return ModelResponse.stream(JdkFlowAdapter.publisherToFlowPublisher(Flux + .error(new RuntimeException("I/O error: connection lost")))); + }; + ModelResponse response = interceptor.interceptModel(request(control), handler); + Flux flux = JdkFlowAdapter.flowPublisherToFlux(response.getStream()); + + AtomicReference failure = new AtomicReference<>(); + CountDownLatch done = new CountDownLatch(1); + flux.subscribe(events -> { + }, error -> { + failure.set(error); + done.countDown(); + }, done::countDown); + // The failed streaming attempt is counted and the sliced backoff has started. + Thread.sleep(200); + + long cancelAt = System.nanoTime(); + control.cancel(); + assertThat(done.await(5, TimeUnit.SECONDS)).as("the wait must terminate on cancellation").isTrue(); + long elapsedMillis = (System.nanoTime() - cancelAt) / 1_000_000; + + assertThat(failure.get()).isInstanceOf(ExecutionStoppedException.class); + assertThat(elapsedMillis).as("the cancellation must be observed by the next 100 ms slice, not after 30 s") + .isLessThan(5000); + assertThat(attempts).hasValue(1); + } + private ModelRequest request(ExecutionControl control) { return ModelRequest.builder().executionGuard(control).build(); } diff --git a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorTest.java b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorTest.java index 019d65c6c..6915b56fa 100644 --- a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorTest.java +++ b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorTest.java @@ -371,6 +371,29 @@ void testBuilderValidation() { }, "backoffMultiplier 必须 >= 1.0"); } + @Test + void testStreamingRetryStateDoesNotLeakAcrossSubscriptions() { + ModelRetryInterceptor interceptor = ModelRetryInterceptor.builder().maxAttempts(2).initialDelay(0).build(); + + AtomicInteger attemptCount = new AtomicInteger(0); + + ModelCallHandler handler = request -> { + attemptCount.incrementAndGet(); + return streaming(Flux.error(new RuntimeException("I/O error on streaming response"))); + }; + + ModelResponse response = interceptor.interceptModel(ModelRequest.builder().build(), handler); + Flux responseFlux = JdkFlowAdapter.flowPublisherToFlux(response.getStream()); + + // First subscription: consumes its own budget (1 streaming retry) and exhausts. + assertThrows(RuntimeException.class, () -> responseFlux.collectList().block(Duration.ofSeconds(2))); + assertEquals(2, attemptCount.get(), "第一次订阅应拥有自己的尝试预算"); + + // Second subscription: a fresh budget, not the exhausted state of the first one. + assertThrows(RuntimeException.class, () -> responseFlux.collectList().block(Duration.ofSeconds(2))); + assertEquals(3, attemptCount.get(), "第二次订阅必须获得独立的重试预算"); + } + private static ModelResponse streaming(Flux events) { return ModelResponse.stream(JdkFlowAdapter.publisherToFlowPublisher(events)); }