Suitable for single-replica deployments and for tests. State is per JVM: it is lost on restart + * and not shared between replicas, so in a multi-replica deployment every replica trips its own + * circuit independently. Use a distributed store when that matters. + * + *
Entries are created lazily on first write and removed once a tool is fully healthy again, so
+ * tools that never fail cost nothing.
+ */
+public class InMemoryToolCircuitBreakerStore implements ToolCircuitBreakerStore {
+
+ private final Map This class holds all policy and no transport. It is a plain object with no reactive or agent
+ * dependencies, so the state machine can be tested directly against a fixed {@link Clock}. {@link
+ * ToolCircuitBreakerMiddleware} is the adapter that wires it into an agent.
+ *
+ * A ReAct loop reasons, calls a tool, reads the result, and reasons again. When a tool returns an
+ * error the model commonly retries it, because from the model's point of view a single failure looks
+ * incidental. Each retry costs a model call, an outbound request and seconds of latency, and against
+ * a dependency that is genuinely down it fails again.
+ *
+ * A classic breaker sits between caller and dependency and fails fast once open. Here there is a
+ * better option: stop advertising the tool. A tool absent from the schema list is a tool the model
+ * cannot ask for, which removes the failure loop at the source rather than absorbing it. The model
+ * needs no prompt telling it to avoid the tool and cannot argue with the decision.
+ *
+ * Nothing runs in the background: {@link #state(String)} derives the state from the stored
+ * snapshot and the current time, so a cooldown that elapsed while the agent was idle is recognised
+ * on the next read. There is no timer to leak and no scheduler to configure.
+ *
+ * Safe for concurrent use as long as the {@link ToolCircuitBreakerStore} is. Read-modify-write
+ * sequences are not globally serialised: two tool calls failing at the same instant may both observe
+ * the threshold and call {@link ToolCircuitBreakerStore#open(String, long)}. That is harmless —
+ * opening is idempotent apart from advancing the generation, so the worst case is one extra backoff
+ * step.
+ */
+public class ToolCircuitBreaker {
+
+ private static final Logger logger = LoggerFactory.getLogger(ToolCircuitBreaker.class);
+
+ private final ToolCircuitBreakerConfig config;
+ private final ToolCircuitBreakerStore store;
+ private final Clock clock;
+
+ /**
+ * Create a breaker with the default in-process store and the system clock.
+ *
+ * @param config supervision and backoff policy
+ */
+ public ToolCircuitBreaker(ToolCircuitBreakerConfig config) {
+ this(config, new InMemoryToolCircuitBreakerStore(), Clock.systemUTC());
+ }
+
+ /**
+ * Create a breaker with a caller-supplied store and the system clock.
+ *
+ * @param config supervision and backoff policy
+ * @param store state persistence
+ */
+ public ToolCircuitBreaker(ToolCircuitBreakerConfig config, ToolCircuitBreakerStore store) {
+ this(config, store, Clock.systemUTC());
+ }
+
+ /**
+ * Create a breaker with a caller-supplied store and clock.
+ *
+ * @param config supervision and backoff policy
+ * @param store state persistence
+ * @param clock time source; inject a fixed or adjustable clock in tests to step through
+ * cooldowns without sleeping. Cooldowns compare a stored wall-clock stamp against this
+ * clock, so a backwards jump (a manual correction, not NTP slew) can hold a tool back for
+ * up to the size of that jump. Use {@link #reset(String)} to clear it immediately.
+ */
+ public ToolCircuitBreaker(
+ ToolCircuitBreakerConfig config, ToolCircuitBreakerStore store, Clock clock) {
+ this.config = Objects.requireNonNull(config, "config must not be null");
+ this.store = Objects.requireNonNull(store, "store must not be null");
+ this.clock = Objects.requireNonNull(clock, "clock must not be null");
+ }
+
+ /**
+ * Whether this tool is under supervision.
+ *
+ * Exclusions win over both the monitored set and {@code monitorAllTools}, so a tool can be
+ * exempted without editing the supervised list.
+ *
+ * @param toolName tool to test; null is never supervised
+ * @return true when failures of this tool count towards a trip
+ */
+ public boolean supervises(String toolName) {
+ if (toolName == null || !config.isEnabled()) {
+ return false;
+ }
+ if (config.getExcludedTools().contains(toolName)) {
+ return false;
+ }
+ return config.isMonitorAllTools() || config.getMonitoredTools().contains(toolName);
+ }
+
+ /**
+ * Current state of a tool's circuit, derived from stored state and the current time.
+ *
+ * Unsupervised tools always report {@link ToolCircuitState#CLOSED}.
+ *
+ * @param toolName tool to inspect
+ * @return the current state, never null
+ */
+ public ToolCircuitState state(String toolName) {
+ if (!supervises(toolName)) {
+ return ToolCircuitState.CLOSED;
+ }
+ ToolCircuitSnapshot snapshot = store.snapshot(toolName);
+ if (!snapshot.isOpen()) {
+ return ToolCircuitState.CLOSED;
+ }
+ return hasCooldownElapsed(snapshot) ? ToolCircuitState.HALF_OPEN : ToolCircuitState.OPEN;
+ }
+
+ /**
+ * Whether the tool should be kept out of the schema list offered to the model.
+ *
+ * True only in {@link ToolCircuitState#OPEN}: a half-open circuit deliberately advertises the
+ * tool again so the model's next call doubles as the recovery probe.
+ *
+ * @param toolName tool to test
+ * @return true when the tool must not be advertised
+ */
+ public boolean isWithheld(String toolName) {
+ return state(toolName) == ToolCircuitState.OPEN;
+ }
+
+ /**
+ * Cooldown for a given trip generation: {@code initial * multiplier^(generation-1)}, capped at
+ * the configured maximum.
+ *
+ * @param generation trip count, 1 for the first trip
+ * @return the cooldown, or {@link Duration#ZERO} for a circuit that has never tripped
+ */
+ public Duration cooldownFor(long generation) {
+ if (generation <= 0L) {
+ return Duration.ZERO;
+ }
+ double scaled =
+ config.getInitialCooldown().toMillis()
+ * Math.pow(config.getBackoffMultiplier(), (double) generation - 1.0);
+ // A large generation overflows to Infinity; both that and any value past the ceiling clamp
+ // to maxCooldown, so backoff can never isolate a tool indefinitely.
+ if (!Double.isFinite(scaled) || scaled >= config.getMaxCooldown().toMillis()) {
+ return config.getMaxCooldown();
+ }
+ return Duration.ofMillis((long) Math.ceil(scaled));
+ }
+
+ /**
+ * Record a successful tool execution.
+ *
+ * In {@link ToolCircuitState#HALF_OPEN} this closes the circuit. In {@link
+ * ToolCircuitState#CLOSED} it clears any partial failure streak, which is what makes the
+ * threshold count consecutive failures and stops an occasional blip from ever tripping a
+ * healthy tool.
+ *
+ * @param toolName tool that succeeded
+ */
+ public void recordSuccess(String toolName) {
+ if (!supervises(toolName)) {
+ return;
+ }
+ ToolCircuitState current = state(toolName);
+ if (current == ToolCircuitState.OPEN) {
+ // Withheld yet still executed: tolerated, see recordFailure.
+ return;
+ }
+ if (current == ToolCircuitState.HALF_OPEN) {
+ store.close(toolName);
+ store.resetFailures(toolName);
+ logger.info("Tool circuit closed after successful probe: tool={}", toolName);
+ return;
+ }
+ if (store.failureCount(toolName) > 0L) {
+ store.resetFailures(toolName);
+ logger.debug("Tool circuit failure streak cleared by success: tool={}", toolName);
+ }
+ }
+
+ /**
+ * Record a failed tool execution, tripping the circuit once the threshold is reached.
+ *
+ * Only genuine execution failures belong here. A call refused by permission rules or
+ * cancelled by the user says nothing about the health of the dependency and must not count
+ * towards a trip.
+ *
+ * @param toolName tool that failed
+ */
+ public void recordFailure(String toolName) {
+ if (!supervises(toolName)) {
+ return;
+ }
+ ToolCircuitState current = state(toolName);
+ if (current == ToolCircuitState.OPEN) {
+ // The tool was withheld, so the model should not have been able to call it. This is
+ // still reachable: the model may have chosen the call in the same turn the circuit
+ // tripped. Ignore it rather than counting a failure the policy never authorised.
+ logger.debug(
+ "Ignoring failure of withheld tool, likely decided before the circuit opened:"
+ + " tool={}",
+ toolName);
+ return;
+ }
+ if (current == ToolCircuitState.HALF_OPEN) {
+ long generation = store.open(toolName, clock.millis());
+ logger.warn(
+ "Tool circuit re-opened after failed probe: tool={}, generation={},"
+ + " cooldown={}",
+ toolName,
+ generation,
+ cooldownFor(generation));
+ return;
+ }
+ long failures = store.recordFailure(toolName);
+ if (failures < config.getFailureThreshold()) {
+ logger.debug(
+ "Tool failure recorded: tool={}, consecutiveFailures={}/{}",
+ toolName,
+ failures,
+ config.getFailureThreshold());
+ return;
+ }
+ long generation = store.open(toolName, clock.millis());
+ // Clear the streak on trip so the counter always means "failures seen while closed".
+ store.resetFailures(toolName);
+ logger.warn(
+ "Tool circuit opened: tool={}, consecutiveFailures={}, generation={}, cooldown={}."
+ + " The tool will not be offered to the model until the cooldown elapses.",
+ toolName,
+ failures,
+ generation,
+ cooldownFor(generation));
+ }
+
+ /**
+ * Force a tool back to {@link ToolCircuitState#CLOSED}, discarding its failure streak and
+ * accumulated backoff.
+ *
+ * Intended for operators who know a dependency is healthy again and do not want to wait out
+ * the cooldown.
+ *
+ * @param toolName tool to reset
+ */
+ public void reset(String toolName) {
+ store.close(toolName);
+ store.resetFailures(toolName);
+ logger.info("Tool circuit manually reset: tool={}", toolName);
+ }
+
+ /**
+ * The policy in force.
+ *
+ * @return the configuration this breaker was built with
+ */
+ public ToolCircuitBreakerConfig getConfig() {
+ return config;
+ }
+
+ private boolean hasCooldownElapsed(ToolCircuitSnapshot snapshot) {
+ Duration cooldown = cooldownFor(snapshot.generation());
+ return clock.millis() >= snapshot.openedAtEpochMilli() + cooldown.toMillis();
+ }
+}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java
new file mode 100644
index 0000000000..f847a8e477
--- /dev/null
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java
@@ -0,0 +1,327 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.tool.circuitbreaker;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/**
+ * Policy for {@link ToolCircuitBreaker}: which tools are supervised, when a circuit trips, and how
+ * long it stays open.
+ *
+ * A breaker is only ever applied to tools named by {@link Builder#monitorTools(Collection)} (or
+ * to every tool once {@link Builder#monitorAllTools(boolean)} is set). With neither configured the
+ * breaker is inert.
+ *
+ * That default is deliberate. Withholding a tool is the right response for a flaky external
+ * dependency, but the wrong response for infrastructure the agent cannot work without: breaking a
+ * database or filesystem tool does not degrade the agent gracefully, it cripples it. Requiring the
+ * supervised set to be named means a breaker can never take down a tool its author never
+ * considered. Prefer tools whose loss leaves the main flow viable.
+ *
+ * Cooldown is {@code min(initialCooldown * backoffMultiplier^(generation-1), maxCooldown)}. With
+ * the defaults (60s, x2, capped at 600s) successive trips wait 60s, 120s, 240s, 480s, 600s, 600s...
+ * A tool that keeps failing is isolated for longer, which cuts both the probe traffic aimed at a
+ * struggling dependency and the tokens spent re-discovering that it is still down. The cap stops
+ * backoff from isolating a tool for hours after a long outage.
+ */
+public final class ToolCircuitBreakerConfig {
+
+ private final boolean enabled;
+ private final Set Read the class-level note on infrastructure tools before enabling this. Opt-out
+ * supervision also covers tools the framework adds per call, such as the structured-output
+ * tool used to produce a final answer; withholding one of those does not degrade the agent,
+ * it stops it completing. Name the unstable dependencies instead, or exclude the tools that
+ * must always stay reachable.
+ *
+ * @param monitorAllTools true to switch to opt-out supervision
+ * @return this builder
+ */
+ public Builder monitorAllTools(boolean monitorAllTools) {
+ this.monitorAllTools = monitorAllTools;
+ return this;
+ }
+
+ /**
+ * Exempt the named tools from supervision, overriding every other setting.
+ *
+ * @param toolNames tool names to exempt
+ * @return this builder
+ */
+ public Builder excludeTools(Collection It occupies two interception points, which together close the state machine:
+ *
+ * Filtering happens per turn on a copy of the schema list. Nothing registered on the {@link
+ * io.agentscope.core.tool.Toolkit} is mutated, so a circuit tripped while serving one session cannot
+ * remove a tool from a concurrent session, and the tool registrations the application declared stay
+ * authoritative. Recovery needs no repair step for the same reason: once the breaker stops
+ * withholding a tool, the unfiltered list is already correct.
+ *
+ * Usage
+ *
+ * Share one breaker (and therefore one store) across the agents that call the same dependency, so
+ * they learn from each other's failures instead of each discovering the outage separately.
+ */
+public class ToolCircuitBreakerMiddleware implements MiddlewareBase {
+
+ private static final Logger logger =
+ LoggerFactory.getLogger(ToolCircuitBreakerMiddleware.class);
+
+ private final ToolCircuitBreaker breaker;
+
+ /**
+ * Wrap a breaker as middleware.
+ *
+ * @param breaker the breaker holding policy and state
+ */
+ public ToolCircuitBreakerMiddleware(ToolCircuitBreaker breaker) {
+ this.breaker = Objects.requireNonNull(breaker, "breaker must not be null");
+ }
+
+ /**
+ * Convenience constructor building a breaker with the default in-process store.
+ *
+ * @param config supervision and backoff policy
+ */
+ public ToolCircuitBreakerMiddleware(ToolCircuitBreakerConfig config) {
+ this(new ToolCircuitBreaker(config));
+ }
+
+ /**
+ * The breaker being applied, exposed so callers can inspect circuit state or reset a tool.
+ *
+ * @return the underlying breaker
+ */
+ public ToolCircuitBreaker getBreaker() {
+ return breaker;
+ }
+
+ @Override
+ public Flux Only {@link ToolResultState#ERROR} counts as a failure. {@code DENIED} is a policy refusal
+ * and {@code INTERRUPTED} a cancellation — neither is evidence about the dependency, and
+ * counting them would let a user who declines a confirmation prompt trip the circuit. {@code
+ * RUNNING} marks a suspended call whose outcome is not known yet.
+ */
+ private void recordOutcome(AgentEvent event) {
+ if (!(event instanceof ToolResultEndEvent result)) {
+ return;
+ }
+ String toolName = result.getToolCallName();
+ ToolResultState state = result.getState();
+ if (toolName == null || state == null) {
+ return;
+ }
+ if (state == ToolResultState.ERROR) {
+ breaker.recordFailure(toolName);
+ } else if (state == ToolResultState.SUCCESS) {
+ breaker.recordSuccess(toolName);
+ }
+ }
+}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerStore.java b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerStore.java
new file mode 100644
index 0000000000..58d6aacbc0
--- /dev/null
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerStore.java
@@ -0,0 +1,94 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.tool.circuitbreaker;
+
+/**
+ * Persistence contract for tool circuit-breaker state.
+ *
+ * Implementations are pure state holders: they must not apply the backoff policy, decide when a
+ * circuit trips, or consult a clock. All policy lives in {@link ToolCircuitBreaker}, which keeps
+ * this SPI stable when the policy evolves and makes the policy unit-testable without a store.
+ *
+ * {@link InMemoryToolCircuitBreakerStore} is the default and is sufficient for a single
+ * process. A distributed implementation (for example the Redis-backed store in
+ * {@code agentscope-extensions-redis}) lets every replica share one view of a broken tool, so a
+ * dependency that node A found down is not re-probed by nodes B and C in parallel.
+ *
+ * Implementations must be safe for concurrent use from multiple threads. {@link
+ * #recordFailure(String)} and {@link #open(String, long)} must be atomic, since concurrent tool
+ * calls in one ReAct turn race on both.
+ */
+public interface ToolCircuitBreakerStore {
+
+ /**
+ * Atomically increment the consecutive-failure counter and return the new value.
+ *
+ * @param toolName tool being counted
+ * @return the counter value after incrementing, starting at 1
+ */
+ long recordFailure(String toolName);
+
+ /**
+ * Clear the consecutive-failure counter.
+ *
+ * Called whenever a tool succeeds, which is what makes the threshold count
+ * consecutive failures rather than lifetime failures.
+ *
+ * @param toolName tool to reset
+ */
+ void resetFailures(String toolName);
+
+ /**
+ * Read the consecutive-failure counter without modifying it.
+ *
+ * @param toolName tool to read
+ * @return current count, or {@code 0} when nothing is recorded
+ */
+ long failureCount(String toolName);
+
+ /**
+ * Atomically move the circuit to OPEN: increment the backoff generation and stamp the open
+ * instant, returning the new generation.
+ *
+ * The caller supplies the timestamp so that it shares a clock with the cooldown comparison
+ * in {@link ToolCircuitBreaker}; a store must never substitute its own clock.
+ *
+ * @param toolName tool to trip
+ * @param openedAtEpochMilli instant the circuit opened, in epoch milliseconds
+ * @return the backoff generation after incrementing, starting at 1
+ */
+ long open(String toolName, long openedAtEpochMilli);
+
+ /**
+ * Reset the circuit to CLOSED, discarding both the open timestamp and the backoff generation.
+ *
+ * Dropping the generation means a tool that recovers starts its next incident from the
+ * initial cooldown instead of inheriting an old, long backoff.
+ *
+ * @param toolName tool to close
+ */
+ void close(String toolName);
+
+ /**
+ * Read the trip state in a single round trip.
+ *
+ * @param toolName tool to read
+ * @return current snapshot, never null; {@link ToolCircuitSnapshot#CLOSED} when no state exists
+ */
+ ToolCircuitSnapshot snapshot(String toolName);
+}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java
new file mode 100644
index 0000000000..ddd27e8e9e
--- /dev/null
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java
@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.tool.circuitbreaker;
+
+/**
+ * Immutable point-in-time view of a tool's trip state, read in a single store round trip.
+ *
+ * The cooldown duration is deliberately not part of the snapshot: it is derived from
+ * {@code generation} by {@link ToolCircuitBreaker#cooldownFor(long)}, so changing the backoff
+ * policy takes effect immediately and never has to be migrated in the store.
+ *
+ * @param generation number of times the circuit has tripped, driving exponential backoff;
+ * {@code 0} means it has never tripped
+ * @param openedAtEpochMilli wall-clock instant the circuit was last opened, or {@code 0} when
+ * the circuit is not open — a positive value is the sole marker of the OPEN state, so no
+ * separate boolean has to be kept consistent with it
+ */
+public record ToolCircuitSnapshot(long generation, long openedAtEpochMilli) {
+
+ /** Snapshot of a tool that has never tripped. */
+ public static final ToolCircuitSnapshot CLOSED = new ToolCircuitSnapshot(0L, 0L);
+
+ /**
+ * Whether the circuit is currently open, ignoring whether its cooldown has elapsed.
+ *
+ * An open circuit whose cooldown has elapsed is reported as
+ * {@link ToolCircuitState#HALF_OPEN} by {@link ToolCircuitBreaker#state(String)}; this method
+ * only reports the persisted flag.
+ *
+ * @return true when an open timestamp is recorded
+ */
+ public boolean isOpen() {
+ return openedAtEpochMilli > 0L;
+ }
+}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitState.java b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitState.java
new file mode 100644
index 0000000000..de833f92bb
--- /dev/null
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitState.java
@@ -0,0 +1,53 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.tool.circuitbreaker;
+
+/**
+ * Lifecycle state of a single tool's circuit.
+ *
+ * The state is always derived from the persisted snapshot plus the current time
+ * rather than stored directly, so a cooldown that elapsed while no traffic flowed is observed
+ * correctly on the next read without any background scheduler.
+ *
+ * Named after a closed electrical circuit — current flows, so {@code CLOSED} means
+ * "healthy", not "unavailable".
+ */
+ CLOSED,
+
+ /**
+ * Tripped and still cooling down: the tool is withheld from the model.
+ */
+ OPEN,
+
+ /**
+ * Cooldown elapsed: the tool is exposed again as a single probe.
+ *
+ * A successful probe closes the circuit; a failing probe re-opens it with the next
+ * (longer) backoff generation.
+ */
+ HALF_OPEN
+}
diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java
new file mode 100644
index 0000000000..9a5a286660
--- /dev/null
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.tool.circuitbreaker;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.Test;
+
+/** Contract of {@link InMemoryToolCircuitBreakerStore}, including its atomicity guarantees. */
+class InMemoryToolCircuitBreakerStoreTest {
+
+ private static final String TOOL = "query_weather";
+ private static final String OTHER_TOOL = "query_news";
+
+ private final InMemoryToolCircuitBreakerStore store = new InMemoryToolCircuitBreakerStore();
+
+ @Test
+ void unknownToolReadsAsClosedWithNoFailures() {
+ assertEquals(0L, store.failureCount(TOOL));
+ assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
+ assertFalse(store.snapshot(TOOL).isOpen());
+ }
+
+ @Test
+ void failureCounterStartsAtOneAndIncrements() {
+ assertEquals(1L, store.recordFailure(TOOL));
+ assertEquals(2L, store.recordFailure(TOOL));
+ assertEquals(2L, store.failureCount(TOOL));
+ }
+
+ @Test
+ void resetClearsOnlyTheNamedToolsCounter() {
+ store.recordFailure(TOOL);
+ store.recordFailure(OTHER_TOOL);
+
+ store.resetFailures(TOOL);
+
+ assertEquals(0L, store.failureCount(TOOL));
+ assertEquals(1L, store.failureCount(OTHER_TOOL));
+ }
+
+ @Test
+ void openStampsTimestampAndAdvancesGeneration() {
+ assertEquals(1L, store.open(TOOL, 1_000L));
+
+ ToolCircuitSnapshot first = store.snapshot(TOOL);
+ assertTrue(first.isOpen());
+ assertEquals(1L, first.generation());
+ assertEquals(1_000L, first.openedAtEpochMilli());
+
+ assertEquals(2L, store.open(TOOL, 5_000L));
+
+ ToolCircuitSnapshot second = store.snapshot(TOOL);
+ assertEquals(2L, second.generation());
+ assertEquals(5_000L, second.openedAtEpochMilli());
+ }
+
+ @Test
+ void closeDiscardsGenerationSoBackoffRestarts() {
+ store.open(TOOL, 1_000L);
+ store.open(TOOL, 2_000L);
+
+ store.close(TOOL);
+
+ assertFalse(store.snapshot(TOOL).isOpen());
+ assertEquals(0L, store.snapshot(TOOL).generation());
+ assertEquals(1L, store.open(TOOL, 3_000L));
+ }
+
+ @Test
+ void toolsDoNotShareState() {
+ store.open(TOOL, 1_000L);
+
+ assertTrue(store.snapshot(TOOL).isOpen());
+ assertFalse(store.snapshot(OTHER_TOOL).isOpen());
+ }
+
+ @Test
+ void concurrentFailureCountsAreNotLost() throws Exception {
+ int threads = 8;
+ int perThread = 500;
+
+ runConcurrently(threads, perThread, () -> store.recordFailure(TOOL));
+
+ assertEquals((long) threads * perThread, store.failureCount(TOOL));
+ }
+
+ @Test
+ void concurrentOpensYieldContiguousGenerations() throws Exception {
+ int threads = 8;
+ int perThread = 200;
+ AtomicLong maxGeneration = new AtomicLong();
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ for (int t = 0; t < threads; t++) {
+ pool.submit(
+ () -> {
+ for (int i = 0; i < perThread; i++) {
+ long generation = store.open(TOOL, 1_000L + i);
+ maxGeneration.accumulateAndGet(generation, Math::max);
+ }
+ });
+ }
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS));
+ } finally {
+ pool.shutdownNow();
+ }
+
+ // Every open must observe a distinct, gap-free generation, so the highest value seen equals
+ // the number of opens performed.
+ assertEquals((long) threads * perThread, maxGeneration.get());
+ assertEquals((long) threads * perThread, store.snapshot(TOOL).generation());
+ }
+
+ private void runConcurrently(int threads, int perThread, Runnable task) throws Exception {
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ for (int t = 0; t < threads; t++) {
+ pool.submit(
+ () -> {
+ for (int i = 0; i < perThread; i++) {
+ task.run();
+ }
+ });
+ }
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS));
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+}
diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/MutableClock.java b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/MutableClock.java
new file mode 100644
index 0000000000..e5e7ba324d
--- /dev/null
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/MutableClock.java
@@ -0,0 +1,61 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.tool.circuitbreaker;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+
+/**
+ * Hand-advanced {@link Clock} so cooldown transitions can be exercised without sleeping.
+ *
+ * Every circuit transition is a function of stored state plus "now", so driving time explicitly
+ * keeps these tests deterministic and instant.
+ */
+final class MutableClock extends Clock {
+
+ private final ZoneId zone;
+ private Instant instant;
+
+ MutableClock(Instant start) {
+ this(start, ZoneId.of("UTC"));
+ }
+
+ private MutableClock(Instant instant, ZoneId zone) {
+ this.instant = instant;
+ this.zone = zone;
+ }
+
+ void advance(Duration amount) {
+ instant = instant.plus(amount);
+ }
+
+ @Override
+ public ZoneId getZone() {
+ return zone;
+ }
+
+ @Override
+ public Clock withZone(ZoneId newZone) {
+ return new MutableClock(instant, newZone);
+ }
+
+ @Override
+ public Instant instant() {
+ return instant;
+ }
+}
diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java
new file mode 100644
index 0000000000..ea89eb0415
--- /dev/null
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java
@@ -0,0 +1,297 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.tool.circuitbreaker;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.agentscope.core.event.AgentEvent;
+import io.agentscope.core.event.ToolResultEndEvent;
+import io.agentscope.core.message.Msg;
+import io.agentscope.core.message.MsgRole;
+import io.agentscope.core.message.TextBlock;
+import io.agentscope.core.message.ToolResultState;
+import io.agentscope.core.middleware.ActingInput;
+import io.agentscope.core.middleware.ReasoningInput;
+import io.agentscope.core.model.GenerateOptions;
+import io.agentscope.core.model.ToolSchema;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+/**
+ * End-to-end behaviour of {@link ToolCircuitBreakerMiddleware}: outcomes observed on the acting
+ * stream must drive what the reasoning phase advertises to the model.
+ */
+class ToolCircuitBreakerMiddlewareTest {
+
+ private static final String REPLY_ID = "reply-1";
+ private static final String WEATHER = "query_weather";
+ private static final String DATABASE = "query_database";
+
+ private final MutableClock clock = new MutableClock(Instant.parse("2026-01-01T00:00:00Z"));
+
+ // ==================== Reasoning: withholding tripped tools ====================
+
+ @Test
+ void trippedToolIsRemovedFromTheToolsOfferedToTheModel() {
+ ToolCircuitBreakerMiddleware middleware = middleware(1);
+
+ failTool(middleware, WEATHER);
+
+ List With the in-process store each replica has to rediscover an outage for itself, so an N-replica
+ * deployment sends roughly N times the failing traffic and burns N times the tokens before the tool
+ * is withheld everywhere. Sharing the state through Redis means the first replica to trip a circuit
+ * withholds the tool for all of them, and the state survives a restart or a rescheduled pod.
+ *
+ * Two keys per tool, both addressed individually so the store works unchanged on Redis Cluster —
+ * no multi-key script needs its arguments to share a hash slot:
+ *
+ * Encoding generation and timestamp in one value keeps {@link #snapshot(String)} — the hot read,
+ * executed for every supervised tool on every reasoning turn — down to a single {@code GET}.
+ *
+ * {@link #recordFailure(String)} and {@link #open(String, long)} are Lua scripts, so their
+ * read-modify-write steps cannot interleave. Doing {@code INCR} and {@code EXPIRE} as two round
+ * trips would leave a counter without a TTL whenever the second call is lost, and computing the next
+ * generation client-side would let two replicas tripping at once write the same generation.
+ *
+ * Both keys carry a TTL so tools that misbehave once do not accumulate state forever. Keep the
+ * TTL comfortably longer than the breaker's maximum cooldown: if an open circuit's key expires
+ * mid-cooldown the tool is offered again early, which fails open — safe, but not what was
+ * configured. The default of 24h clears the default 600s ceiling by a wide margin.
+ */
+public class RedisToolCircuitBreakerStore implements ToolCircuitBreakerStore {
+
+ private static final Logger logger =
+ LoggerFactory.getLogger(RedisToolCircuitBreakerStore.class);
+
+ private static final String DEFAULT_KEY_PREFIX = "agentscope:tool-cb:";
+ private static final Duration DEFAULT_TTL = Duration.ofHours(24);
+
+ private static final String FAILURE_SUFFIX = ":fail";
+ private static final String CIRCUIT_SUFFIX = ":circuit";
+
+ /**
+ * Increment the failure counter and refresh its TTL in one step.
+ *
+ * KEYS[1] = failure key; ARGV[1] = TTL seconds. Returns the new count.
+ */
+ private static final String INCREMENT_FAILURE_SCRIPT =
+ "local count = redis.call('INCR', KEYS[1]) "
+ + "redis.call('EXPIRE', KEYS[1], ARGV[1]) "
+ + "return count";
+
+ /**
+ * Advance the generation and stamp the open instant in one step.
+ *
+ * KEYS[1] = circuit key; ARGV[1] = open instant in epoch millis; ARGV[2] = TTL seconds.
+ * Returns the new generation.
+ */
+ private static final String OPEN_NEXT_GENERATION_SCRIPT =
+ "local current = redis.call('GET', KEYS[1]) local generation = 0 if current then "
+ + " local sep = string.find(current, ':', 1, true) if sep then generation ="
+ + " tonumber(string.sub(current, 1, sep - 1)) or 0 end end generation = generation"
+ + " + 1 redis.call('SET', KEYS[1], generation .. ':' .. ARGV[1], 'EX', ARGV[2])"
+ + " return generation";
+
+ private final RedisClientAdapter client;
+ private final String keyPrefix;
+ private final long ttlSeconds;
+
+ /**
+ * Create a store with the default key prefix ({@code agentscope:tool-cb:}) and a 24h TTL.
+ *
+ * @param client Redis client adapter
+ */
+ public RedisToolCircuitBreakerStore(RedisClientAdapter client) {
+ this(client, DEFAULT_KEY_PREFIX, DEFAULT_TTL);
+ }
+
+ /**
+ * Create a store with an explicit key prefix and TTL.
+ *
+ * @param client Redis client adapter
+ * @param keyPrefix prefix for every key, letting environments share one Redis instance
+ * @param stateTtl how long unused state is retained; must be positive and should exceed the
+ * breaker's maximum cooldown
+ */
+ public RedisToolCircuitBreakerStore(
+ RedisClientAdapter client, String keyPrefix, Duration stateTtl) {
+ this.client = Objects.requireNonNull(client, "client must not be null");
+ if (keyPrefix == null || keyPrefix.isBlank()) {
+ throw new IllegalArgumentException("keyPrefix must not be blank");
+ }
+ if (stateTtl == null || stateTtl.isNegative() || stateTtl.isZero()) {
+ throw new IllegalArgumentException("stateTtl must be positive, got " + stateTtl);
+ }
+ this.keyPrefix = keyPrefix;
+ this.ttlSeconds = Math.max(1L, stateTtl.toSeconds());
+ }
+
+ @Override
+ public long recordFailure(String toolName) {
+ return client.evalScript(
+ INCREMENT_FAILURE_SCRIPT,
+ List.of(failureKey(toolName)),
+ List.of(Long.toString(ttlSeconds)));
+ }
+
+ @Override
+ public void resetFailures(String toolName) {
+ client.deleteKeys(failureKey(toolName));
+ }
+
+ @Override
+ public long failureCount(String toolName) {
+ return parseLong(client.get(failureKey(toolName)));
+ }
+
+ @Override
+ public long open(String toolName, long openedAtEpochMilli) {
+ return client.evalScript(
+ OPEN_NEXT_GENERATION_SCRIPT,
+ List.of(circuitKey(toolName)),
+ List.of(Long.toString(openedAtEpochMilli), Long.toString(ttlSeconds)));
+ }
+
+ @Override
+ public void close(String toolName) {
+ client.deleteKeys(circuitKey(toolName));
+ }
+
+ @Override
+ public ToolCircuitSnapshot snapshot(String toolName) {
+ String value = client.get(circuitKey(toolName));
+ if (value == null || value.isEmpty()) {
+ return ToolCircuitSnapshot.CLOSED;
+ }
+ int separator = value.indexOf(':');
+ if (separator <= 0 || separator == value.length() - 1) {
+ // Unreadable value: treat as closed rather than withholding a tool forever on the
+ // strength of state nobody can interpret.
+ logger.warn(
+ "Ignoring malformed circuit state for tool={}, value={}. Treating the circuit"
+ + " as closed.",
+ toolName,
+ value);
+ return ToolCircuitSnapshot.CLOSED;
+ }
+ long generation = parseLong(value.substring(0, separator));
+ long openedAt = parseLong(value.substring(separator + 1));
+ if (generation <= 0L || openedAt <= 0L) {
+ logger.warn(
+ "Ignoring out-of-range circuit state for tool={}, value={}. Treating the"
+ + " circuit as closed.",
+ toolName,
+ value);
+ return ToolCircuitSnapshot.CLOSED;
+ }
+ return new ToolCircuitSnapshot(generation, openedAt);
+ }
+
+ private String failureKey(String toolName) {
+ return keyPrefix + toolName + FAILURE_SUFFIX;
+ }
+
+ private String circuitKey(String toolName) {
+ return keyPrefix + toolName + CIRCUIT_SUFFIX;
+ }
+
+ private static long parseLong(String value) {
+ if (value == null || value.isEmpty()) {
+ return 0L;
+ }
+ try {
+ return Long.parseLong(value.trim());
+ } catch (NumberFormatException e) {
+ return 0L;
+ }
+ }
+}
diff --git a/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java
new file mode 100644
index 0000000000..5b13b10e0b
--- /dev/null
+++ b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java
@@ -0,0 +1,261 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.extensions.redis.circuitbreaker;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.agentscope.core.tool.circuitbreaker.ToolCircuitSnapshot;
+import io.agentscope.extensions.redis.state.RedisClientAdapter;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Client-side behaviour of {@link RedisToolCircuitBreakerStore}: key naming, argument passing and
+ * the decoding of persisted circuit values.
+ *
+ * Scope note: the two Lua scripts are executed by Redis, so a fake client cannot run them. These
+ * tests cover the Java side — which keys are addressed, which arguments the scripts receive, and how
+ * stored values are decoded, including values a healthy writer would never produce. The scripts'
+ * server-side effects need a live Redis to verify.
+ */
+class RedisToolCircuitBreakerStoreTest {
+
+ private static final String TOOL = "query_weather";
+
+ private final RecordingRedisClient client = new RecordingRedisClient();
+
+ // ==================== Key naming ====================
+
+ @Test
+ void keysCarryThePrefixAndDistinctSuffixes() {
+ RedisToolCircuitBreakerStore store = store();
+
+ store.recordFailure(TOOL);
+ store.open(TOOL, 1_000L);
+
+ assertEquals(
+ List.of("cb:query_weather:fail", "cb:query_weather:circuit"), client.scriptKeys);
+ }
+
+ @Test
+ void resetFailuresDeletesOnlyTheCounter() {
+ RedisToolCircuitBreakerStore store = store();
+
+ store.resetFailures(TOOL);
+
+ assertEquals(List.of("cb:query_weather:fail"), client.deleted);
+ }
+
+ @Test
+ void closeDeletesOnlyTheCircuitKey() {
+ RedisToolCircuitBreakerStore store = store();
+
+ store.close(TOOL);
+
+ assertEquals(List.of("cb:query_weather:circuit"), client.deleted);
+ }
+
+ // ==================== Script arguments ====================
+
+ @Test
+ void failureScriptReceivesTheTtlInSeconds() {
+ RedisToolCircuitBreakerStore store =
+ new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofMinutes(30));
+
+ store.recordFailure(TOOL);
+
+ assertEquals(List.of("1800"), client.scriptArgs.get(0));
+ }
+
+ @Test
+ void openScriptReceivesTheTimestampThenTheTtl() {
+ RedisToolCircuitBreakerStore store =
+ new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofHours(24));
+
+ store.open(TOOL, 1_767_225_600_000L);
+
+ assertEquals(List.of("1767225600000", "86400"), client.scriptArgs.get(0));
+ }
+
+ @Test
+ void subSecondTtlIsFlooredToOneSecondSoKeysNeverPersistForever() {
+ RedisToolCircuitBreakerStore store =
+ new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofMillis(200));
+
+ store.recordFailure(TOOL);
+
+ assertEquals(List.of("1"), client.scriptArgs.get(0));
+ }
+
+ // ==================== Decoding persisted state ====================
+
+ @Test
+ void snapshotDecodesGenerationAndTimestamp() {
+ RedisToolCircuitBreakerStore store = store();
+ client.values.put("cb:query_weather:circuit", "3:1767225600000");
+
+ ToolCircuitSnapshot snapshot = store.snapshot(TOOL);
+
+ assertTrue(snapshot.isOpen());
+ assertEquals(3L, snapshot.generation());
+ assertEquals(1_767_225_600_000L, snapshot.openedAtEpochMilli());
+ }
+
+ @Test
+ void missingKeyDecodesAsClosed() {
+ RedisToolCircuitBreakerStore store = store();
+
+ assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
+ assertFalse(store.snapshot(TOOL).isOpen());
+ }
+
+ @Test
+ void unreadableValuesFailOpenRatherThanWithholdingForever() {
+ RedisToolCircuitBreakerStore store = store();
+ String key = "cb:query_weather:circuit";
+
+ for (String malformed :
+ List.of("", "garbage", ":", "3:", ":1767225600000", "0:1767225600000", "3:0")) {
+ client.values.put(key, malformed);
+ assertEquals(
+ ToolCircuitSnapshot.CLOSED,
+ store.snapshot(TOOL),
+ "expected a closed circuit for stored value: '" + malformed + "'");
+ }
+ }
+
+ @Test
+ void nonNumericFailureCountReadsAsZero() {
+ RedisToolCircuitBreakerStore store = store();
+ client.values.put("cb:query_weather:fail", "not-a-number");
+
+ assertEquals(0L, store.failureCount(TOOL));
+ }
+
+ @Test
+ void failureCountIsReadFromTheCounterKey() {
+ RedisToolCircuitBreakerStore store = store();
+ client.values.put("cb:query_weather:fail", "7");
+
+ assertEquals(7L, store.failureCount(TOOL));
+ }
+
+ // ==================== Construction ====================
+
+ @Test
+ void constructorRejectsBlankPrefixAndNonPositiveTtl() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new RedisToolCircuitBreakerStore(client, " ", Duration.ofHours(1)));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new RedisToolCircuitBreakerStore(client, "cb:", Duration.ZERO));
+ assertThrows(
+ NullPointerException.class,
+ () -> new RedisToolCircuitBreakerStore(null, "cb:", Duration.ofHours(1)));
+ }
+
+ private RedisToolCircuitBreakerStore store() {
+ return new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofHours(24));
+ }
+
+ /** Fake client recording the keys and arguments each call addresses. */
+ private static final class RecordingRedisClient implements RedisClientAdapter {
+
+ private final Map A classic breaker sits between caller and dependency and fails fast once open. Here there is a
* better option: stop advertising the tool. A tool absent from the schema list is a tool the model
- * cannot ask for, which removes the failure loop at the source rather than absorbing it. The model
- * needs no prompt telling it to avoid the tool and cannot argue with the decision.
+ * cannot ask for, which removes the failure loop at the source rather than absorbing it.
*
* Safe for concurrent use as long as the {@link ToolCircuitBreakerStore} is. Read-modify-write
- * sequences are not globally serialised: two tool calls failing at the same instant may both observe
- * the threshold and call {@link ToolCircuitBreakerStore#open(String, long)}. That is harmless —
- * opening is idempotent apart from advancing the generation, so the worst case is one extra backoff
- * step.
+ * A breaker is shared by every concurrent turn, session and (with a distributed store) replica
+ * that can reach the tool, so every transition is published as one compare-and-set against the exact
+ * snapshot the caller observed. A caller that loses the race re-reads and recomputes, which is what
+ * makes two guarantees hold rather than merely being documented:
+ *
+ * The lease bounds the damage when a probe never reports an outcome (a turn that never calls the
+ * advertised tool, a lost execution, a crashed replica): once it expires another caller may claim the
+ * probe. Results carrying a superseded token are ignored, so a late reply cannot close or re-open a
+ * newer probe.
*/
public class ToolCircuitBreaker {
@@ -95,9 +111,10 @@ public ToolCircuitBreaker(ToolCircuitBreakerConfig config, ToolCircuitBreakerSto
* @param config supervision and backoff policy
* @param store state persistence
* @param clock time source; inject a fixed or adjustable clock in tests to step through
- * cooldowns without sleeping. Cooldowns compare a stored wall-clock stamp against this
- * clock, so a backwards jump (a manual correction, not NTP slew) can hold a tool back for
- * up to the size of that jump. Use {@link #reset(String)} to clear it immediately.
+ * cooldowns without sleeping. Cooldowns and probe leases compare stored wall-clock stamps
+ * against this clock, so a backwards jump (a manual correction, not NTP slew) can hold a
+ * tool back for up to the size of that jump. Use {@link #reset(String)} to clear it
+ * immediately.
*/
public ToolCircuitBreaker(
ToolCircuitBreakerConfig config, ToolCircuitBreakerStore store, Clock clock) {
@@ -128,7 +145,8 @@ public boolean supervises(String toolName) {
/**
* Current state of a tool's circuit, derived from stored state and the current time.
*
- * Unsupervised tools always report {@link ToolCircuitState#CLOSED}.
+ * Unsupervised tools always report {@link ToolCircuitState#CLOSED}. This is a pure query: it
+ * never claims the recovery probe, so it is safe to call for logging or metrics.
*
* @param toolName tool to inspect
* @return the current state, never null
@@ -137,21 +155,19 @@ public ToolCircuitState state(String toolName) {
if (!supervises(toolName)) {
return ToolCircuitState.CLOSED;
}
- ToolCircuitSnapshot snapshot = store.snapshot(toolName);
- if (!snapshot.isOpen()) {
- return ToolCircuitState.CLOSED;
- }
- return hasCooldownElapsed(snapshot) ? ToolCircuitState.HALF_OPEN : ToolCircuitState.OPEN;
+ return stateOf(store.snapshot(toolName), clock.millis());
}
/**
- * Whether the tool should be kept out of the schema list offered to the model.
+ * Whether the tool is inside a cooldown and must be kept out of the schema list.
*
- * True only in {@link ToolCircuitState#OPEN}: a half-open circuit deliberately advertises the
- * tool again so the model's next call doubles as the recovery probe.
+ * A half-open circuit reports false here because the tool may be advertised again — but only
+ * to the single turn that wins {@link #tryAcquireProbe(String)}. Callers deciding what to
+ * advertise should therefore branch on {@link #state(String)} and claim the probe, not rely on
+ * this method alone; it exists as a pure query for observability.
*
* @param toolName tool to test
- * @return true when the tool must not be advertised
+ * @return true when the tool is in an unexpired cooldown
*/
public boolean isWithheld(String toolName) {
return state(toolName) == ToolCircuitState.OPEN;
@@ -180,36 +196,141 @@ public Duration cooldownFor(long generation) {
}
/**
- * Record a successful tool execution.
+ * Try to claim the single recovery probe for a half-open circuit.
*
- * In {@link ToolCircuitState#HALF_OPEN} this closes the circuit. In {@link
- * ToolCircuitState#CLOSED} it clears any partial failure streak, which is what makes the
- * threshold count consecutive failures and stops an occasional blip from ever tripping a
- * healthy tool.
+ * The winner may advertise the tool for one turn and must report the outcome through {@link
+ * #recordSuccess(String, String)} / {@link #recordFailure(String, String)} with the returned
+ * token, or hand the permit back with {@link #releaseProbe(String, String)} if the tool is never
+ * called. Doing neither is not fatal: the claim carries a lease and expires.
*
- * @param toolName tool that succeeded
+ * @param toolName tool to probe
+ * @return the probe token when this caller owns the probe; empty when the circuit is not
+ * half-open or another caller already holds it
*/
- public void recordSuccess(String toolName) {
+ public Optional Used when the advertised tool was never called, or when the call was refused or cancelled
+ * and therefore tested nothing. Does nothing if the claim has already expired or been replaced.
+ *
+ * @param toolName tool whose probe is being released
+ * @param probeToken token returned by {@link #tryAcquireProbe(String)}
+ */
+ public void releaseProbe(String toolName, String probeToken) {
+ if (probeToken == null || !supervises(toolName)) {
return;
}
- if (current == ToolCircuitState.HALF_OPEN) {
- store.close(toolName);
- store.resetFailures(toolName);
- logger.info("Tool circuit closed after successful probe: tool={}", toolName);
+ while (true) {
+ ToolCircuitSnapshot current = store.snapshot(toolName);
+ if (!probeToken.equals(current.probeToken())) {
+ return;
+ }
+ ToolCircuitSnapshot update =
+ new ToolCircuitSnapshot(
+ current.failureCount(),
+ current.generation(),
+ current.openedAtEpochMilli(),
+ null,
+ 0L);
+ if (store.compareAndSet(toolName, current, update)) {
+ logger.debug("Recovery probe released unused: tool={}", toolName);
+ return;
+ }
+ }
+ }
+
+ /**
+ * Record a successful tool execution that did not hold a recovery probe.
+ *
+ * @param toolName tool that succeeded
+ */
+ public void recordSuccess(String toolName) {
+ recordSuccess(toolName, null);
+ }
+
+ /**
+ * Record a successful tool execution.
+ *
+ * A probe success closes the circuit and discards the accumulated backoff. A success while
+ * closed clears any partial failure streak, which is what makes the threshold count consecutive
+ * failures and stops an occasional blip from ever tripping a healthy tool.
+ *
+ * @param toolName tool that succeeded
+ * @param probeToken token from {@link #tryAcquireProbe(String)}, or null when the call was not a
+ * recovery probe
+ */
+ public void recordSuccess(String toolName, String probeToken) {
+ if (!supervises(toolName)) {
return;
}
- if (store.failureCount(toolName) > 0L) {
- store.resetFailures(toolName);
- logger.debug("Tool circuit failure streak cleared by success: tool={}", toolName);
+ while (true) {
+ ToolCircuitSnapshot current = store.snapshot(toolName);
+ long now = clock.millis();
+ ToolCircuitState state = stateOf(current, now);
+ if (state == ToolCircuitState.OPEN) {
+ // Withheld yet still executed: tolerated, see recordFailure.
+ return;
+ }
+ if (state == ToolCircuitState.HALF_OPEN) {
+ if (current.hasActiveProbe(now)
+ && !Objects.equals(probeToken, current.probeToken())) {
+ // Someone else owns the live probe; a superseded result must not close it.
+ return;
+ }
+ if (store.compareAndSet(toolName, current, ToolCircuitSnapshot.CLOSED)) {
+ logger.info("Tool circuit closed after successful probe: tool={}", toolName);
+ return;
+ }
+ continue;
+ }
+ if (current.failureCount() == 0L) {
+ return;
+ }
+ if (store.compareAndSet(toolName, current, ToolCircuitSnapshot.CLOSED)) {
+ logger.debug("Tool circuit failure streak cleared by success: tool={}", toolName);
+ return;
+ }
}
}
+ /**
+ * Record a failed tool execution that did not hold a recovery probe.
+ *
+ * @param toolName tool that failed
+ */
+ public void recordFailure(String toolName) {
+ recordFailure(toolName, null);
+ }
+
/**
* Record a failed tool execution, tripping the circuit once the threshold is reached.
*
@@ -218,56 +339,68 @@ public void recordSuccess(String toolName) {
* towards a trip.
*
* @param toolName tool that failed
+ * @param probeToken token from {@link #tryAcquireProbe(String)}, or null when the call was not a
+ * recovery probe
*/
- public void recordFailure(String toolName) {
+ public void recordFailure(String toolName, String probeToken) {
if (!supervises(toolName)) {
return;
}
- ToolCircuitState current = state(toolName);
- if (current == ToolCircuitState.OPEN) {
- // The tool was withheld, so the model should not have been able to call it. This is
- // still reachable: the model may have chosen the call in the same turn the circuit
- // tripped. Ignore it rather than counting a failure the policy never authorised.
- logger.debug(
- "Ignoring failure of withheld tool, likely decided before the circuit opened:"
- + " tool={}",
- toolName);
- return;
- }
- if (current == ToolCircuitState.HALF_OPEN) {
- long generation = store.open(toolName, clock.millis());
- logger.warn(
- "Tool circuit re-opened after failed probe: tool={}, generation={},"
- + " cooldown={}",
- toolName,
- generation,
- cooldownFor(generation));
- return;
- }
- long failures = store.recordFailure(toolName);
- if (failures < config.getFailureThreshold()) {
- logger.debug(
- "Tool failure recorded: tool={}, consecutiveFailures={}/{}",
- toolName,
- failures,
- config.getFailureThreshold());
- return;
+ while (true) {
+ ToolCircuitSnapshot current = store.snapshot(toolName);
+ long now = clock.millis();
+ ToolCircuitState state = stateOf(current, now);
+ if (state == ToolCircuitState.OPEN) {
+ // The tool was withheld, so the model should not have been able to call it. This is
+ // still reachable: the model may have chosen the call in the same turn the circuit
+ // tripped. Ignore it rather than counting a failure the policy never authorised.
+ logger.debug(
+ "Ignoring failure of withheld tool, likely decided before the circuit"
+ + " opened: tool={}",
+ toolName);
+ return;
+ }
+ if (state == ToolCircuitState.HALF_OPEN) {
+ if (current.hasActiveProbe(now)
+ && !Objects.equals(probeToken, current.probeToken())) {
+ // Someone else owns the live probe; a superseded result must not re-open it.
+ return;
+ }
+ if (reopen(toolName, current, now)) {
+ return;
+ }
+ continue;
+ }
+ long failures = current.failureCount() + 1L;
+ if (failures < config.getFailureThreshold()) {
+ ToolCircuitSnapshot update =
+ new ToolCircuitSnapshot(failures, current.generation(), 0L, null, 0L);
+ if (store.compareAndSet(toolName, current, update)) {
+ logger.debug(
+ "Tool failure recorded: tool={}, consecutiveFailures={}/{}",
+ toolName,
+ failures,
+ config.getFailureThreshold());
+ return;
+ }
+ continue;
+ }
+ // Trip in the same compare-and-set that counted the final failure, so no caller can
+ // observe "threshold reached but not open" and act on it.
+ if (reopen(toolName, current, now)) {
+ logger.warn(
+ "Tool circuit opened: tool={}, consecutiveFailures={}. The tool will not be"
+ + " offered to the model until the cooldown elapses.",
+ toolName,
+ failures);
+ return;
+ }
}
- long generation = store.open(toolName, clock.millis());
- // Clear the streak on trip so the counter always means "failures seen while closed".
- store.resetFailures(toolName);
- logger.warn(
- "Tool circuit opened: tool={}, consecutiveFailures={}, generation={}, cooldown={}."
- + " The tool will not be offered to the model until the cooldown elapses.",
- toolName,
- failures,
- generation,
- cooldownFor(generation));
}
/**
- * Force a tool back to {@link ToolCircuitState#CLOSED}, discarding its failure streak and
- * accumulated backoff.
+ * Force a tool back to {@link ToolCircuitState#CLOSED}, discarding its failure streak,
+ * accumulated backoff and any probe claim.
*
* Intended for operators who know a dependency is healthy again and do not want to wait out
* the cooldown.
@@ -275,8 +408,7 @@ public void recordFailure(String toolName) {
* @param toolName tool to reset
*/
public void reset(String toolName) {
- store.close(toolName);
- store.resetFailures(toolName);
+ store.reset(toolName);
logger.info("Tool circuit manually reset: tool={}", toolName);
}
@@ -289,8 +421,25 @@ public ToolCircuitBreakerConfig getConfig() {
return config;
}
- private boolean hasCooldownElapsed(ToolCircuitSnapshot snapshot) {
+ private boolean reopen(String toolName, ToolCircuitSnapshot current, long now) {
+ long generation = current.generation() + 1L;
+ if (!store.compareAndSet(toolName, current, new ToolCircuitSnapshot(generation, now))) {
+ return false;
+ }
+ logger.warn(
+ "Tool circuit open: tool={}, generation={}, cooldown={}",
+ toolName,
+ generation,
+ cooldownFor(generation));
+ return true;
+ }
+
+ private ToolCircuitState stateOf(ToolCircuitSnapshot snapshot, long now) {
+ if (!snapshot.isOpen()) {
+ return ToolCircuitState.CLOSED;
+ }
Duration cooldown = cooldownFor(snapshot.generation());
- return clock.millis() >= snapshot.openedAtEpochMilli() + cooldown.toMillis();
+ boolean elapsed = now >= snapshot.openedAtEpochMilli() + cooldown.toMillis();
+ return elapsed ? ToolCircuitState.HALF_OPEN : ToolCircuitState.OPEN;
}
}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java
index f847a8e477..42e5eb095d 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java
@@ -54,6 +54,7 @@ public final class ToolCircuitBreakerConfig {
private final Duration initialCooldown;
private final double backoffMultiplier;
private final Duration maxCooldown;
+ private final Duration probeTimeout;
private ToolCircuitBreakerConfig(Builder builder) {
this.enabled = builder.enabled;
@@ -64,6 +65,7 @@ private ToolCircuitBreakerConfig(Builder builder) {
this.initialCooldown = builder.initialCooldown;
this.backoffMultiplier = builder.backoffMultiplier;
this.maxCooldown = builder.maxCooldown;
+ this.probeTimeout = builder.probeTimeout;
}
/**
@@ -148,6 +150,19 @@ public Duration getMaxCooldown() {
return maxCooldown;
}
+ /**
+ * Maximum time one half-open recovery probe owns the permit.
+ *
+ * Configure this no shorter than the supervised tool's execution timeout. Expiry prevents a
+ * reasoning turn that never calls the advertised tool, or a lost execution, from withholding
+ * recovery forever.
+ *
+ * @return positive probe timeout
+ */
+ public Duration getProbeTimeout() {
+ return probeTimeout;
+ }
+
/** Builder for {@link ToolCircuitBreakerConfig}. */
public static final class Builder {
@@ -159,6 +174,7 @@ public static final class Builder {
private Duration initialCooldown = Duration.ofSeconds(60);
private double backoffMultiplier = 2.0;
private Duration maxCooldown = Duration.ofSeconds(600);
+ private Duration probeTimeout = Duration.ofMinutes(5);
private Builder() {}
@@ -286,6 +302,17 @@ public Builder maxCooldown(Duration maxCooldown) {
return this;
}
+ /**
+ * Set how long a claimed half-open probe remains exclusive.
+ *
+ * @param probeTimeout positive duration, normally no shorter than tool execution timeout
+ * @return this builder
+ */
+ public Builder probeTimeout(Duration probeTimeout) {
+ this.probeTimeout = probeTimeout;
+ return this;
+ }
+
/**
* Validate and build the configuration.
*
@@ -308,6 +335,10 @@ public ToolCircuitBreakerConfig build() {
throw new IllegalArgumentException(
"maxCooldown must be positive, got " + maxCooldown);
}
+ if (probeTimeout == null || probeTimeout.isNegative() || probeTimeout.isZero()) {
+ throw new IllegalArgumentException(
+ "probeTimeout must be positive, got " + probeTimeout);
+ }
if (backoffMultiplier < 1.0 || !Double.isFinite(backoffMultiplier)) {
throw new IllegalArgumentException(
"backoffMultiplier must be a finite value of at least 1.0, got "
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddleware.java b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddleware.java
index cf711a7528..eafc28a8f3 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddleware.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddleware.java
@@ -18,6 +18,7 @@
import io.agentscope.core.agent.Agent;
import io.agentscope.core.agent.RuntimeContext;
import io.agentscope.core.event.AgentEvent;
+import io.agentscope.core.event.ToolCallStartEvent;
import io.agentscope.core.event.ToolResultEndEvent;
import io.agentscope.core.message.ToolResultState;
import io.agentscope.core.middleware.ActingInput;
@@ -25,8 +26,12 @@
import io.agentscope.core.middleware.ReasoningInput;
import io.agentscope.core.model.ToolSchema;
import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,10 +44,11 @@
* It occupies two interception points, which together close the state machine:
*
* Filtering happens per turn on a copy of the schema list. Nothing registered on the {@link
@@ -51,6 +57,20 @@
* authoritative. Recovery needs no repair step for the same reason: once the breaker stops
* withholding a tool, the unfiltered list is already correct.
*
+ * When a cooldown elapses the tool is advertised again, but only to the one turn that wins the
+ * probe claim; concurrent turns keep withholding it. The claim is carried on the per-call {@link
+ * RuntimeContext} and bound to the tool-call id the model chooses, so the matching result completes
+ * exactly that probe. Three paths hand the claim back early rather than waiting for its lease to
+ * expire: the model never calling the advertised tool, a call refused by permission rules, and a
+ * cancelled call — none of them tested the dependency. A suspended call keeps the claim, because its
+ * outcome is still pending.
+ *
+ * A null {@code RuntimeContext} (possible when the middleware is driven directly rather than by an
+ * agent) disables the binding: outcomes are then recorded without a token, and any probe claim is
+ * released when the reasoning stream terminates.
+ *
* Usage
*
* Marking the tool as selected also stops {@link #releaseUnusedProbes} from handing the claim
+ * back when the reasoning stream ends: the call is in flight and its outcome still pending.
+ */
+ private void bindProbe(
+ Map Only {@link ToolResultState#ERROR} counts as a failure. {@code DENIED} is a policy refusal
- * and {@code INTERRUPTED} a cancellation — neither is evidence about the dependency, and
- * counting them would let a user who declines a confirmation prompt trip the circuit. {@code
- * RUNNING} marks a suspended call whose outcome is not known yet.
+ * and {@code INTERRUPTED} a cancellation — neither is evidence about the dependency, and counting
+ * them would let a user who declines a confirmation prompt trip the circuit; both hand a probe
+ * claim back instead. {@code RUNNING} marks a suspended call whose outcome is not known yet, so
+ * its claim is left in place to be completed later or to expire.
*/
- private void recordOutcome(AgentEvent event) {
+ private void recordOutcome(RuntimeContext ctx, AgentEvent event) {
if (!(event instanceof ToolResultEndEvent result)) {
return;
}
@@ -165,10 +262,42 @@ private void recordOutcome(AgentEvent event) {
if (toolName == null || state == null) {
return;
}
- if (state == ToolResultState.ERROR) {
- breaker.recordFailure(toolName);
- } else if (state == ToolResultState.SUCCESS) {
- breaker.recordSuccess(toolName);
+ if (state == ToolResultState.RUNNING) {
+ return;
+ }
+ String token = consumeProbeToken(ctx, result.getToolCallId(), toolName);
+ switch (state) {
+ case ERROR -> breaker.recordFailure(toolName, token);
+ case SUCCESS -> breaker.recordSuccess(toolName, token);
+ case DENIED, INTERRUPTED -> breaker.releaseProbe(toolName, token);
+ default -> {
+ // RUNNING handled above; no other states exist.
+ }
+ }
+ }
+
+ private String consumeProbeToken(RuntimeContext ctx, String callId, String toolName) {
+ if (ctx == null || callId == null) {
+ return null;
+ }
+ Map Implementations are pure state holders: they must not apply the backoff policy, decide when a
- * circuit trips, or consult a clock. All policy lives in {@link ToolCircuitBreaker}, which keeps
- * this SPI stable when the policy evolves and makes the policy unit-testable without a store.
+ * circuit trips, or consult a clock. All policy lives in {@link ToolCircuitBreaker}. The breaker
+ * computes immutable snapshots and commits them with {@link #compareAndSet}; this makes a complete
+ * state transition atomic without moving policy into the persistence layer.
*
* {@link InMemoryToolCircuitBreakerStore} is the default and is sufficient for a single
* process. A distributed implementation (for example the Redis-backed store in
@@ -29,66 +30,38 @@
*
* Implementations must be safe for concurrent use from multiple threads. {@link
- * #recordFailure(String)} and {@link #open(String, long)} must be atomic, since concurrent tool
- * calls in one ReAct turn race on both.
+ * Implementations must be safe for concurrent use from multiple threads and processes. {@link
+ * #compareAndSet(String, ToolCircuitSnapshot, ToolCircuitSnapshot)} must compare and replace the
+ * complete snapshot atomically.
*/
public interface ToolCircuitBreakerStore {
/**
- * Atomically increment the consecutive-failure counter and return the new value.
- *
- * @param toolName tool being counted
- * @return the counter value after incrementing, starting at 1
- */
- long recordFailure(String toolName);
-
- /**
- * Clear the consecutive-failure counter.
- *
- * Called whenever a tool succeeds, which is what makes the threshold count
- * consecutive failures rather than lifetime failures.
- *
- * @param toolName tool to reset
- */
- void resetFailures(String toolName);
-
- /**
- * Read the consecutive-failure counter without modifying it.
+ * Read all state for one tool in a single round trip.
*
* @param toolName tool to read
- * @return current count, or {@code 0} when nothing is recorded
- */
- long failureCount(String toolName);
-
- /**
- * Atomically move the circuit to OPEN: increment the backoff generation and stamp the open
- * instant, returning the new generation.
- *
- * The caller supplies the timestamp so that it shares a clock with the cooldown comparison
- * in {@link ToolCircuitBreaker}; a store must never substitute its own clock.
- *
- * @param toolName tool to trip
- * @param openedAtEpochMilli instant the circuit opened, in epoch milliseconds
- * @return the backoff generation after incrementing, starting at 1
+ * @return current snapshot, never null; {@link ToolCircuitSnapshot#CLOSED} when no state exists
*/
- long open(String toolName, long openedAtEpochMilli);
+ ToolCircuitSnapshot snapshot(String toolName);
/**
- * Reset the circuit to CLOSED, discarding both the open timestamp and the backoff generation.
+ * Atomically replace the current snapshot if it still equals {@code expected}.
*
- * Dropping the generation means a tool that recovers starts its next incident from the
- * initial cooldown instead of inheriting an old, long backoff.
+ * {@link ToolCircuitSnapshot#CLOSED} is the logical value of a missing entry. Implementations
+ * should remove storage when {@code update} is CLOSED so healthy tools do not accumulate state.
*
- * @param toolName tool to close
+ * @param toolName tool to update
+ * @param expected snapshot the caller observed
+ * @param update complete replacement snapshot
+ * @return true when the replacement was committed; false when another caller changed the state
*/
- void close(String toolName);
+ boolean compareAndSet(
+ String toolName, ToolCircuitSnapshot expected, ToolCircuitSnapshot update);
/**
- * Read the trip state in a single round trip.
+ * Unconditionally discard all state for one tool.
*
- * @param toolName tool to read
- * @return current snapshot, never null; {@link ToolCircuitSnapshot#CLOSED} when no state exists
+ * @param toolName tool to reset
*/
- ToolCircuitSnapshot snapshot(String toolName);
+ void reset(String toolName);
}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java
index ddd27e8e9e..b2edd810d2 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java
@@ -16,22 +16,42 @@
package io.agentscope.core.tool.circuitbreaker;
/**
- * Immutable point-in-time view of a tool's trip state, read in a single store round trip.
+ * Immutable point-in-time view of all state for one tool, read in a single store round trip.
*
* The cooldown duration is deliberately not part of the snapshot: it is derived from
* {@code generation} by {@link ToolCircuitBreaker#cooldownFor(long)}, so changing the backoff
* policy takes effect immediately and never has to be migrated in the store.
*
+ * @param failureCount consecutive failures recorded while the circuit is closed
* @param generation number of times the circuit has tripped, driving exponential backoff;
* {@code 0} means it has never tripped
* @param openedAtEpochMilli wall-clock instant the circuit was last opened, or {@code 0} when
* the circuit is not open — a positive value is the sole marker of the OPEN state, so no
* separate boolean has to be kept consistent with it
+ * @param probeToken opaque owner token for the current half-open probe, or {@code null} when no
+ * probe is claimed
+ * @param probeLeaseUntilEpochMilli wall-clock instant at which the current probe claim expires, or
+ * {@code 0} when no probe is claimed
*/
-public record ToolCircuitSnapshot(long generation, long openedAtEpochMilli) {
+public record ToolCircuitSnapshot(
+ long failureCount,
+ long generation,
+ long openedAtEpochMilli,
+ String probeToken,
+ long probeLeaseUntilEpochMilli) {
- /** Snapshot of a tool that has never tripped. */
- public static final ToolCircuitSnapshot CLOSED = new ToolCircuitSnapshot(0L, 0L);
+ /** Snapshot of a healthy tool with no partial failure streak. */
+ public static final ToolCircuitSnapshot CLOSED = new ToolCircuitSnapshot(0L, 0L, 0L, null, 0L);
+
+ /**
+ * Create an open snapshot without a failure streak or claimed probe.
+ *
+ * @param generation trip generation
+ * @param openedAtEpochMilli instant the circuit opened
+ */
+ public ToolCircuitSnapshot(long generation, long openedAtEpochMilli) {
+ this(0L, generation, openedAtEpochMilli, null, 0L);
+ }
/**
* Whether the circuit is currently open, ignoring whether its cooldown has elapsed.
@@ -45,4 +65,16 @@ public record ToolCircuitSnapshot(long generation, long openedAtEpochMilli) {
public boolean isOpen() {
return openedAtEpochMilli > 0L;
}
+
+ /**
+ * Whether a recovery probe currently owns an unexpired lease.
+ *
+ * @param nowEpochMilli current wall-clock instant
+ * @return true when another caller must not acquire the probe
+ */
+ public boolean hasActiveProbe(long nowEpochMilli) {
+ return probeToken != null
+ && !probeToken.isEmpty()
+ && probeLeaseUntilEpochMilli > nowEpochMilli;
+ }
}
diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java
index 9a5a286660..c1f44ba3a3 100644
--- a/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java
@@ -22,10 +22,10 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;
-/** Contract of {@link InMemoryToolCircuitBreakerStore}, including its atomicity guarantees. */
+/** Compare-and-set contract of {@link InMemoryToolCircuitBreakerStore}. */
class InMemoryToolCircuitBreakerStoreTest {
private static final String TOOL = "query_weather";
@@ -34,89 +34,98 @@ class InMemoryToolCircuitBreakerStoreTest {
private final InMemoryToolCircuitBreakerStore store = new InMemoryToolCircuitBreakerStore();
@Test
- void unknownToolReadsAsClosedWithNoFailures() {
- assertEquals(0L, store.failureCount(TOOL));
+ void unknownToolReadsAsClosed() {
assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
assertFalse(store.snapshot(TOOL).isOpen());
}
@Test
- void failureCounterStartsAtOneAndIncrements() {
- assertEquals(1L, store.recordFailure(TOOL));
- assertEquals(2L, store.recordFailure(TOOL));
- assertEquals(2L, store.failureCount(TOOL));
- }
+ void compareAndSetCommitsWhenTheObservedValueStillHolds() {
+ ToolCircuitSnapshot update = new ToolCircuitSnapshot(2L, 0L, 0L, null, 0L);
- @Test
- void resetClearsOnlyTheNamedToolsCounter() {
- store.recordFailure(TOOL);
- store.recordFailure(OTHER_TOOL);
+ assertTrue(store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, update));
- store.resetFailures(TOOL);
+ assertEquals(update, store.snapshot(TOOL));
+ }
- assertEquals(0L, store.failureCount(TOOL));
- assertEquals(1L, store.failureCount(OTHER_TOOL));
+ @Test
+ void compareAndSetRejectsAStaleExpectation() {
+ ToolCircuitSnapshot first = new ToolCircuitSnapshot(1L, 0L, 0L, null, 0L);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, first);
+
+ // A caller that still believes the tool is untouched must not be able to commit.
+ boolean committed =
+ store.compareAndSet(
+ TOOL,
+ ToolCircuitSnapshot.CLOSED,
+ new ToolCircuitSnapshot(9L, 0L, 0L, null, 0L));
+
+ assertFalse(committed);
+ assertEquals(first, store.snapshot(TOOL));
}
@Test
- void openStampsTimestampAndAdvancesGeneration() {
- assertEquals(1L, store.open(TOOL, 1_000L));
+ void updatingToClosedRemovesTheEntry() {
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(3L, 1_000L));
- ToolCircuitSnapshot first = store.snapshot(TOOL);
- assertTrue(first.isOpen());
- assertEquals(1L, first.generation());
- assertEquals(1_000L, first.openedAtEpochMilli());
+ assertTrue(store.compareAndSet(TOOL, store.snapshot(TOOL), ToolCircuitSnapshot.CLOSED));
- assertEquals(2L, store.open(TOOL, 5_000L));
+ assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
+ // An absent entry must still be a valid expectation, proving it was removed rather than
+ // stored as an explicit zero value.
+ assertTrue(
+ store.compareAndSet(
+ TOOL,
+ ToolCircuitSnapshot.CLOSED,
+ new ToolCircuitSnapshot(1L, 0L, 0L, null, 0L)));
+ }
- ToolCircuitSnapshot second = store.snapshot(TOOL);
- assertEquals(2L, second.generation());
- assertEquals(5_000L, second.openedAtEpochMilli());
+ @Test
+ void probeClaimIsPartOfTheComparedValue() {
+ ToolCircuitSnapshot open = new ToolCircuitSnapshot(4L, 1_000L);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, open);
+ ToolCircuitSnapshot claimed = new ToolCircuitSnapshot(0L, 4L, 1_000L, "token-a", 9_000L);
+ assertTrue(store.compareAndSet(TOOL, open, claimed));
+
+ // Another caller holding the pre-claim value must lose, which is what makes the single
+ // recovery probe exclusive.
+ assertFalse(
+ store.compareAndSet(
+ TOOL, open, new ToolCircuitSnapshot(0L, 4L, 1_000L, "token-b", 9_000L)));
+ assertEquals("token-a", store.snapshot(TOOL).probeToken());
}
@Test
- void closeDiscardsGenerationSoBackoffRestarts() {
- store.open(TOOL, 1_000L);
- store.open(TOOL, 2_000L);
+ void resetDiscardsStateUnconditionally() {
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(2L, 1_000L));
- store.close(TOOL);
+ store.reset(TOOL);
- assertFalse(store.snapshot(TOOL).isOpen());
- assertEquals(0L, store.snapshot(TOOL).generation());
- assertEquals(1L, store.open(TOOL, 3_000L));
+ assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
}
@Test
void toolsDoNotShareState() {
- store.open(TOOL, 1_000L);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1_000L));
assertTrue(store.snapshot(TOOL).isOpen());
assertFalse(store.snapshot(OTHER_TOOL).isOpen());
}
@Test
- void concurrentFailureCountsAreNotLost() throws Exception {
- int threads = 8;
- int perThread = 500;
-
- runConcurrently(threads, perThread, () -> store.recordFailure(TOOL));
-
- assertEquals((long) threads * perThread, store.failureCount(TOOL));
- }
-
- @Test
- void concurrentOpensYieldContiguousGenerations() throws Exception {
- int threads = 8;
- int perThread = 200;
- AtomicLong maxGeneration = new AtomicLong();
+ void exactlyOneOfManyConcurrentCompareAndSetsWins() throws Exception {
+ int threads = 16;
+ ToolCircuitSnapshot expected = ToolCircuitSnapshot.CLOSED;
+ AtomicInteger winners = new AtomicInteger();
ExecutorService pool = Executors.newFixedThreadPool(threads);
try {
- for (int t = 0; t < threads; t++) {
+ for (int i = 0; i < threads; i++) {
+ long generation = i + 1L;
pool.submit(
() -> {
- for (int i = 0; i < perThread; i++) {
- long generation = store.open(TOOL, 1_000L + i);
- maxGeneration.accumulateAndGet(generation, Math::max);
+ if (store.compareAndSet(
+ TOOL, expected, new ToolCircuitSnapshot(generation, 1_000L))) {
+ winners.incrementAndGet();
}
});
}
@@ -126,27 +135,6 @@ void concurrentOpensYieldContiguousGenerations() throws Exception {
pool.shutdownNow();
}
- // Every open must observe a distinct, gap-free generation, so the highest value seen equals
- // the number of opens performed.
- assertEquals((long) threads * perThread, maxGeneration.get());
- assertEquals((long) threads * perThread, store.snapshot(TOOL).generation());
- }
-
- private void runConcurrently(int threads, int perThread, Runnable task) throws Exception {
- ExecutorService pool = Executors.newFixedThreadPool(threads);
- try {
- for (int t = 0; t < threads; t++) {
- pool.submit(
- () -> {
- for (int i = 0; i < perThread; i++) {
- task.run();
- }
- });
- }
- pool.shutdown();
- assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS));
- } finally {
- pool.shutdownNow();
- }
+ assertEquals(1, winners.get());
}
}
diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java
index ea89eb0415..e082386867 100644
--- a/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java
@@ -20,7 +20,9 @@
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import io.agentscope.core.agent.RuntimeContext;
import io.agentscope.core.event.AgentEvent;
+import io.agentscope.core.event.ToolCallStartEvent;
import io.agentscope.core.event.ToolResultEndEvent;
import io.agentscope.core.message.Msg;
import io.agentscope.core.message.MsgRole;
@@ -34,6 +36,12 @@
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
@@ -134,6 +142,63 @@ void toolIsOfferedAgainOnceTheCooldownElapses() {
assertEquals(List.of(WEATHER, DATABASE), offeredToolNames(middleware, WEATHER, DATABASE));
}
+ @Test
+ void onlyOneConcurrentReasoningTurnReceivesTheHalfOpenProbe() throws Exception {
+ ToolCircuitBreakerMiddleware middleware = middleware(1);
+ failTool(middleware, WEATHER);
+ clock.advance(Duration.ofSeconds(60));
+
+ int turns = 8;
+ ExecutorService pool = Executors.newFixedThreadPool(turns);
+ CountDownLatch ready = new CountDownLatch(turns);
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch observed = new CountDownLatch(turns);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger offers = new AtomicInteger();
+ List With the in-process store each replica has to rediscover an outage for itself, so an N-replica
* deployment sends roughly N times the failing traffic and burns N times the tokens before the tool
* is withheld everywhere. Sharing the state through Redis means the first replica to trip a circuit
- * withholds the tool for all of them, and the state survives a restart or a rescheduled pod.
+ * withholds the tool for all of them, exactly one replica probes for recovery, and the state survives
+ * a restart or a rescheduled pod.
*
- * Two keys per tool, both addressed individually so the store works unchanged on Redis Cluster —
- * no multi-key script needs its arguments to share a hash slot:
+ * One key per tool, {@code {prefix}{tool}}, holding the whole snapshot as a delimited value:
*
- * Encoding generation and timestamp in one value keeps {@link #snapshot(String)} — the hot read,
- * executed for every supervised tool on every reasoning turn — down to a single {@code GET}.
+ * Keeping the entire state in one key is what lets a complete transition be one compare-and-set,
+ * and it means every script touches a single key — so the store needs no hash tags and works
+ * unchanged on Redis Cluster. A missing key is the encoding of {@link ToolCircuitSnapshot#CLOSED},
+ * so a recovered tool leaves nothing behind.
*
* {@link #recordFailure(String)} and {@link #open(String, long)} are Lua scripts, so their
- * read-modify-write steps cannot interleave. Doing {@code INCR} and {@code EXPIRE} as two round
- * trips would leave a counter without a TTL whenever the second call is lost, and computing the next
- * generation client-side would let two replicas tripping at once write the same generation.
+ * {@link #compareAndSet} compares the stored value against the caller's expected encoding and
+ * replaces it in one Lua script. Doing the comparison client-side would reintroduce exactly the races
+ * the breaker's compare-and-set protocol exists to remove: two replicas could each read the same
+ * state and both commit a transition based on it.
*
* Both keys carry a TTL so tools that misbehave once do not accumulate state forever. Keep the
- * TTL comfortably longer than the breaker's maximum cooldown: if an open circuit's key expires
- * mid-cooldown the tool is offered again early, which fails open — safe, but not what was
- * configured. The default of 24h clears the default 600s ceiling by a wide margin.
+ * Non-closed states carry a TTL so tools that misbehave once do not accumulate state forever. Keep
+ * the TTL comfortably longer than the breaker's maximum cooldown: if an open circuit's key expires
+ * mid-cooldown the tool is offered again early, which fails open — safe, but not what was configured.
+ * The default of 24h clears the default 600s ceiling by a wide margin.
*/
public class RedisToolCircuitBreakerStore implements ToolCircuitBreakerStore {
@@ -69,31 +69,28 @@ public class RedisToolCircuitBreakerStore implements ToolCircuitBreakerStore {
private static final String DEFAULT_KEY_PREFIX = "agentscope:tool-cb:";
private static final Duration DEFAULT_TTL = Duration.ofHours(24);
- private static final String FAILURE_SUFFIX = ":fail";
- private static final String CIRCUIT_SUFFIX = ":circuit";
+ /** Encoding of {@link ToolCircuitSnapshot#CLOSED}: an absent key. */
+ private static final String ABSENT = "";
- /**
- * Increment the failure counter and refresh its TTL in one step.
- *
- * KEYS[1] = failure key; ARGV[1] = TTL seconds. Returns the new count.
- */
- private static final String INCREMENT_FAILURE_SCRIPT =
- "local count = redis.call('INCR', KEYS[1]) "
- + "redis.call('EXPIRE', KEYS[1], ARGV[1]) "
- + "return count";
+ private static final int FIELD_COUNT = 5;
/**
- * Advance the generation and stamp the open instant in one step.
+ * Replace the stored value only if it still equals what the caller observed.
*
- * KEYS[1] = circuit key; ARGV[1] = open instant in epoch millis; ARGV[2] = TTL seconds.
- * Returns the new generation.
+ * KEYS[1] = circuit key; ARGV[1] = expected encoding ({@code ""} for absent); ARGV[2] = new
+ * encoding ({@code ""} to delete); ARGV[3] = TTL seconds. Returns 1 when committed, 0 when the
+ * value had changed.
*/
- private static final String OPEN_NEXT_GENERATION_SCRIPT =
- "local current = redis.call('GET', KEYS[1]) local generation = 0 if current then "
- + " local sep = string.find(current, ':', 1, true) if sep then generation ="
- + " tonumber(string.sub(current, 1, sep - 1)) or 0 end end generation = generation"
- + " + 1 redis.call('SET', KEYS[1], generation .. ':' .. ARGV[1], 'EX', ARGV[2])"
- + " return generation";
+ private static final String COMPARE_AND_SET_SCRIPT =
+ "local current = redis.call('GET', KEYS[1]) "
+ + "if current == false then current = '' end "
+ + "if current ~= ARGV[1] then return 0 end "
+ + "if ARGV[2] == '' then "
+ + " redis.call('DEL', KEYS[1]) "
+ + "else "
+ + " redis.call('SET', KEYS[1], ARGV[2], 'EX', ARGV[3]) "
+ + "end "
+ + "return 1";
private final RedisClientAdapter client;
private final String keyPrefix;
@@ -113,7 +110,7 @@ public RedisToolCircuitBreakerStore(RedisClientAdapter client) {
*
* @param client Redis client adapter
* @param keyPrefix prefix for every key, letting environments share one Redis instance
- * @param stateTtl how long unused state is retained; must be positive and should exceed the
+ * @param stateTtl how long non-closed state is retained; must be positive and should exceed the
* breaker's maximum cooldown
*/
public RedisToolCircuitBreakerStore(
@@ -130,46 +127,55 @@ public RedisToolCircuitBreakerStore(
}
@Override
- public long recordFailure(String toolName) {
- return client.evalScript(
- INCREMENT_FAILURE_SCRIPT,
- List.of(failureKey(toolName)),
- List.of(Long.toString(ttlSeconds)));
- }
-
- @Override
- public void resetFailures(String toolName) {
- client.deleteKeys(failureKey(toolName));
- }
-
- @Override
- public long failureCount(String toolName) {
- return parseLong(client.get(failureKey(toolName)));
+ public ToolCircuitSnapshot snapshot(String toolName) {
+ return decode(toolName, client.get(circuitKey(toolName)));
}
@Override
- public long open(String toolName, long openedAtEpochMilli) {
+ public boolean compareAndSet(
+ String toolName, ToolCircuitSnapshot expected, ToolCircuitSnapshot update) {
return client.evalScript(
- OPEN_NEXT_GENERATION_SCRIPT,
- List.of(circuitKey(toolName)),
- List.of(Long.toString(openedAtEpochMilli), Long.toString(ttlSeconds)));
+ COMPARE_AND_SET_SCRIPT,
+ List.of(circuitKey(toolName)),
+ List.of(encode(expected), encode(update), Long.toString(ttlSeconds)))
+ == 1L;
}
@Override
- public void close(String toolName) {
+ public void reset(String toolName) {
client.deleteKeys(circuitKey(toolName));
}
- @Override
- public ToolCircuitSnapshot snapshot(String toolName) {
- String value = client.get(circuitKey(toolName));
+ /**
+ * Encode a snapshot, mapping CLOSED to the absent-key marker so "missing" and "closed" compare
+ * equal.
+ */
+ private static String encode(ToolCircuitSnapshot snapshot) {
+ if (snapshot == null || ToolCircuitSnapshot.CLOSED.equals(snapshot)) {
+ return ABSENT;
+ }
+ String token = snapshot.probeToken() == null ? "" : snapshot.probeToken();
+ return snapshot.failureCount()
+ + ":"
+ + snapshot.generation()
+ + ":"
+ + snapshot.openedAtEpochMilli()
+ + ":"
+ + token
+ + ":"
+ + snapshot.probeLeaseUntilEpochMilli();
+ }
+
+ /**
+ * Decode a stored value. Anything unreadable is treated as closed rather than withholding a tool
+ * forever on the strength of state nobody can interpret.
+ */
+ private ToolCircuitSnapshot decode(String toolName, String value) {
if (value == null || value.isEmpty()) {
return ToolCircuitSnapshot.CLOSED;
}
- int separator = value.indexOf(':');
- if (separator <= 0 || separator == value.length() - 1) {
- // Unreadable value: treat as closed rather than withholding a tool forever on the
- // strength of state nobody can interpret.
+ String[] parts = value.split(":", -1);
+ if (parts.length != FIELD_COUNT) {
logger.warn(
"Ignoring malformed circuit state for tool={}, value={}. Treating the circuit"
+ " as closed.",
@@ -177,35 +183,32 @@ public ToolCircuitSnapshot snapshot(String toolName) {
value);
return ToolCircuitSnapshot.CLOSED;
}
- long generation = parseLong(value.substring(0, separator));
- long openedAt = parseLong(value.substring(separator + 1));
- if (generation <= 0L || openedAt <= 0L) {
+ try {
+ long failureCount = Long.parseLong(parts[0]);
+ long generation = Long.parseLong(parts[1]);
+ long openedAt = Long.parseLong(parts[2]);
+ String token = parts[3].isEmpty() ? null : parts[3];
+ long probeLease = Long.parseLong(parts[4]);
+ if (failureCount < 0L || generation < 0L || openedAt < 0L || probeLease < 0L) {
+ logger.warn(
+ "Ignoring out-of-range circuit state for tool={}, value={}. Treating the"
+ + " circuit as closed.",
+ toolName,
+ value);
+ return ToolCircuitSnapshot.CLOSED;
+ }
+ return new ToolCircuitSnapshot(failureCount, generation, openedAt, token, probeLease);
+ } catch (NumberFormatException e) {
logger.warn(
- "Ignoring out-of-range circuit state for tool={}, value={}. Treating the"
- + " circuit as closed.",
+ "Ignoring unparsable circuit state for tool={}, value={}. Treating the circuit"
+ + " as closed.",
toolName,
value);
return ToolCircuitSnapshot.CLOSED;
}
- return new ToolCircuitSnapshot(generation, openedAt);
- }
-
- private String failureKey(String toolName) {
- return keyPrefix + toolName + FAILURE_SUFFIX;
}
private String circuitKey(String toolName) {
- return keyPrefix + toolName + CIRCUIT_SUFFIX;
- }
-
- private static long parseLong(String value) {
- if (value == null || value.isEmpty()) {
- return 0L;
- }
- try {
- return Long.parseLong(value.trim());
- } catch (NumberFormatException e) {
- return 0L;
- }
+ return keyPrefix + toolName;
}
}
diff --git a/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java
index 5b13b10e0b..3287dd15bd 100644
--- a/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java
+++ b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java
@@ -17,6 +17,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -31,172 +32,215 @@
import org.junit.jupiter.api.Test;
/**
- * Client-side behaviour of {@link RedisToolCircuitBreakerStore}: key naming, argument passing and
- * the decoding of persisted circuit values.
+ * Client-side behaviour of {@link RedisToolCircuitBreakerStore}: key layout, the arguments handed to
+ * the compare-and-set script, and the encoding of persisted state.
*
- * Scope note: the two Lua scripts are executed by Redis, so a fake client cannot run them. These
- * tests cover the Java side — which keys are addressed, which arguments the scripts receive, and how
- * stored values are decoded, including values a healthy writer would never produce. The scripts'
- * server-side effects need a live Redis to verify.
+ * Scope note: the Lua script is executed by Redis, so a fake client cannot run it. These tests
+ * cover the Java side — which key is addressed, what the script receives, and how stored values are
+ * encoded and decoded, including values a healthy writer would never produce. The script's
+ * server-side effect needs a live Redis to verify.
*/
class RedisToolCircuitBreakerStoreTest {
private static final String TOOL = "query_weather";
+ private static final String KEY = "cb:query_weather";
private final RecordingRedisClient client = new RecordingRedisClient();
- // ==================== Key naming ====================
+ // ==================== Key layout ====================
@Test
- void keysCarryThePrefixAndDistinctSuffixes() {
+ void allStateLivesUnderOneKeySoClusterNeedsNoHashTag() {
RedisToolCircuitBreakerStore store = store();
- store.recordFailure(TOOL);
- store.open(TOOL, 1_000L);
+ store.snapshot(TOOL);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1_000L));
+ store.reset(TOOL);
- assertEquals(
- List.of("cb:query_weather:fail", "cb:query_weather:circuit"), client.scriptKeys);
+ assertEquals(List.of(KEY), client.reads);
+ assertEquals(List.of(List.of(KEY)), client.scriptKeys);
+ assertEquals(List.of(KEY), client.deleted);
}
@Test
- void resetFailuresDeletesOnlyTheCounter() {
- RedisToolCircuitBreakerStore store = store();
+ void defaultConstructorUsesTheDocumentedPrefix() {
+ RedisToolCircuitBreakerStore store = new RedisToolCircuitBreakerStore(client);
- store.resetFailures(TOOL);
+ store.reset(TOOL);
- assertEquals(List.of("cb:query_weather:fail"), client.deleted);
+ assertEquals(List.of("agentscope:tool-cb:query_weather"), client.deleted);
}
+ // ==================== Compare-and-set arguments ====================
+
@Test
- void closeDeletesOnlyTheCircuitKey() {
+ void closedIsEncodedAsTheAbsentKeyOnBothSidesOfTheSwap() {
RedisToolCircuitBreakerStore store = store();
- store.close(TOOL);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(2L, 5_000L));
- assertEquals(List.of("cb:query_weather:circuit"), client.deleted);
+ // Expected "" makes "missing" and "closed" compare equal; update carries the new state.
+ assertEquals(List.of("", "0:2:5000::0", "86400"), client.scriptArgs.get(0));
}
- // ==================== Script arguments ====================
-
@Test
- void failureScriptReceivesTheTtlInSeconds() {
- RedisToolCircuitBreakerStore store =
- new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofMinutes(30));
+ void updatingToClosedRequestsDeletionViaAnEmptyUpdate() {
+ RedisToolCircuitBreakerStore store = store();
+ ToolCircuitSnapshot open = new ToolCircuitSnapshot(3L, 7_000L);
- store.recordFailure(TOOL);
+ store.compareAndSet(TOOL, open, ToolCircuitSnapshot.CLOSED);
- assertEquals(List.of("1800"), client.scriptArgs.get(0));
+ assertEquals(List.of("0:3:7000::0", "", "86400"), client.scriptArgs.get(0));
}
@Test
- void openScriptReceivesTheTimestampThenTheTtl() {
- RedisToolCircuitBreakerStore store =
- new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofHours(24));
+ void probeClaimIsCarriedInTheEncodedValue() {
+ RedisToolCircuitBreakerStore store = store();
+ ToolCircuitSnapshot claimed = new ToolCircuitSnapshot(0L, 4L, 1_000L, "tok", 9_000L);
- store.open(TOOL, 1_767_225_600_000L);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, claimed);
- assertEquals(List.of("1767225600000", "86400"), client.scriptArgs.get(0));
+ assertEquals(List.of("", "0:4:1000:tok:9000", "86400"), client.scriptArgs.get(0));
}
@Test
- void subSecondTtlIsFlooredToOneSecondSoKeysNeverPersistForever() {
- RedisToolCircuitBreakerStore store =
- new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofMillis(200));
+ void ttlIsPassedInSecondsAndFlooredToOne() {
+ new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofMinutes(30))
+ .compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1L));
+ assertEquals("1800", client.scriptArgs.get(0).get(2));
+
+ client.scriptArgs.clear();
+ new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofMillis(200))
+ .compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1L));
+ assertEquals("1", client.scriptArgs.get(0).get(2));
+ }
- store.recordFailure(TOOL);
+ @Test
+ void scriptResultDecidesWhetherTheSwapCommitted() {
+ RedisToolCircuitBreakerStore store = store();
- assertEquals(List.of("1"), client.scriptArgs.get(0));
+ client.scriptResult = 1L;
+ assertTrue(
+ store.compareAndSet(
+ TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1_000L)));
+
+ client.scriptResult = 0L;
+ assertFalse(
+ store.compareAndSet(
+ TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1_000L)));
}
- // ==================== Decoding persisted state ====================
+ // ==================== Decoding ====================
@Test
- void snapshotDecodesGenerationAndTimestamp() {
+ void snapshotDecodesEveryField() {
RedisToolCircuitBreakerStore store = store();
- client.values.put("cb:query_weather:circuit", "3:1767225600000");
+ client.values.put(KEY, "2:3:1767225600000:tok:1767225660000");
ToolCircuitSnapshot snapshot = store.snapshot(TOOL);
- assertTrue(snapshot.isOpen());
+ assertEquals(2L, snapshot.failureCount());
assertEquals(3L, snapshot.generation());
assertEquals(1_767_225_600_000L, snapshot.openedAtEpochMilli());
+ assertEquals("tok", snapshot.probeToken());
+ assertEquals(1_767_225_660_000L, snapshot.probeLeaseUntilEpochMilli());
+ assertTrue(snapshot.isOpen());
}
@Test
- void missingKeyDecodesAsClosed() {
+ void anEmptyProbeFieldDecodesToNoClaim() {
RedisToolCircuitBreakerStore store = store();
+ client.values.put(KEY, "0:1:1000::0");
- assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
- assertFalse(store.snapshot(TOOL).isOpen());
+ ToolCircuitSnapshot snapshot = store.snapshot(TOOL);
+
+ assertNull(snapshot.probeToken());
+ assertFalse(snapshot.hasActiveProbe(0L));
}
@Test
- void unreadableValuesFailOpenRatherThanWithholdingForever() {
- RedisToolCircuitBreakerStore store = store();
- String key = "cb:query_weather:circuit";
-
- for (String malformed :
- List.of("", "garbage", ":", "3:", ":1767225600000", "0:1767225600000", "3:0")) {
- client.values.put(key, malformed);
- assertEquals(
- ToolCircuitSnapshot.CLOSED,
- store.snapshot(TOOL),
- "expected a closed circuit for stored value: '" + malformed + "'");
- }
+ void missingKeyDecodesAsClosed() {
+ assertEquals(ToolCircuitSnapshot.CLOSED, store().snapshot(TOOL));
}
@Test
- void nonNumericFailureCountReadsAsZero() {
+ void encodingRoundTripsThroughDecoding() {
RedisToolCircuitBreakerStore store = store();
- client.values.put("cb:query_weather:fail", "not-a-number");
+ ToolCircuitSnapshot original = new ToolCircuitSnapshot(2L, 3L, 1_000L, "tok", 9_000L);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, original);
+
+ // Feed the encoding the store just produced back through the read path.
+ client.values.put(KEY, client.scriptArgs.get(0).get(1));
- assertEquals(0L, store.failureCount(TOOL));
+ assertEquals(original, store.snapshot(TOOL));
}
@Test
- void failureCountIsReadFromTheCounterKey() {
+ void unreadableValuesFailOpenRatherThanWithholdingForever() {
RedisToolCircuitBreakerStore store = store();
- client.values.put("cb:query_weather:fail", "7");
- assertEquals(7L, store.failureCount(TOOL));
+ for (String malformed :
+ List.of(
+ "",
+ "garbage",
+ "0:1:1000",
+ "0:1:1000::0:extra",
+ "x:1:1000::0",
+ "0:-1:1000::0",
+ "0:1:-5::0")) {
+ client.values.put(KEY, malformed);
+ assertEquals(
+ ToolCircuitSnapshot.CLOSED,
+ store.snapshot(TOOL),
+ "expected a closed circuit for stored value: '" + malformed + "'");
+ }
}
// ==================== Construction ====================
@Test
- void constructorRejectsBlankPrefixAndNonPositiveTtl() {
+ void constructorRejectsInvalidArguments() {
+ assertThrows(
+ NullPointerException.class,
+ () -> new RedisToolCircuitBreakerStore(null, "cb:", Duration.ofHours(1)));
assertThrows(
IllegalArgumentException.class,
() -> new RedisToolCircuitBreakerStore(client, " ", Duration.ofHours(1)));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new RedisToolCircuitBreakerStore(client, null, Duration.ofHours(1)));
assertThrows(
IllegalArgumentException.class,
() -> new RedisToolCircuitBreakerStore(client, "cb:", Duration.ZERO));
assertThrows(
- NullPointerException.class,
- () -> new RedisToolCircuitBreakerStore(null, "cb:", Duration.ofHours(1)));
+ IllegalArgumentException.class,
+ () -> new RedisToolCircuitBreakerStore(client, "cb:", null));
}
private RedisToolCircuitBreakerStore store() {
return new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofHours(24));
}
- /** Fake client recording the keys and arguments each call addresses. */
+ /** Fake client recording the key and arguments each call addresses. */
private static final class RecordingRedisClient implements RedisClientAdapter {
private final MapWhy withhold the tool instead of rejecting the call
+ *
+ * State machine
+ *
+ *
+ * CLOSED --failureThreshold consecutive failures--> OPEN
+ * OPEN --cooldown elapsed--> HALF_OPEN (tool advertised again, as a probe)
+ * HALF_OPEN --probe succeeds--> CLOSED (failure counter cleared)
+ * HALF_OPEN --probe fails--> OPEN (next generation, longer cooldown)
+ *
+ *
+ * Threading
+ *
+ * Tools are supervised by opt-in, not by default
+ *
+ * Cooldown grows with each trip
+ *
+ *
+ *
+ *
+ * {@code
+ * ToolCircuitBreakerConfig config = ToolCircuitBreakerConfig.builder()
+ * .monitorTools("query_weather", "query_destination_news")
+ * .failureThreshold(3)
+ * .initialCooldown(Duration.ofSeconds(60))
+ * .maxCooldown(Duration.ofSeconds(600))
+ * .build();
+ *
+ * ReActAgent agent = ReActAgent.builder()
+ * .model(model)
+ * .toolkit(toolkit)
+ * .middleware(new ToolCircuitBreakerMiddleware(new ToolCircuitBreaker(config)))
+ * .build();
+ * }
+ *
+ * Threading
+ *
+ *
+ * CLOSED --failure x threshold--> OPEN --cooldown elapsed--> HALF_OPEN --success--> CLOSED
+ * ^ |
+ * +---------- failure -----------+
+ *
+ */
+public enum ToolCircuitState {
+
+ /**
+ * Normal operation: the tool is exposed to the model and consecutive failures are counted.
+ *
+ * Keys
+ *
+ *
+ *
+ *
+ * Atomicity
+ *
+ * Expiry
+ *
+ * State machine
*
*
* CLOSED --failureThreshold consecutive failures--> OPEN
- * OPEN --cooldown elapsed--> HALF_OPEN (tool advertised again, as a probe)
- * HALF_OPEN --probe succeeds--> CLOSED (failure counter cleared)
+ * OPEN --cooldown elapsed--> HALF_OPEN (one turn may probe; see below)
+ * HALF_OPEN --probe succeeds--> CLOSED (failure streak cleared)
* HALF_OPEN --probe fails--> OPEN (next generation, longer cooldown)
*
*
@@ -54,13 +55,28 @@
* snapshot and the current time, so a cooldown that elapsed while the agent was idle is recognised
* on the next read. There is no timer to leak and no scheduler to configure.
*
- * Threading
+ * Concurrency
*
- *
+ *
+ *
+ *
- *
*
* Recovery probes
+ *
+ * {@code
@@ -77,6 +97,7 @@ public class ToolCircuitBreakerMiddleware implements MiddlewareBase {
LoggerFactory.getLogger(ToolCircuitBreakerMiddleware.class);
private final ToolCircuitBreaker breaker;
+ private final String probeAttributeKey;
/**
* Wrap a breaker as middleware.
@@ -85,6 +106,12 @@ public class ToolCircuitBreakerMiddleware implements MiddlewareBase {
*/
public ToolCircuitBreakerMiddleware(ToolCircuitBreaker breaker) {
this.breaker = Objects.requireNonNull(breaker, "breaker must not be null");
+ // Scope the context attribute to this instance so two breakers on one agent cannot claim
+ // each other's probe bindings.
+ this.probeAttributeKey =
+ ToolCircuitBreakerMiddleware.class.getName()
+ + ".probes#"
+ + Integer.toHexString(System.identityHashCode(breaker));
}
/**
@@ -117,26 +144,57 @@ public FluxThreading
*
- * Keys
+ * Key layout
*
- *
- *
+ *
+ * "<failureCount>:<generation>:<openedAt>:<probeToken>:<probeLeaseUntil>"
+ *
*
- * Atomicity
*
- * Expiry
*
- *