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..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 @@ -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))); } // 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,26 +165,32 @@ 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: 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) { 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()) { // 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 +199,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, controller).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); } 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,23 +292,33 @@ private CancellationException cancellation(Throwable failure) { } private Flux toStreamingFlux(ModelRequest request, ModelCallHandler handler, ModelResponse modelResponse, - int attempt, long currentDelay) { + int attempt) { if (modelResponse.isStreaming()) { return withStreamingRetry(request, handler, JdkFlowAdapter.flowPublisherToFlux(modelResponse.getStream()), - attempt, currentDelay); + 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)); - } - - private long nextDelay(long currentDelay) { - return RetryDelayCalculator.nextDelay(currentDelay, backoffMultiplier, maxDelay); + 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) { @@ -256,6 +328,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 +392,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 +458,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..76fda5522 --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/interceptors/ModelRetryInterceptorControlledTest.java @@ -0,0 +1,269 @@ +/* + * 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.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; +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 reactor.adapter.JdkFlowAdapter; +import reactor.core.publisher.Flux; + +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); + } + + /** + * 验证流式重试的响应式退避按切片观察运行状态：取消运行后等待在下一个切片终止， + * 而不是等满整个配置的退避时长。 + */ + @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(); + } + + 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/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)); } 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.