feat(quota): add opt-in Codex auto-ping

Generalize Claude 5h auto-ping into a provider-generic scheduler and add opt-in Codex auto-ping that warms the next 5h window via a tiny gpt-5.5 request when session.resetAt slides. Default off, per-connection toggle, failure cooldown, blocking-quota skip, drains stream before success. Closes #2107 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-22 20:00:47 +00:00 · 2026-06-29 15:13:17 +07:00
parent fc8722e897
commit b66b5c68ce
9 changed files with 757 additions and 168 deletions
@@ -62,16 +62,34 @@ export const CONSOLE_LOG_CONFIG = {
 // Client-side store TTL: how long fetched data stays fresh before re-fetching
 export const CLIENT_STORE_TTL_MS = 60000;

-// Claude auto-ping: keep 5h window warm by sending a tiny request right after reset
-export const CLAUDE_AUTOPING_CONFIG = {
-  settingsKey: "claudeAutoPing",        // settings table field
+// Quota auto-ping: keep 5h windows warm by sending a tiny request right after reset.
+export const QUOTA_AUTOPING_CONFIG = {
  tickIntervalMs: 60000,                // scheduler tick
  pingLeadMs: 5000,                     // fire once reset passes (within tolerance)
-  pingModel: "claude-haiku-4-5-20251001", // cheapest model
-  pingText: "hi",
-  pingMaxTokens: 1,
  refreshAheadMs: 300000,               // refetch usage when within 5min of reset
-  fiveHourKey: "session (5h)",          // quota key returned by usage handler
+  failureCooldownMs: 900000,            // avoid failed ping spam while upstream/auth is unhealthy
+  providers: {
+    claude: {
+      settingsKey: "claudeAutoPing",    // preserve existing settings contract
+      quotaKey: "session (5h)",         // quota key returned by usage handler
+      pingModel: "claude-haiku-4-5-20251001",
+      pingText: "hi",
+      pingMaxTokens: 1,
+    },
+    codex: {
+      settingsKey: "codexAutoPing",
+      quotaKey: "session",
+      pingWhenResetAtSlides: true,
+      resetAtDriftMs: 30000,
+      minPingIntervalMs: 600000,
+      skipWhenBlockingQuotaExhausted: true,
+      // Free and Plus Codex accounts both expose gpt-5.5; avoid fallback probes that waste requests.
+      pingModel: "gpt-5.5",
+      pingText: "hi",
+      pingInstructions: "Reply with OK.",
+      pingReasoningEffort: "none",
+    },
+  },
 };

 // Re-export from providers.js for backward compatibility
@@ -1,117 +0,0 @@
-// Claude auto-ping scheduler: warms the 5h window by sending a tiny request right after reset.
-import "open-sse/index.js";
-
-import { getSettings, getProviderConnections, updateProviderConnection } from "@/lib/localDb";
-import { getClaudeUsage } from "open-sse/services/usage/claude.js";
-import { CLAUDE_CLI_SPOOF_HEADERS } from "open-sse/providers/shared.js";
-import { proxyAwareFetch } from "open-sse/utils/proxyFetch.js";
-import { resolveConnectionProxyConfig } from "@/lib/network/connectionProxy";
-import { refreshAndUpdateCredentials } from "@/app/api/usage/[connectionId]/route.js";
-import { CLAUDE_AUTOPING_CONFIG } from "@/shared/constants/config";
-
-const C = CLAUDE_AUTOPING_CONFIG;
-const PING_URL = "https://api.anthropic.com/v1/messages?beta=true";
-
-const g = (global.__claudeAutoPing ??= { interval: null, running: false, resetCache: {} });
-
-function buildProxyOptions(cfg) {
-  return {
-    connectionProxyEnabled: cfg.connectionProxyEnabled === true,
-    connectionProxyUrl: cfg.connectionProxyUrl || "",
-    connectionNoProxy: cfg.connectionNoProxy || "",
-    vercelRelayUrl: cfg.vercelRelayUrl || "",
-    strictProxy: false,
-  };
-}
-
-// Send minimal "hi" to start a fresh 5h window
-async function sendPing(accessToken, proxyOptions) {
-  const res = await proxyAwareFetch(PING_URL, {
-    method: "POST",
-    headers: {
-      ...CLAUDE_CLI_SPOOF_HEADERS,
-      "Authorization": `Bearer ${accessToken}`,
-      "content-type": "application/json",
-    },
-    body: JSON.stringify({
-      model: C.pingModel,
-      max_tokens: C.pingMaxTokens,
-      messages: [{ role: "user", content: C.pingText }],
-    }),
-  }, proxyOptions);
-  return res.ok;
-}
-
-async function pingConnection(conn) {
-  // Cached resetAt is stable for the whole 5h window; skip usage poll until near reset
-  const cachedReset = g.resetCache[conn.id];
-  if (cachedReset && Date.now() < new Date(cachedReset).getTime() - C.refreshAheadMs) return;
-
-  const proxyCfg = await resolveConnectionProxyConfig(conn.providerSpecificData);
-  const proxyOptions = buildProxyOptions(proxyCfg);
-
-  // Refresh token if needed, then read 5h reset time
-  let connection = conn;
-  try {
-    const r = await refreshAndUpdateCredentials(connection, false, proxyOptions);
-    connection = r.connection;
-  } catch (e) {
-    console.warn(`[AutoPing] ${conn.id}: refresh failed: ${e.message}`);
-    return;
-  }
-
-  const usage = await getClaudeUsage(connection.accessToken, proxyOptions);
-  const resetAt = usage?.quotas?.[C.fiveHourKey]?.resetAt;
-  if (!resetAt) return;
-
-  // Cache resetAt to gate future ticks
-  g.resetCache[conn.id] = resetAt;
-
-  const resetMs = new Date(resetAt).getTime();
-  const now = Date.now();
-
-  // Only ping once per reset cycle, right after window flips
-  if (now < resetMs - C.pingLeadMs) return;
-  if (connection.lastPingedResetAt === resetAt) return;
-
-  const ok = await sendPing(connection.accessToken, proxyOptions);
-  await updateProviderConnection(connection.id, {
-    lastPingedResetAt: resetAt,
-    lastPingAt: new Date().toISOString(),
-    updatedAt: new Date().toISOString(),
-  });
-  console.log(`[AutoPing] ${connection.id}: ping ${ok ? "sent" : "failed"} (reset ${resetAt})`);
-}
-
-async function tick() {
-  if (g.running) return;
-  g.running = true;
-  try {
-    const settings = await getSettings();
-    const enabledMap = settings[C.settingsKey]?.connections || {};
-    if (Object.keys(enabledMap).length === 0) return;
-
-    const conns = await getProviderConnections({ provider: "claude", isActive: true });
-    // Only ping connections the user explicitly enabled
-    const targets = conns.filter((c) => c.authType === "oauth" && enabledMap[c.id] === true);
-    if (targets.length === 0) return;
-
-    for (const conn of targets) {
-      try {
-        await pingConnection(conn);
-      } catch (e) {
-        console.warn(`[AutoPing] ${conn.id}: ${e.message}`);
-      }
-    }
-  } catch (e) {
-    console.warn("[AutoPing] tick error:", e.message);
-  } finally {
-    g.running = false;
-  }
-}
-
-export function startClaudeAutoPing() {
-  if (g.interval) return;
-  g.interval = setInterval(() => { tick().catch(() => {}); }, C.tickIntervalMs);
-  if (g.interval.unref) g.interval.unref();
-}
@@ -14,7 +14,7 @@ import {
  WATCHDOG_INTERVAL_MS, NETWORK_CHECK_INTERVAL_MS, VIRTUAL_IFACE_REGEX,
 } from "@/lib/tunnel";
 import { getMitmStatus, startMitm, loadEncryptedPassword, initDbHooks, restoreToolDNS, removeAllDNSEntriesSync } from "@/mitm/manager";
-import { startClaudeAutoPing } from "@/shared/services/claudeAutoPing";
+import { startQuotaAutoPing } from "@/shared/services/quotaAutoPing";
 import { syncToJson as syncMitmAliasCache } from "@/lib/mitmAliasCache";

 // Inject correct paths and DB hooks into manager.js (CJS) from ESM context
@@ -89,7 +89,7 @@ export async function initializeApp() {
    startWatchdog();
    startNetworkMonitor();
    autoStartMitm();
-    startClaudeAutoPing();
+    startQuotaAutoPing();
  } catch (error) {
    console.error("[InitApp] Error:", error);
  }
@@ -0,0 +1,298 @@
+// Quota auto-ping scheduler: warms 5h windows by sending tiny opt-in requests right after reset.
+import "open-sse/index.js";
+
+import { getSettings, getProviderConnections, updateProviderConnection } from "@/lib/localDb";
+import { getClaudeUsage } from "open-sse/services/usage/claude.js";
+import { getCodexUsage } from "open-sse/services/usage/codex.js";
+import { getExecutor } from "open-sse/executors/index.js";
+import { CLAUDE_CLI_SPOOF_HEADERS } from "open-sse/providers/shared.js";
+import { proxyAwareFetch } from "open-sse/utils/proxyFetch.js";
+import { resolveConnectionProxyConfig } from "@/lib/network/connectionProxy";
+import { refreshAndUpdateCredentials } from "@/app/api/usage/[connectionId]/route.js";
+import { QUOTA_AUTOPING_CONFIG } from "@/shared/constants/config";
+
+const C = QUOTA_AUTOPING_CONFIG;
+const CLAUDE_PING_URL = "https://api.anthropic.com/v1/messages?beta=true";
+
+const providerHandlers = {
+  claude: {
+    getUsage: getClaudeUsage,
+    sendPing: sendClaudePing,
+  },
+  codex: {
+    getUsage: getCodexUsage,
+    sendPing: sendCodexPing,
+  },
+};
+
+// Survive Next.js hot reload and keep one scheduler per server process.
+const g = (global.__quotaAutoPing ??= {
+  interval: null,
+  running: false,
+  resetCache: {},
+  failureCache: {},
+});
+
+function cacheKey(provider, connectionId) {
+  return `${provider}:${connectionId}`;
+}
+
+function normalizeResetKey(resetAt) {
+  const ms = new Date(resetAt).getTime();
+  if (!Number.isFinite(ms)) return resetAt;
+  return new Date(Math.floor(ms / 60000) * 60000).toISOString();
+}
+
+function getResetDriftMs(previousResetAt, nextResetAt) {
+  const previousMs = new Date(previousResetAt).getTime();
+  const nextMs = new Date(nextResetAt).getTime();
+  if (!Number.isFinite(previousMs) || !Number.isFinite(nextMs)) return 0;
+  return nextMs - previousMs;
+}
+
+function toFiniteNumber(value, fallback = null) {
+  if (typeof value === "number" && Number.isFinite(value)) return value;
+  if (typeof value === "string" && value.trim()) {
+    const parsed = Number(value);
+    if (Number.isFinite(parsed)) return parsed;
+  }
+  return fallback;
+}
+
+function isQuotaExhausted(quota) {
+  if (!quota || quota.unlimited === true) return false;
+  const remaining = toFiniteNumber(quota.remaining);
+  if (remaining !== null) return remaining <= 0;
+
+  const used = toFiniteNumber(quota.used);
+  const total = toFiniteNumber(quota.total);
+  return total !== null && total > 0 && used !== null && used >= total;
+}
+
+function wasPingedRecently(connection, intervalMs, nowMs = Date.now()) {
+  if (!intervalMs) return false;
+  const lastPingAtMs = new Date(connection.lastPingAt).getTime();
+  return Number.isFinite(lastPingAtMs) && nowMs - lastPingAtMs < intervalMs;
+}
+
+function isBlockingQuotaName(name, sessionKey) {
+  if (name === sessionKey) return false;
+  return !String(name).toLowerCase().includes("session");
+}
+
+function hasExhaustedBlockingQuota(quotas, sessionKey) {
+  return Object.entries(quotas || {}).some(([name, quota]) => isBlockingQuotaName(name, sessionKey) && isQuotaExhausted(quota));
+}
+
+function shouldPingForReset(providerConfig, cachedReset, resetAt, now) {
+  if (providerConfig.pingWhenResetAtSlides) {
+    return Boolean(cachedReset) && getResetDriftMs(cachedReset, resetAt) >= (providerConfig.resetAtDriftMs || 0);
+  }
+
+  const resetMs = new Date(resetAt).getTime();
+  return Number.isFinite(resetMs) && now >= resetMs - C.pingLeadMs;
+}
+
+function buildProxyOptions(cfg) {
+  return {
+    connectionProxyEnabled: cfg.connectionProxyEnabled === true,
+    connectionProxyUrl: cfg.connectionProxyUrl || "",
+    connectionNoProxy: cfg.connectionNoProxy || "",
+    vercelRelayUrl: cfg.vercelRelayUrl || "",
+    strictProxy: false,
+  };
+}
+
+async function sendClaudePing(connection, providerConfig, proxyOptions, deps) {
+  const res = await deps.proxyAwareFetch(CLAUDE_PING_URL, {
+    method: "POST",
+    headers: {
+      ...CLAUDE_CLI_SPOOF_HEADERS,
+      "Authorization": `Bearer ${connection.accessToken}`,
+      "content-type": "application/json",
+    },
+    body: JSON.stringify({
+      model: providerConfig.pingModel,
+      max_tokens: providerConfig.pingMaxTokens,
+      messages: [{ role: "user", content: providerConfig.pingText }],
+    }),
+  }, proxyOptions);
+  return res.ok;
+}
+
+function buildCodexPingInput(text) {
+  return [{
+    type: "message",
+    role: "user",
+    content: [{ type: "input_text", text }],
+  }];
+}
+
+async function drainResponseBody(response) {
+  if (typeof response?.text === "function") {
+    await response.text();
+    return;
+  }
+
+  const reader = response?.body?.getReader?.();
+  if (!reader) return;
+
+  try {
+    while (true) {
+      const { done } = await reader.read();
+      if (done) return;
+    }
+  } finally {
+    reader.releaseLock?.();
+  }
+}
+
+async function sendCodexPing(connection, providerConfig, proxyOptions, deps) {
+  const executor = deps.getExecutor("codex");
+  const { response } = await executor.execute({
+    model: providerConfig.pingModel,
+    stream: true,
+    credentials: {
+      accessToken: connection.accessToken,
+      connectionId: connection.id,
+      providerSpecificData: connection.providerSpecificData,
+    },
+    proxyOptions,
+    log: console,
+    body: {
+      model: providerConfig.pingModel,
+      input: buildCodexPingInput(providerConfig.pingText),
+      instructions: providerConfig.pingInstructions,
+      reasoning: providerConfig.pingReasoningEffort
+        ? { effort: providerConfig.pingReasoningEffort, summary: "auto" }
+        : undefined,
+      store: false,
+      stream: true,
+    },
+  });
+  if (!response.ok) {
+    try { await response.body?.cancel?.(); } catch { /* noop */ }
+    return false;
+  }
+
+  // Codex only starts the 5h window after the streaming response completes.
+  await drainResponseBody(response);
+  return true;
+}
+
+function shouldSkipAfterFailure(state, key, nowMs = Date.now()) {
+  const failedAt = state.failureCache[key];
+  return failedAt && nowMs - failedAt < C.failureCooldownMs;
+}
+
+async function pingConnection(conn, provider, providerConfig, handler, deps, state = g) {
+  const key = cacheKey(provider, conn.id);
+
+  // resetAt is stable for time-based windows; Codex polls every tick because inactive windows slide forward.
+  const cachedReset = state.resetCache[key];
+  if (!providerConfig.pingWhenResetAtSlides && cachedReset && Date.now() < new Date(cachedReset).getTime() - C.refreshAheadMs) return;
+
+  // Avoid hammering provider auth/quota endpoints if a ping failed recently.
+  if (shouldSkipAfterFailure(state, key)) return;
+
+  const proxyCfg = await deps.resolveConnectionProxyConfig(conn.providerSpecificData);
+  const proxyOptions = buildProxyOptions(proxyCfg);
+
+  let connection = conn;
+  try {
+    const r = await deps.refreshAndUpdateCredentials(connection, false, proxyOptions);
+    connection = r.connection;
+  } catch (e) {
+    state.failureCache[key] = Date.now();
+    console.warn(`[AutoPing] ${provider}:${conn.id}: refresh failed: ${e.message}`);
+    return;
+  }
+
+  const usage = await handler.getUsage(connection.accessToken, proxyOptions);
+  const quotas = usage?.quotas || {};
+  const quota = quotas?.[providerConfig.quotaKey];
+  const resetAt = quota?.resetAt;
+  if (!resetAt) return;
+
+  state.resetCache[key] = resetAt;
+
+  if (providerConfig.skipWhenBlockingQuotaExhausted && hasExhaustedBlockingQuota(quotas, providerConfig.quotaKey)) return;
+  if (isQuotaExhausted(quota)) return;
+
+  const now = Date.now();
+  const resetKey = normalizeResetKey(resetAt);
+  const lastPingedResetKey = connection.lastPingedResetKey || normalizeResetKey(connection.lastPingedResetAt);
+
+  // Claude waits for reset. Codex pings only when resetAt slides, which means the 5h window is inactive.
+  if (!shouldPingForReset(providerConfig, cachedReset, resetAt, now)) return;
+  if (wasPingedRecently(connection, providerConfig.minPingIntervalMs, now)) return;
+  if (lastPingedResetKey === resetKey) return;
+
+  const ok = await handler.sendPing(connection, providerConfig, proxyOptions, deps);
+  if (!ok) {
+    // Do not mark reset as pinged unless upstream accepted the tiny request.
+    state.failureCache[key] = Date.now();
+    console.warn(`[AutoPing] ${provider}:${connection.id}: ping failed (reset ${resetAt})`);
+    return;
+  }
+
+  delete state.failureCache[key];
+  await deps.updateProviderConnection(connection.id, {
+    lastPingedResetAt: resetAt,
+    lastPingedResetKey: resetKey,
+    lastPingAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+  });
+  console.log(`[AutoPing] ${provider}:${connection.id}: ping sent (reset ${resetAt})`);
+}
+
+function createDefaultDeps() {
+  return {
+    getSettings,
+    getProviderConnections,
+    updateProviderConnection,
+    resolveConnectionProxyConfig,
+    refreshAndUpdateCredentials,
+    proxyAwareFetch,
+    getExecutor,
+  };
+}
+
+export async function runQuotaAutoPingTick(deps = createDefaultDeps(), state = g) {
+  if (state.running) return;
+  state.running = true;
+  try {
+    const settings = await deps.getSettings();
+
+    for (const [provider, providerConfig] of Object.entries(C.providers)) {
+      const handler = providerHandlers[provider];
+      if (!handler) continue;
+
+      const enabledMap = settings?.[providerConfig.settingsKey]?.connections || {};
+      if (Object.keys(enabledMap).length === 0) continue;
+
+      const conns = await deps.getProviderConnections({ provider, isActive: true });
+      const targets = conns.filter((conn) => conn.authType === "oauth" && enabledMap[conn.id] === true);
+      for (const conn of targets) {
+        try {
+          await pingConnection(conn, provider, providerConfig, handler, deps, state);
+        } catch (e) {
+          state.failureCache[cacheKey(provider, conn.id)] = Date.now();
+          console.warn(`[AutoPing] ${provider}:${conn.id}: ${e.message}`);
+        }
+      }
+    }
+  } catch (e) {
+    console.warn("[AutoPing] tick error:", e.message);
+  } finally {
+    state.running = false;
+  }
+}
+
+export function startQuotaAutoPing() {
+  if (g.interval) return;
+  console.log("[AutoPing] scheduler started");
+  runQuotaAutoPingTick().catch(() => {});
+  g.interval = setInterval(() => { runQuotaAutoPingTick().catch(() => {}); }, C.tickIntervalMs);
+  if (g.interval.unref) g.interval.unref();
+}