From c58f804694cf5cfc33307dfbda1ec99bc7fe94b8 Mon Sep 17 00:00:00 2001 From: wing Date: Sun, 15 Feb 2026 17:50:12 +0800 Subject: [PATCH] =?UTF-8?q?fix(agents):=20=E4=BF=AE=E5=A4=8D=20CircuitBrea?= =?UTF-8?q?ker=20=E7=AB=9E=E6=80=81=E5=92=8C=20OverflowRecovery=20?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E6=96=AD=E8=A3=82=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- js/agents/llm/overflow-recovery.js | 187 +++++++++++++++++- js/agents/shared/utils/circuit-breaker.js | 23 ++- .../unit/agents/llm/overflow-recovery.test.js | 110 +++++++++++ .../shared/utils/circuit-breaker.test.js | 24 +++ 4 files changed, 332 insertions(+), 12 deletions(-) diff --git a/js/agents/llm/overflow-recovery.js b/js/agents/llm/overflow-recovery.js index 99a6b8b1..0028762b 100644 --- a/js/agents/llm/overflow-recovery.js +++ b/js/agents/llm/overflow-recovery.js @@ -1,5 +1,8 @@ import { toNonEmptyString } from "../shared/index.js"; +const OVERFLOW_RETRY_STATE_VERSION = 1; +const DEFAULT_OVERFLOW_RETRY_ARCHIVE_KEY = "llm:overflow-recovery"; + /** * @typedef {object} ContextOverflowInfo * @property {number} inputLength @@ -8,6 +11,15 @@ import { toNonEmptyString } from "../shared/index.js"; * @property {string} [provider] */ +/** + * @typedef {object} OverflowRetryState + * @property {number} version + * @property {boolean} inProgress + * @property {number} attempt + * @property {number} [currentMaxTokens] + * @property {number} [updatedAt] + */ + function extractErrorMessage(error) { const direct = toNonEmptyString(error?.message); if (direct) return direct; @@ -97,6 +109,147 @@ export function computeOverflowRetryMaxTokens(info, { minTokens = 256, bufferTok return Math.max(min, Math.floor(available)); } +function normalizeArchiveKey(key) { + return toNonEmptyString(key) || DEFAULT_OVERFLOW_RETRY_ARCHIVE_KEY; +} + +/** + * @param {unknown} raw + * @returns {OverflowRetryState | null} + */ +function normalizeOverflowRetryState(raw) { + const fromNodeStates = raw?.nodeStates?.overflowRecovery; + const fromPayload = raw?.overflowRecovery; + const candidate = + fromNodeStates && typeof fromNodeStates === "object" + ? fromNodeStates + : fromPayload && typeof fromPayload === "object" + ? fromPayload + : raw && typeof raw === "object" + ? raw + : null; + + if (!candidate || typeof candidate !== "object") return null; + + const attempt = Number(candidate.attempt); + if (!Number.isFinite(attempt) || attempt < 0) return null; + + const out = { + version: OVERFLOW_RETRY_STATE_VERSION, + inProgress: candidate.inProgress === undefined ? attempt > 0 : Boolean(candidate.inProgress), + attempt: Math.max(0, Math.floor(attempt)), + }; + + const currentMaxTokens = Number(candidate.currentMaxTokens); + if (Number.isFinite(currentMaxTokens) && currentMaxTokens > 0) { + out.currentMaxTokens = Math.max(1, Math.floor(currentMaxTokens)); + } + + const updatedAt = Number(candidate.updatedAt); + if (Number.isFinite(updatedAt) && updatedAt > 0) { + out.updatedAt = Math.floor(updatedAt); + } + + return out; +} + +/** + * @param {object | null | undefined} archive + * @param {string} archiveKey + * @returns {Promise} + */ +async function loadOverflowRetryState(archive, archiveKey) { + if (!archive || typeof archive !== "object") return null; + + try { + let raw = null; + if (typeof archive.get === "function") { + raw = await archive.get(archiveKey); + } else if (typeof archive.load === "function") { + raw = await archive.load(archiveKey); + } else if (typeof archive.restore === "function") { + raw = await archive.restore(archiveKey); + } else { + return null; + } + return normalizeOverflowRetryState(raw); + } catch { + return null; + } +} + +/** + * @param {object | null | undefined} archive + * @param {string} archiveKey + * @param {Partial} state + * @returns {Promise} + */ +async function persistOverflowRetryState(archive, archiveKey, state) { + if (!archive || typeof archive !== "object") return false; + + const payload = { + version: OVERFLOW_RETRY_STATE_VERSION, + inProgress: Boolean(state?.inProgress), + attempt: Number.isFinite(state?.attempt) ? Math.max(0, Math.floor(state.attempt)) : 0, + updatedAt: Date.now(), + }; + + if (Number.isFinite(state?.currentMaxTokens) && state.currentMaxTokens > 0) { + payload.currentMaxTokens = Math.max(1, Math.floor(state.currentMaxTokens)); + } + + if (Number.isFinite(state?.updatedAt) && state.updatedAt > 0) { + payload.updatedAt = Math.floor(state.updatedAt); + } + + try { + if (typeof archive.set === "function") { + await archive.set(archiveKey, payload); + return true; + } + if (typeof archive.save === "function") { + await archive.save(archiveKey, { + nodeStates: { overflowRecovery: payload }, + timestamp: String(payload.updatedAt), + metadata: { + kind: "overflow_recovery_retry_state", + version: OVERFLOW_RETRY_STATE_VERSION, + }, + }); + return true; + } + } catch { + return false; + } + + return false; +} + +/** + * @param {object | null | undefined} archive + * @param {string} archiveKey + * @param {number} currentMaxTokens + * @returns {Promise} + */ +async function clearOverflowRetryState(archive, archiveKey, currentMaxTokens) { + if (!archive || typeof archive !== "object") return; + + if (typeof archive.delete === "function" && typeof archive.set === "function") { + try { + await archive.delete(archiveKey); + return; + } catch { + // Ignore delete failures and fall back to reset marker. + } + } + + await persistOverflowRetryState(archive, archiveKey, { + inProgress: false, + attempt: 0, + currentMaxTokens, + }); +} + /** * Execute `fn(maxTokens)` with context-overflow recovery: * - Detect a context overflow error @@ -111,6 +264,8 @@ export function computeOverflowRetryMaxTokens(info, { minTokens = 256, bufferTok * bufferTokens?: number, * maxRetries?: number, * onOverflow?: (args: { error: unknown, info: ContextOverflowInfo, nextMaxTokens: number, attempt: number }) => (void | Promise), + * archive?: { get?: (key: string) => Promise, set?: (key: string, value: unknown) => Promise, delete?: (key: string) => Promise, load?: (key: string) => Promise, save?: (runId: string, value: unknown) => Promise, restore?: (checkpointId: string) => Promise }, + * archiveKey?: string, * }} options * @returns {Promise} */ @@ -120,15 +275,35 @@ export async function executeWithOverflowRecovery(fn, options) { const bufferTokens = options?.bufferTokens ?? 128; const maxRetries = Number.isFinite(options?.maxRetries) ? Math.max(0, Math.floor(options.maxRetries)) : 2; const onOverflow = typeof options?.onOverflow === "function" ? options.onOverflow : null; + const archive = options?.archive && typeof options.archive === "object" ? options.archive : null; + const archiveKey = normalizeArchiveKey(options?.archiveKey); let currentMaxTokens = Number.isFinite(initial) ? Math.max(1, Math.floor(initial)) : 1024; + let startAttempt = 0; - for (let attempt = 0; attempt <= maxRetries; attempt++) { + const persistedState = await loadOverflowRetryState(archive, archiveKey); + if (persistedState?.inProgress && persistedState.attempt > 0) { + if (persistedState.attempt <= maxRetries) { + startAttempt = persistedState.attempt; + if (Number.isFinite(persistedState.currentMaxTokens) && persistedState.currentMaxTokens > 0) { + currentMaxTokens = Math.max(1, Math.floor(persistedState.currentMaxTokens)); + } + } else { + await clearOverflowRetryState(archive, archiveKey, currentMaxTokens); + } + } + + for (let attempt = startAttempt; attempt <= maxRetries; attempt++) { try { - return await fn(currentMaxTokens); + const result = await fn(currentMaxTokens); + await clearOverflowRetryState(archive, archiveKey, currentMaxTokens); + return result; } catch (error) { const info = parseContextOverflowError(error); - if (!info || attempt >= maxRetries) throw error; + if (!info || attempt >= maxRetries) { + await clearOverflowRetryState(archive, archiveKey, currentMaxTokens); + throw error; + } const next = computeOverflowRetryMaxTokens(info, { minTokens, bufferTokens }); const nextMaxTokens = Number.isFinite(next) ? Math.max(1, Math.floor(next)) : Math.max(1, Math.floor(currentMaxTokens * 0.75)); @@ -137,6 +312,11 @@ export async function executeWithOverflowRecovery(fn, options) { const reduced = nextMaxTokens < currentMaxTokens ? nextMaxTokens : Math.max(1, Math.floor(currentMaxTokens * 0.75)); currentMaxTokens = reduced; + await persistOverflowRetryState(archive, archiveKey, { + inProgress: true, + attempt: attempt + 1, + currentMaxTokens, + }); if (onOverflow) { await onOverflow({ error, info, nextMaxTokens: currentMaxTokens, attempt: attempt + 1 }); @@ -153,4 +333,3 @@ export default { computeOverflowRetryMaxTokens, executeWithOverflowRecovery, }; - diff --git a/js/agents/shared/utils/circuit-breaker.js b/js/agents/shared/utils/circuit-breaker.js index 21dca6c8..6b921ec8 100644 --- a/js/agents/shared/utils/circuit-breaker.js +++ b/js/agents/shared/utils/circuit-breaker.js @@ -193,14 +193,19 @@ export class CircuitBreaker { // --- 内部方法 --- _checkStateTransition() { - const now = this._time.now(); + if (this._state !== CircuitState.OPEN) return; - if (this._state === CircuitState.OPEN) { - const elapsed = now - this._openedAt; - if (elapsed >= this.openDurationMs) { - this._transitionTo(CircuitState.HALF_OPEN, "timeout_elapsed"); - } - } + const expectedState = this._state; + const expectedOpenedAt = this._openedAt; + const now = this._time.now(); + const elapsed = now - expectedOpenedAt; + + if (elapsed < this.openDurationMs) return; + + this._transitionTo(CircuitState.HALF_OPEN, "timeout_elapsed", { + expectedState, + expectedOpenedAt, + }); } _onSuccess() { @@ -238,8 +243,10 @@ export class CircuitBreaker { } } - _transitionTo(newState, reason) { + _transitionTo(newState, reason, guard = null) { const prevState = this._state; + if (guard && guard.expectedState !== undefined && prevState !== guard.expectedState) return; + if (guard && guard.expectedOpenedAt !== undefined && this._openedAt !== guard.expectedOpenedAt) return; if (prevState === newState) return; this._state = newState; diff --git a/tests/unit/agents/llm/overflow-recovery.test.js b/tests/unit/agents/llm/overflow-recovery.test.js index 23bba62a..d4e8b264 100644 --- a/tests/unit/agents/llm/overflow-recovery.test.js +++ b/tests/unit/agents/llm/overflow-recovery.test.js @@ -223,6 +223,116 @@ describe("executeWithOverflowRecovery", () => { }); }); + it("resumes from archived in-progress retry state", async () => { + let snapshot = { + nodeStates: { + overflowRecovery: { + version: 1, + inProgress: true, + attempt: 1, + currentMaxTokens: 333, + updatedAt: Date.now(), + }, + }, + }; + + const archive = { + load: vi.fn(async () => snapshot), + save: vi.fn(async (_runId, payload) => { + snapshot = payload; + }), + }; + + const fn = vi.fn(async (maxTokens) => maxTokens); + const out = await executeWithOverflowRecovery(fn, { + initialMaxTokens: 500, + maxRetries: 2, + archive, + archiveKey: "overflow:case1", + }); + + expect(out).toBe(333); + expect(fn.mock.calls.map((call) => call[0])).toEqual([333]); + expect(archive.load).toHaveBeenCalledWith("overflow:case1"); + expect(archive.save).toHaveBeenCalled(); + const lastCall = archive.save.mock.calls[archive.save.mock.calls.length - 1]; + expect(lastCall[0]).toBe("overflow:case1"); + expect(lastCall[1]).toMatchObject({ + nodeStates: { + overflowRecovery: { + inProgress: false, + attempt: 0, + currentMaxTokens: 333, + }, + }, + }); + }); + + it("persists retry state when overflow happens and clears it after success", async () => { + const store = new Map(); + const archive = { + get: vi.fn(async (key) => store.get(key) ?? null), + set: vi.fn(async (key, value) => { + store.set(key, value); + }), + delete: vi.fn(async (key) => { + store.delete(key); + }), + }; + + const overflowErr = makeOpenAiError(1000, 1200, 900, 300); + const fn = vi.fn(async (maxTokens) => { + if (fn.mock.calls.length === 1) throw overflowErr; + return maxTokens; + }); + + const out = await executeWithOverflowRecovery(fn, { + initialMaxTokens: 600, + maxRetries: 1, + archive, + archiveKey: "overflow:case2", + }); + + expect(out).toBe(256); + expect(archive.set).toHaveBeenCalledWith( + "overflow:case2", + expect.objectContaining({ + inProgress: true, + attempt: 1, + currentMaxTokens: 256, + }) + ); + expect(archive.delete).toHaveBeenCalledWith("overflow:case2"); + expect(store.has("overflow:case2")).toBe(false); + }); + + it("tolerates archive read/write failures", async () => { + const archive = { + load: vi.fn(async () => { + throw new Error("load failed"); + }), + save: vi.fn(async () => { + throw new Error("save failed"); + }), + }; + + const overflowErr = makeOpenAiError(1000, 1200, 900, 300); + const fn = vi.fn(async (maxTokens) => { + if (fn.mock.calls.length === 1) throw overflowErr; + return maxTokens; + }); + + const out = await executeWithOverflowRecovery(fn, { + initialMaxTokens: 500, + maxRetries: 1, + archive, + archiveKey: "overflow:case3", + }); + + expect(out).toBe(256); + expect(fn.mock.calls.map((call) => call[0])).toEqual([500, 256]); + }); + it("ensures progress when computed nextMaxTokens would not reduce", async () => { const overflowErr = new Error("maximum context length is 8192 tokens"); diff --git a/tests/unit/agents/shared/utils/circuit-breaker.test.js b/tests/unit/agents/shared/utils/circuit-breaker.test.js index 2900ef9c..bd42b5a0 100644 --- a/tests/unit/agents/shared/utils/circuit-breaker.test.js +++ b/tests/unit/agents/shared/utils/circuit-breaker.test.js @@ -150,6 +150,30 @@ describe("CircuitBreaker", () => { expect(breaker.state).toBe(CircuitState.HALF_OPEN); }); + it("ignores stale timeout transition when state changes during check", () => { + const b = new CircuitBreaker({ + openDurationMs: 1, + time: mockTime, + }); + + b.trip("manual"); + now = 10; + + let openedAt = b._openedAt; + Object.defineProperty(b, "_openedAt", { + configurable: true, + get() { + b._state = CircuitState.CLOSED; + return openedAt; + }, + set(value) { + openedAt = value; + }, + }); + + expect(b.state).toBe(CircuitState.CLOSED); + }); + it("trip opens and records openedAt with reason", () => { const events = []; const b = new CircuitBreaker({