github-featured/chopratejas--headroom
2
0
Fork 0
Code Issues Pull Requests Actions 16 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(openclaw-plugin): circuit breaker + per-request timeout for proxy resilience (#639)

Browse Source 
## Description

This change adds bounded timeout and circuit-breaker behavior so
OpenClaw can degrade safely when Headroom or the upstream stream stalls,
while returning structured proxy errors instead of hanging.

Closes #638 by improving OpenClaw/proxy resilience when the Headroom
proxy stalls or Anthropic resets a stream. The PR adds proxy-side
handling for `httpx.RemoteProtocolError`, returns structured 502
responses for otherwise unhandled proxy middleware errors, and adds
OpenClaw plugin timeout/circuit-breaker fallback behavior.

## Type of Change

- [x] Bug fix
- [ ] New feature
- [x] Documentation
- [ ] Refactor
- [x] Tests only

## Changes Made

- Added OpenClaw plugin per-request compression timeout and circuit
breaker fallback.
- Cleared timeout timers after successful or failed compression so
successful calls do not leave pending timers.
- Added a focused Vitest regression for timeout cleanup.
- Added `contracts.tools` for `headroom_retrieve` without whole-file
manifest reformatting.
- Added proxy handling for mid-stream `httpx.RemoteProtocolError` and
structured 502 fallback behavior.
- Documented the new OpenClaw resilience configuration fields.

## Testing

- [x] Unit tests
- [x] Integration-style proxy tests
- [x] Typecheck/build
- [ ] Manual testing

### Test Output

```text
cd plugins/openclaw && npm test
Test Files 6 passed (6), Tests 55 passed (55)

cd plugins/openclaw && npm run typecheck
passed

cd plugins/openclaw && npm run build
Build success

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --with pytest --with pytest-asyncio --with fastapi --with httpx --with uvicorn --with h2 python -m pytest tests/test_proxy_streaming_resilience.py -q
24 passed in 2.16s
```

## Real Behavior Proof

- Environment: Windows 11, Node/npm from local plugin worktree, Python
3.13.3, focused local worktree for PR #639.
- Exact command / steps: Installed plugin dependencies, ran OpenClaw
plugin tests/typecheck/build, and ran the proxy streaming resilience
suite with required async/FastAPI/httpx extras.
- Observed result: Plugin tests, typecheck, build, and proxy resilience
tests all passed.
- Not tested: Live OpenClaw gateway session in this pass; original
reporter previously verified patched files in a container and OpenClaw
degraded/recovered cleanly.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

---------

Co-authored-by: Sergei Baikin <sergei.baikin@fotograf.de>
Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
Co-authored-by: JD Davis <jd@jds-macbook-air.tail2a279.ts.net>
This commit is contained in:
Sergei Baikin
2026-08-13 18:52:01 +02:00
committed by GitHub
parent 82526191a1
commit 6576ef639c
4 changed files with 140 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/openclaw/README.md
+3
View File
@@ -205,6 +205,9 @@ Compression is lossless via CCR (Compress-Cache-Retrieve): originals are stored
| `pythonPath` | auto-detected | Optional Python executable override for Python fallback launcher. |
| `autoStart` | `false` | Opt-in auto-start for a local `headroom proxy` if not already running (local URLs only; ignored for remote proxies). Keep `false` when systemd owns the proxy. |
| `startupTimeoutMs` | `20000` | Time to wait for auto-started proxy to become healthy |
| `requestTimeoutMs` | `30000` | Maximum milliseconds to wait for a single `compress()` call. If the proxy hangs or is slow, the call is cancelled after this deadline and the original uncompressed messages are used as a fallback. |
| `circuitBreakerThreshold` | `3` | Number of consecutive `assemble()` errors before the circuit breaker opens and all requests bypass the proxy. Prevents cascading failures when the proxy is unhealthy. |
| `circuitBreakerCooldownMs` | `60000` | How long (ms) the circuit breaker stays open after the threshold is reached. After the cool-down the breaker resets automatically and the next request re-probes the proxy via `/health`. |
| `routeCodexViaProxy` | `true` | Rewrite OpenClaw's built-in `openai-codex` provider to use the active Headroom proxy in memory so upstream Codex requests pass through Headroom. |
| `gatewayProviderIds` | `[]` | Optional explicit list of OpenClaw provider ids to route through the active Headroom proxy in memory. Friendly aliases `codex`, `claude`, `copilot`, and `gemini` are also accepted. When set, this overrides the default `openai-codex` routing list. |
plugins/openclaw/openclaw.plugin.json
+27
View File
@@ -22,6 +22,18 @@
"label": "Connect Timeout Seconds",
"help": "Optional upstream connection timeout for the auto-started local Headroom proxy. Lower values surface network failures sooner."
},
"requestTimeoutMs": {
"label": "Compression Request Timeout",
"help": "Maximum milliseconds to wait for one compression request before returning the original messages."
},
"circuitBreakerThreshold": {
"label": "Circuit Breaker Threshold",
"help": "Consecutive compression failures before temporarily bypassing Headroom."
},
"circuitBreakerCooldownMs": {
"label": "Circuit Breaker Cooldown",
"help": "Milliseconds to bypass compression after the circuit breaker opens."
},
"routeCodexViaProxy": {
"label": "Route OpenAI Codex Via Headroom",
"help": "When enabled, OpenClaw will use the active Headroom proxy as the in-memory upstream base URL for the built-in openai-codex provider so provider traffic flows through Headroom."
@@ -69,6 +81,21 @@
"type": "integer",
"minimum": 1
},
"requestTimeoutMs": {
"type": "integer",
"minimum": 1,
"default": 30000
},
"circuitBreakerThreshold": {
"type": "integer",
"minimum": 1,
"default": 3
},
"circuitBreakerCooldownMs": {
"type": "integer",
"minimum": 1,
"default": 60000
},
"routeCodexViaProxy": {
"type": "boolean",
"default": true
plugins/openclaw/src/engine.ts
+56 -6
View File
@@ -11,8 +11,22 @@ import { compress } from "headroom-ai";
import { ProxyManager, defaultLogger, type ProxyManagerConfig, type ProxyManagerLogger } from "./proxy-manager.js";
import { agentToOpenAI, normalizeAgentMessages, openAIToAgent } from "./convert.js";
/** Race a promise against a timeout and always release the timer. */
function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
let timerId: ReturnType<typeof setTimeout> | undefined;
const timer = new Promise<never>((_, reject) => {
timerId = setTimeout(() => reject(new Error(`headroom compress() timed out after ${ms}ms`)), ms);
});
return Promise.race([promise, timer]).finally(() => {
if (timerId !== undefined) clearTimeout(timerId);
});
}
export interface HeadroomEngineConfig extends ProxyManagerConfig {
enabled?: boolean;
requestTimeoutMs?: number;
circuitBreakerThreshold?: number;
circuitBreakerCooldownMs?: number;
}
export class HeadroomContextEngine {
@@ -36,6 +50,7 @@ export class HeadroomContextEngine {
totalTokensBefore: 0,
compactions: 0,
};
private circuit = { errors: 0, openUntilMs: 0 };
constructor(config: HeadroomEngineConfig = {}, logger?: ProxyManagerLogger) {
this.config = config;
@@ -97,19 +112,28 @@ export class HeadroomContextEngine {
return { messages: normalizeAgentMessages(params.messages), estimatedTokens: 0 };
}
if (this.isCircuitOpen()) {
this.logger.warn("[headroom] Circuit open — using uncompressed messages");
return { messages: normalizeAgentMessages(params.messages), estimatedTokens: 0 };
}
try {
// Convert AgentMessage → OpenAI format
const openaiMessages = agentToOpenAI(params.messages);
// Compress via proxy — pass tokenBudget so RollingWindow enforces it
const result = await compress(openaiMessages, {
model: params.model ?? "claude-sonnet-4-5",
baseUrl: this.proxyUrl,
fallback: true,
tokenBudget: params.tokenBudget,
} as any);
const result = await withTimeout(
compress(openaiMessages, {
model: params.model ?? "claude-sonnet-4-5",
baseUrl: this.proxyUrl,
fallback: true,
tokenBudget: params.tokenBudget,
} as any),
this.config.requestTimeoutMs ?? 30_000,
);
if (!result.compressed || result.tokensSaved === 0) {
this.resetCircuit();
return {
messages: normalizeAgentMessages(params.messages),
estimatedTokens: result.tokensBefore,
@@ -118,6 +142,7 @@ export class HeadroomContextEngine {
// Convert back to AgentMessage format
const compressedAgentMessages = openAIToAgent(result.messages);
this.resetCircuit();
// Track stats
this.stats.totalCompressions++;
@@ -138,6 +163,7 @@ export class HeadroomContextEngine {
};
} catch (error) {
this.logger.error(`Assemble failed: ${error}`);
this.tripCircuit(error);
// Graceful fallback: return original messages
return { messages: normalizeAgentMessages(params.messages), estimatedTokens: 0 };
}
@@ -240,6 +266,30 @@ export class HeadroomContextEngine {
return this.proxyStartupError;
}
private isCircuitOpen(): boolean {
const threshold = this.config.circuitBreakerThreshold ?? 3;
if (this.circuit.errors < threshold) return false;
if (Date.now() < this.circuit.openUntilMs) return true;
this.circuit = { errors: 0, openUntilMs: 0 };
return false;
}
private tripCircuit(error: unknown): void {
this.circuit.errors += 1;
const threshold = this.config.circuitBreakerThreshold ?? 3;
if (this.circuit.errors < threshold) return;
const cooldownMs = this.config.circuitBreakerCooldownMs ?? 60_000;
this.circuit.openUntilMs = Date.now() + cooldownMs;
this.logger.warn(
`[headroom] Circuit breaker opened after ${this.circuit.errors} errors ` +
`(last: ${String(error)}); bypassing compression for ${cooldownMs}ms`,
);
}
private resetCircuit(): void {
this.circuit = { errors: 0, openUntilMs: 0 };
}
ensureProxyStarted(): void {
if (this.config.enabled === false || this.proxyUrl || this.proxyStartupPromise) {
return;
plugins/openclaw/test/engine.test.ts
+54
View File
@@ -24,8 +24,10 @@ vi.mock("../src/proxy-manager.js", () => ({
}));
import { HeadroomContextEngine } from "../src/engine.js";
import { compress } from "headroom-ai";
afterEach(() => {
vi.mocked(compress).mockReset();
mocked.start.mockReset();
mocked.start.mockResolvedValue("http://127.0.0.1:8787");
mocked.stop.mockClear();
@@ -196,4 +198,56 @@ describe("HeadroomContextEngine proxy startup helpers", () => {
});
expect(mocked.start).toHaveBeenCalledTimes(1);
});
it("clears the request timeout after successful compression", async () => {
vi.useFakeTimers();
try {
vi.mocked(compress).mockResolvedValue({
compressed: false,
messages: [{ role: "user", content: "hello" }],
tokensBefore: 5,
tokensAfter: 5,
tokensSaved: 0,
});
const engine = new HeadroomContextEngine({ requestTimeoutMs: 30_000 });
(engine as { proxyUrl: string | null }).proxyUrl = "http://127.0.0.1:8787";
await expect(
engine.assemble({
sessionId: "session-1",
messages: [{ role: "user", content: "hello" }],
}),
).resolves.toEqual({
messages: [{ role: "user", content: "hello" }],
estimatedTokens: 5,
});
expect(vi.getTimerCount()).toBe(0);
} finally {
vi.useRealTimers();
}
});
it("opens the circuit after consecutive compression failures", async () => {
vi.mocked(compress).mockRejectedValue(new Error("proxy stalled"));
const messages = [{ role: "user", content: "hello" }];
const engine = new HeadroomContextEngine({
circuitBreakerThreshold: 2,
circuitBreakerCooldownMs: 60_000,
});
(engine as { proxyUrl: string | null }).proxyUrl = "http://127.0.0.1:8787";
await engine.assemble({ sessionId: "session-1", messages });
await engine.assemble({ sessionId: "session-1", messages });
await expect(engine.assemble({ sessionId: "session-1", messages })).resolves.toEqual({
messages,
estimatedTokens: 0,
});
expect(compress).toHaveBeenCalledTimes(2);
expect(mocked.logger.warn).toHaveBeenCalledWith(
expect.stringContaining("Circuit breaker opened"),
);
});
});