feat(plugin): add rate-limit-circuit-breaker to prevent death loops in group chats

guci314 · claude · guci314 · commit 0fcd5faac687 · 2026-04-10T10:35:21.000+08:00
When multiple agents share a group chat with requireMention:false, a rate-limit
error surfaced by one agent triggers other agents to respond, causing them to
also hit rate limits in an infinite cascade. This plugin detects consecutive
rate-limit error messages per room and suppresses them using a circuit breaker
pattern with exponential backoff.

Verified in production: a 5-agent Matrix group hit a 13+ minute death loop
(146 surface_error events) before this fix. After deployment, zero surface_error
events with the same rate limit conditions.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/extensions/rate-limit-circuit-breaker/README.md b/extensions/rate-limit-circuit-breaker/README.md
@@ -0,0 +1,68 @@
+# Rate Limit Circuit Breaker
+
+Prevents death loops in multi-agent group chats (Matrix, Discord, etc.) when LLM API rate limits are hit.
+
+## Problem
+
+When multiple agents share a group chat with `requireMention: false`, a single rate-limit error can cascade into an infinite loop:
+
+1. Agent A receives a message → calls LLM API → gets rate-limited (429)
+2. Gateway surfaces the error as a chat message: "API rate limit reached"
+3. Agent B sees this message → tries to respond → also gets rate-limited
+4. Repeat indefinitely
+
+This loop is self-sustaining because the error messages themselves trigger more API calls.
+
+## Solution
+
+This plugin hooks into `message_sending` to detect and suppress repeated rate-limit error messages using a circuit breaker pattern:
+
+- **CLOSED**: Normal operation. Counts consecutive rate-limit errors per room.
+- **OPEN**: After N consecutive errors (default: 3), suppresses further error messages for a cooldown period. Normal messages still flow through.
+- **HALF_OPEN**: After cooldown expires, allows one error message through as a retry probe. If the next message is normal (non-error), the circuit fully resets. If another error occurs, the circuit re-opens with doubled cooldown.
+
+Exponential backoff ensures the cooldown grows (60s → 120s → 240s → ... up to 10 minutes) if the rate limit persists.
+
+## Configuration
+
+```json
+{
+  "plugins": {
+    "entries": {
+      "rate-limit-circuit-breaker": {
+        "enabled": true,
+        "config": {
+          "maxConsecutiveErrors": 3,
+          "baseCooldownMs": 60000,
+          "maxCooldownMs": 600000
+        }
+      }
+    }
+  }
+}
+```
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `maxConsecutiveErrors` | 3 | Consecutive rate-limit errors before circuit opens |
+| `baseCooldownMs` | 60000 | Base cooldown (60s), doubles each trip |
+| `maxCooldownMs` | 600000 | Maximum cooldown cap (10 minutes) |
+
+## Recommended companion config
+
+For best results, also configure fallback models so rate limits trigger model failover before reaching `surface_error`:
+
+```json
+{
+  "agents": {
+    "defaults": {
+      "model": {
+        "primary": "provider/model-a",
+        "fallbacks": ["provider/model-b", "provider/model-c"]
+      }
+    }
+  }
+}
+```
+
+With fallback models, rate limits are handled by model switching (first line of defense). The circuit breaker acts as a second line of defense for when all models are exhausted.
diff --git a/extensions/rate-limit-circuit-breaker/index.ts b/extensions/rate-limit-circuit-breaker/index.ts
@@ -0,0 +1,97 @@
+/**
+ * Rate Limit Circuit Breaker Plugin for OpenClaw
+ *
+ * Prevents death loops in multi-agent group chats (Matrix, Discord, etc.)
+ * where a rate-limit error surfaced by one agent triggers other agents to
+ * respond, causing them to also hit rate limits in an infinite cascade.
+ *
+ * Mechanism:
+ *   1. Hooks into `message_sending` to inspect every outgoing message
+ *   2. Detects rate-limit/overload error messages by pattern matching
+ *   3. Tracks consecutive rate-limit errors per room (channel + target)
+ *   4. After N consecutive errors, opens the circuit breaker:
+ *      - Suppresses further error messages for a cooldown period
+ *      - Uses exponential backoff on repeated trips
+ *   5. After cooldown, allows one retry (half-open state)
+ *   6. On success (non-error message), fully resets the circuit
+ */
+
+import { RateLimitCircuitBreaker } from "./src/circuit-breaker.js";
+
+// Singleton — shared across all hooks for the lifetime of the gateway process
+let breaker: RateLimitCircuitBreaker | null = null;
+
+// Periodic cleanup interval handle
+let cleanupInterval: ReturnType<typeof setInterval> | null = null;
+
+export default function register(api: any) {
+  const pluginConfig = api.pluginConfig ?? {};
+  const logger = api.logger ?? { warn: console.warn, debug: undefined };
+
+  breaker = new RateLimitCircuitBreaker(
+    {
+      maxConsecutiveErrors: pluginConfig.maxConsecutiveErrors ?? 3,
+      baseCooldownMs: pluginConfig.baseCooldownMs ?? 60_000,
+      maxCooldownMs: pluginConfig.maxCooldownMs ?? 600_000,
+    },
+    {
+      warn: (msg: string) => logger.warn(msg),
+      debug: logger.debug ? (msg: string) => logger.debug!(msg) : undefined,
+    },
+  );
+
+  // --- message_sending hook: intercept outgoing messages ---
+  api.on(
+    "message_sending",
+    (
+      event: { to: string; content: string; metadata?: Record<string, unknown> },
+      ctx: { channelId: string; accountId?: string; conversationId?: string },
+    ) => {
+      if (!breaker || !event.content) return;
+
+      const channelId = ctx.channelId ?? (event.metadata?.channel as string) ?? "unknown";
+      const to = event.to ?? "";
+
+      if (!to) return;
+
+      const suppress = breaker.shouldSuppress(channelId, to, event.content);
+      if (suppress) {
+        return { cancel: true };
+      }
+      // Allow the message through (no modification)
+      return undefined;
+    },
+    { priority: -100 }, // Run early so we can cancel before other hooks process
+  );
+
+  // --- gateway_start hook: set up periodic cleanup ---
+  api.on("gateway_start", () => {
+    // Clean up stale circuit breaker entries every 30 minutes
+    cleanupInterval = setInterval(() => {
+      breaker?.cleanup(3_600_000); // 1 hour max age
+    }, 30 * 60 * 1000);
+  });
+
+  // --- gateway_stop hook: teardown ---
+  api.on("gateway_stop", () => {
+    if (cleanupInterval) {
+      clearInterval(cleanupInterval);
+      cleanupInterval = null;
+    }
+  });
+
+  // --- Register a gateway method for diagnostics ---
+  api.registerGatewayMethod(
+    "circuit-breaker-status",
+    async (params: { channel?: string; to?: string }) => {
+      if (!breaker) return { status: "not_initialized" };
+      if (params.channel && params.to) {
+        const state = breaker.getState(params.channel, params.to);
+        return { room: `${params.channel}:${params.to}`, state: state ?? "no_data" };
+      }
+      return { status: "ok", message: "Pass channel and to params to query a specific room" };
+    },
+  );
+
+  logger.warn("[rate-limit-circuit-breaker] Plugin registered");
+}
diff --git a/extensions/rate-limit-circuit-breaker/openclaw.plugin.json b/extensions/rate-limit-circuit-breaker/openclaw.plugin.json
@@ -0,0 +1,26 @@
+{
+  "id": "rate-limit-circuit-breaker",
+  "name": "Rate Limit Circuit Breaker",
+  "description": "Prevents death loops in group chats when LLM API rate limits are hit. Suppresses repeated rate-limit error messages with exponential backoff to stop agents from triggering each other.",
+  "configSchema": {
+    "type": "object",
+    "additionalProperties": false,
+    "properties": {
+      "maxConsecutiveErrors": {
+        "type": "number",
+        "description": "Number of consecutive rate-limit errors before the circuit breaker opens (suppresses further error messages). Default: 3",
+        "default": 3
+      },
+      "baseCooldownMs": {
+        "type": "number",
+        "description": "Base cooldown in milliseconds after the circuit breaker opens. Doubles with each subsequent trip (exponential backoff). Default: 60000 (60 seconds)",
+        "default": 60000
+      },
+      "maxCooldownMs": {
+        "type": "number",
+        "description": "Maximum cooldown in milliseconds (cap for exponential backoff). Default: 600000 (10 minutes)",
+        "default": 600000
+      }
+    }
+  }
+}
diff --git a/extensions/rate-limit-circuit-breaker/package.json b/extensions/rate-limit-circuit-breaker/package.json
@@ -0,0 +1,17 @@
+{
+  "name": "@openclaw/rate-limit-circuit-breaker",
+  "version": "2026.4.10",
+  "description": "Prevents death loops in multi-agent group chats caused by LLM API rate limit errors",
+  "type": "module",
+  "dependencies": {},
+  "devDependencies": {
+    "@openclaw/plugin-sdk": "workspace:*",
+    "openclaw": "workspace:*"
+  },
+  "peerDependencies": {
+    "openclaw": ">=2026.3.0"
+  },
+  "exports": {
+    ".": "./index.ts"
+  }
+}
diff --git a/extensions/rate-limit-circuit-breaker/src/circuit-breaker.test.ts b/extensions/rate-limit-circuit-breaker/src/circuit-breaker.test.ts
@@ -0,0 +1,181 @@
+import { describe, it, expect, beforeEach, vi } from "vitest";
+import {
+  RateLimitCircuitBreaker,
+  isRateLimitErrorMessage,
+  isTransientErrorMessage,
+} from "./circuit-breaker.js";
+
+describe("isRateLimitErrorMessage", () => {
+  it("matches the standard OpenClaw rate limit message", () => {
+    expect(isRateLimitErrorMessage("API rate limit reached. Please try again later.")).toBe(true);
+  });
+
+  it("matches messages with emoji prefix", () => {
+    expect(isRateLimitErrorMessage("\u26a0\ufe0f API rate limit reached. Please try again later.")).toBe(true);
+  });
+
+  it("matches 429 status code references", () => {
+    expect(isRateLimitErrorMessage("HTTP 429 Too Many Requests")).toBe(true);
+  });
+
+  it("matches rate_limit_exceeded error type", () => {
+    expect(isRateLimitErrorMessage('{"type":"rate_limit_exceeded"}')).toBe(true);
+  });
+
+  it("matches overloaded messages", () => {
+    expect(isRateLimitErrorMessage("The AI service is temporarily overloaded. Please try again in a moment.")).toBe(true);
+  });
+
+  it("does not match normal messages", () => {
+    expect(isRateLimitErrorMessage("Hello, how are you?")).toBe(false);
+    expect(isRateLimitErrorMessage("The rate of improvement is excellent")).toBe(false);
+  });
+});
+
+describe("isTransientErrorMessage", () => {
+  it("matches timeout messages", () => {
+    expect(isTransientErrorMessage("LLM request timed out.")).toBe(true);
+  });
+
+  it("matches rate limit messages (superset)", () => {
+    expect(isTransientErrorMessage("API rate limit reached. Please try again later.")).toBe(true);
+  });
+
+  it("does not match normal messages", () => {
+    expect(isTransientErrorMessage("Here is your report.")).toBe(false);
+  });
+});
+
+describe("RateLimitCircuitBreaker", () => {
+  let breaker: RateLimitCircuitBreaker;
+  const channel = "matrix";
+  const room = "!room123:server.org";
+  const errorMsg = "\u26a0\ufe0f API rate limit reached. Please try again later.";
+  const normalMsg = "Here is the analysis you requested.";
+
+  const warnSpy = vi.fn();
+  const debugSpy = vi.fn();
+
+  beforeEach(() => {
+    warnSpy.mockReset();
+    debugSpy.mockReset();
+    breaker = new RateLimitCircuitBreaker(
+      { maxConsecutiveErrors: 3, baseCooldownMs: 1000, maxCooldownMs: 8000 },
+      { warn: warnSpy, debug: debugSpy },
+    );
+  });
+
+  it("allows normal messages through", () => {
+    expect(breaker.shouldSuppress(channel, room, normalMsg)).toBe(false);
+  });
+
+  it("allows first N-1 rate limit errors through", () => {
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(false); // 1/3
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(false); // 2/3
+  });
+
+  it("suppresses the Nth consecutive error (trips the circuit)", () => {
+    breaker.shouldSuppress(channel, room, errorMsg); // 1
+    breaker.shouldSuppress(channel, room, errorMsg); // 2
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(true); // 3 -> OPEN
+    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("closed -> open"));
+  });
+
+  it("suppresses subsequent errors while circuit is open", () => {
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(true);
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(true);
+  });
+
+  it("allows normal messages while circuit is open (does not suppress non-error)", () => {
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    // Non-error messages always go through
+    expect(breaker.shouldSuppress(channel, room, normalMsg)).toBe(false);
+  });
+
+  it("transitions to half_open after cooldown and allows one retry", () => {
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    const state = breaker.getState(channel, room)!;
+    // Simulate cooldown expiration by backdating openedAt
+    state.openedAt = Date.now() - 2000; // 2s > 1s cooldown
+    // Next error triggers half_open transition + is allowed through
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(false);
+    expect(breaker.getState(channel, room)!.state).toBe("half_open");
+  });
+
+  it("re-opens with doubled cooldown if retry fails in half_open", () => {
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    const state = breaker.getState(channel, room)!;
+    state.openedAt = Date.now() - 2000;
+    breaker.shouldSuppress(channel, room, errorMsg); // -> half_open, allowed
+    // Another error in half_open
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(true);
+    const updatedState = breaker.getState(channel, room)!;
+    expect(updatedState.state).toBe("open");
+    expect(updatedState.cooldownMs).toBe(2000); // doubled from 1000
+    expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining("half_open -> open"));
+  });
+
+  it("resets to closed on success in half_open state", () => {
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    const state = breaker.getState(channel, room)!;
+    state.openedAt = Date.now() - 2000;
+    breaker.shouldSuppress(channel, room, errorMsg); // -> half_open
+    // Normal message = success
+    expect(breaker.shouldSuppress(channel, room, normalMsg)).toBe(false);
+    const updatedState = breaker.getState(channel, room)!;
+    expect(updatedState.state).toBe("closed");
+    expect(updatedState.consecutiveErrors).toBe(0);
+    expect(updatedState.tripCount).toBe(0);
+  });
+
+  it("resets error count on normal message in closed state", () => {
+    breaker.shouldSuppress(channel, room, errorMsg); // 1
+    breaker.shouldSuppress(channel, room, errorMsg); // 2
+    breaker.shouldSuppress(channel, room, normalMsg); // reset
+    breaker.shouldSuppress(channel, room, errorMsg); // 1 again
+    breaker.shouldSuppress(channel, room, errorMsg); // 2 again
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(true); // 3 -> trips
+  });
+
+  it("tracks rooms independently", () => {
+    const room2 = "!room456:server.org";
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    // room1 is open
+    expect(breaker.shouldSuppress(channel, room, errorMsg)).toBe(true);
+    // room2 is still closed
+    expect(breaker.shouldSuppress(channel, room2, errorMsg)).toBe(false);
+  });
+
+  it("caps cooldown at maxCooldownMs", () => {
+    // Trip multiple times
+    for (let trip = 0; trip < 10; trip++) {
+      // Fill up consecutive errors
+      for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+      // Now it's open. Expire the cooldown.
+      const state = breaker.getState(channel, room)!;
+      state.openedAt = Date.now() - state.cooldownMs - 100;
+      breaker.shouldSuppress(channel, room, errorMsg); // -> half_open, allowed
+      // Fail the retry
+      breaker.shouldSuppress(channel, room, errorMsg); // -> open with doubled cooldown
+    }
+    const finalState = breaker.getState(channel, room)!;
+    expect(finalState.cooldownMs).toBeLessThanOrEqual(8000);
+  });
+
+  it("recordSuccess resets the circuit fully", () => {
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    expect(breaker.getState(channel, room)!.state).toBe("open");
+    breaker.recordSuccess(channel, room);
+    expect(breaker.getState(channel, room)!.state).toBe("closed");
+    expect(breaker.getState(channel, room)!.consecutiveErrors).toBe(0);
+  });
+
+  it("cleanup removes stale entries", () => {
+    for (let i = 0; i < 3; i++) breaker.shouldSuppress(channel, room, errorMsg);
+    const state = breaker.getState(channel, room)!;
+    state.openedAt = Date.now() - 7_200_000; // 2 hours ago
+    breaker.cleanup(3_600_000); // 1 hour max age
+    expect(breaker.getState(channel, room)).toBeUndefined();
+  });
+});
diff --git a/extensions/rate-limit-circuit-breaker/src/circuit-breaker.ts b/extensions/rate-limit-circuit-breaker/src/circuit-breaker.ts
