fix(agents): 修复 CircuitBreaker 竞态和 OverflowRecovery 状态断裂点

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<OverflowRetryState | null>}
+ */
+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<OverflowRetryState>} state
+ * @returns {Promise<boolean>}
+ */
+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<void>}
+ */
+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<void>),
+ *   archive?: { get?: (key: string) => Promise<unknown>, set?: (key: string, value: unknown) => Promise<unknown>, delete?: (key: string) => Promise<unknown>, load?: (key: string) => Promise<unknown>, save?: (runId: string, value: unknown) => Promise<unknown>, restore?: (checkpointId: string) => Promise<unknown> },
+ *   archiveKey?: string,
  * }} options
  * @returns {Promise<T>}
  */
@@ -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,
 };
-

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;

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");
 

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({