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.
+ *
+ * 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.
+ *
+ * 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 {
+
+ 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 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) {
+ 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}. 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
+ */
+ public ToolCircuitState state(String toolName) {
+ if (!supervises(toolName)) {
+ return ToolCircuitState.CLOSED;
+ }
+ return stateOf(store.snapshot(toolName), clock.millis());
+ }
+
+ /**
+ * Whether the tool is inside a cooldown and must be kept out of the schema list.
+ *
+ * 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 is in an unexpired cooldown
+ */
+ 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));
+ }
+
+ /**
+ * Try to claim the single recovery probe for a half-open circuit.
+ *
+ * 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 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 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;
+ }
+ 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;
+ }
+ 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.
+ *
+ * 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
+ * @param probeToken token from {@link #tryAcquireProbe(String)}, or null when the call was not a
+ * recovery probe
+ */
+ public void recordFailure(String toolName, String probeToken) {
+ if (!supervises(toolName)) {
+ 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;
+ }
+ }
+ }
+
+ /**
+ * 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.
+ *
+ * @param toolName tool to reset
+ */
+ public void reset(String toolName) {
+ store.reset(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 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());
+ 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
new file mode 100644
index 0000000000..42e5eb095d
--- /dev/null
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerConfig.java
@@ -0,0 +1,358 @@
+/*
+ * 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 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 {
+
+ private boolean enabled = true;
+ 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.
+ *
+ * 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
+ *
+ * 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;
+ private final String probeAttributeKey;
+
+ /**
+ * 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");
+ // 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));
+ }
+
+ /**
+ * 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 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; 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(RuntimeContext ctx, 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.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}. 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
+ * {@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 and processes. {@link
+ * #compareAndSet(String, ToolCircuitSnapshot, ToolCircuitSnapshot)} must compare and replace the
+ * complete snapshot atomically.
+ */
+public interface ToolCircuitBreakerStore {
+
+ /**
+ * Read all state for one tool 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);
+
+ /**
+ * Atomically replace the current snapshot if it still equals {@code expected}.
+ *
+ * {@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 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
+ */
+ boolean compareAndSet(
+ String toolName, ToolCircuitSnapshot expected, ToolCircuitSnapshot update);
+
+ /**
+ * Unconditionally discard all state for one tool.
+ *
+ * @param toolName tool to reset
+ */
+ 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
new file mode 100644
index 0000000000..b2edd810d2
--- /dev/null
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitSnapshot.java
@@ -0,0 +1,80 @@
+/*
+ * 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 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 failureCount,
+ long generation,
+ long openedAtEpochMilli,
+ String probeToken,
+ long probeLeaseUntilEpochMilli) {
+
+ /** 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.
+ *
+ * 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;
+ }
+
+ /**
+ * 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/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..c1f44ba3a3
--- /dev/null
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/InMemoryToolCircuitBreakerStoreTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+/** Compare-and-set contract of {@link InMemoryToolCircuitBreakerStore}. */
+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 unknownToolReadsAsClosed() {
+ assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
+ assertFalse(store.snapshot(TOOL).isOpen());
+ }
+
+ @Test
+ void compareAndSetCommitsWhenTheObservedValueStillHolds() {
+ ToolCircuitSnapshot update = new ToolCircuitSnapshot(2L, 0L, 0L, null, 0L);
+
+ assertTrue(store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, update));
+
+ assertEquals(update, store.snapshot(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 updatingToClosedRemovesTheEntry() {
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(3L, 1_000L));
+
+ assertTrue(store.compareAndSet(TOOL, store.snapshot(TOOL), ToolCircuitSnapshot.CLOSED));
+
+ 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)));
+ }
+
+ @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 resetDiscardsStateUnconditionally() {
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(2L, 1_000L));
+
+ store.reset(TOOL);
+
+ assertEquals(ToolCircuitSnapshot.CLOSED, store.snapshot(TOOL));
+ }
+
+ @Test
+ void toolsDoNotShareState() {
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1_000L));
+
+ assertTrue(store.snapshot(TOOL).isOpen());
+ assertFalse(store.snapshot(OTHER_TOOL).isOpen());
+ }
+
+ @Test
+ void exactlyOneOfManyConcurrentCompareAndSetsWins() throws Exception {
+ int threads = 16;
+ ToolCircuitSnapshot expected = ToolCircuitSnapshot.CLOSED;
+ AtomicInteger winners = new AtomicInteger();
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ for (int i = 0; i < threads; i++) {
+ long generation = i + 1L;
+ pool.submit(
+ () -> {
+ if (store.compareAndSet(
+ TOOL, expected, new ToolCircuitSnapshot(generation, 1_000L))) {
+ winners.incrementAndGet();
+ }
+ });
+ }
+ 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/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..e082386867
--- /dev/null
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/circuitbreaker/ToolCircuitBreakerMiddlewareTest.java
@@ -0,0 +1,459 @@
+/*
+ * 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.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;
+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.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;
+
+/**
+ * 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, exactly one replica probes for recovery, and the state survives
+ * a restart or a rescheduled pod.
+ *
+ * One key per tool, {@code {prefix}{tool}}, holding the whole snapshot as a delimited value:
+ *
+ * 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 #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.
+ *
+ * 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 {
+
+ 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);
+
+ /** Encoding of {@link ToolCircuitSnapshot#CLOSED}: an absent key. */
+ private static final String ABSENT = "";
+
+ private static final int FIELD_COUNT = 5;
+
+ /**
+ * Replace the stored value only if it still equals what the caller observed.
+ *
+ * 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 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;
+ 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 non-closed 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 ToolCircuitSnapshot snapshot(String toolName) {
+ return decode(toolName, client.get(circuitKey(toolName)));
+ }
+
+ @Override
+ public boolean compareAndSet(
+ String toolName, ToolCircuitSnapshot expected, ToolCircuitSnapshot update) {
+ return client.evalScript(
+ COMPARE_AND_SET_SCRIPT,
+ List.of(circuitKey(toolName)),
+ List.of(encode(expected), encode(update), Long.toString(ttlSeconds)))
+ == 1L;
+ }
+
+ @Override
+ public void reset(String toolName) {
+ client.deleteKeys(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;
+ }
+ String[] parts = value.split(":", -1);
+ if (parts.length != FIELD_COUNT) {
+ logger.warn(
+ "Ignoring malformed circuit state for tool={}, value={}. Treating the circuit"
+ + " as closed.",
+ toolName,
+ value);
+ return ToolCircuitSnapshot.CLOSED;
+ }
+ 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 unparsable circuit state for tool={}, value={}. Treating the circuit"
+ + " as closed.",
+ toolName,
+ value);
+ return ToolCircuitSnapshot.CLOSED;
+ }
+ }
+
+ private String circuitKey(String toolName) {
+ 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
new file mode 100644
index 0000000000..3287dd15bd
--- /dev/null
+++ b/agentscope-extensions/agentscope-extensions-redis/src/test/java/io/agentscope/extensions/redis/circuitbreaker/RedisToolCircuitBreakerStoreTest.java
@@ -0,0 +1,305 @@
+/*
+ * 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.assertNull;
+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 layout, the arguments handed to
+ * the compare-and-set script, and the encoding of persisted state.
+ *
+ * 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 layout ====================
+
+ @Test
+ void allStateLivesUnderOneKeySoClusterNeedsNoHashTag() {
+ RedisToolCircuitBreakerStore store = store();
+
+ store.snapshot(TOOL);
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(1L, 1_000L));
+ store.reset(TOOL);
+
+ assertEquals(List.of(KEY), client.reads);
+ assertEquals(List.of(List.of(KEY)), client.scriptKeys);
+ assertEquals(List.of(KEY), client.deleted);
+ }
+
+ @Test
+ void defaultConstructorUsesTheDocumentedPrefix() {
+ RedisToolCircuitBreakerStore store = new RedisToolCircuitBreakerStore(client);
+
+ store.reset(TOOL);
+
+ assertEquals(List.of("agentscope:tool-cb:query_weather"), client.deleted);
+ }
+
+ // ==================== Compare-and-set arguments ====================
+
+ @Test
+ void closedIsEncodedAsTheAbsentKeyOnBothSidesOfTheSwap() {
+ RedisToolCircuitBreakerStore store = store();
+
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, new ToolCircuitSnapshot(2L, 5_000L));
+
+ // Expected "" makes "missing" and "closed" compare equal; update carries the new state.
+ assertEquals(List.of("", "0:2:5000::0", "86400"), client.scriptArgs.get(0));
+ }
+
+ @Test
+ void updatingToClosedRequestsDeletionViaAnEmptyUpdate() {
+ RedisToolCircuitBreakerStore store = store();
+ ToolCircuitSnapshot open = new ToolCircuitSnapshot(3L, 7_000L);
+
+ store.compareAndSet(TOOL, open, ToolCircuitSnapshot.CLOSED);
+
+ assertEquals(List.of("0:3:7000::0", "", "86400"), client.scriptArgs.get(0));
+ }
+
+ @Test
+ void probeClaimIsCarriedInTheEncodedValue() {
+ RedisToolCircuitBreakerStore store = store();
+ ToolCircuitSnapshot claimed = new ToolCircuitSnapshot(0L, 4L, 1_000L, "tok", 9_000L);
+
+ store.compareAndSet(TOOL, ToolCircuitSnapshot.CLOSED, claimed);
+
+ assertEquals(List.of("", "0:4:1000:tok:9000", "86400"), client.scriptArgs.get(0));
+ }
+
+ @Test
+ 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));
+ }
+
+ @Test
+ void scriptResultDecidesWhetherTheSwapCommitted() {
+ RedisToolCircuitBreakerStore store = store();
+
+ 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 ====================
+
+ @Test
+ void snapshotDecodesEveryField() {
+ RedisToolCircuitBreakerStore store = store();
+ client.values.put(KEY, "2:3:1767225600000:tok:1767225660000");
+
+ ToolCircuitSnapshot snapshot = store.snapshot(TOOL);
+
+ 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 anEmptyProbeFieldDecodesToNoClaim() {
+ RedisToolCircuitBreakerStore store = store();
+ client.values.put(KEY, "0:1:1000::0");
+
+ ToolCircuitSnapshot snapshot = store.snapshot(TOOL);
+
+ assertNull(snapshot.probeToken());
+ assertFalse(snapshot.hasActiveProbe(0L));
+ }
+
+ @Test
+ void missingKeyDecodesAsClosed() {
+ assertEquals(ToolCircuitSnapshot.CLOSED, store().snapshot(TOOL));
+ }
+
+ @Test
+ void encodingRoundTripsThroughDecoding() {
+ RedisToolCircuitBreakerStore store = store();
+ 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(original, store.snapshot(TOOL));
+ }
+
+ @Test
+ void unreadableValuesFailOpenRatherThanWithholdingForever() {
+ RedisToolCircuitBreakerStore store = store();
+
+ 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 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(
+ IllegalArgumentException.class,
+ () -> new RedisToolCircuitBreakerStore(client, "cb:", null));
+ }
+
+ private RedisToolCircuitBreakerStore store() {
+ return new RedisToolCircuitBreakerStore(client, "cb:", Duration.ofHours(24));
+ }
+
+ /** 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 (one turn may probe; see below)
+ * HALF_OPEN --probe succeeds--> CLOSED (failure streak cleared)
+ * HALF_OPEN --probe fails--> OPEN (next generation, longer cooldown)
+ *
+ *
+ * Concurrency
+ *
+ *
+ *
+ *
+ * Tools are supervised by opt-in, not by default
+ *
+ * Cooldown grows with each trip
+ *
+ *
+ *
+ *
+ * Recovery probes
+ *
+ * {@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.
+ *
+ * Key layout
+ *
+ *
+ * "<failureCount>:<generation>:<openedAt>:<probeToken>:<probeLeaseUntil>"
+ *
+ *
+ * Atomicity
+ *
+ * Expiry
+ *
+ *