From 9f566d20e45c2d7850d2781012df68c3b53fef26 Mon Sep 17 00:00:00 2001 From: Colin McDonnell Date: Tue, 31 Mar 2026 02:18:08 +0000 Subject: [PATCH] fix proxy key usage tracking and add OSS spend reporting (#500) * fix proxy key usage tracking and add OSS spend reporting replace ProxyKey.disabled with finalizedAt to fix a race where keys were disabled before their usage was synced, causing all HWM values to be zero. retireKey now fetches final usage from OpenRouter and records it atomically with optimistic concurrency. syncAccountUsage skips keys that fail to fetch rather than recording false zeros. other fixes: - wrap OpenRouter API calls in retry logic (exponential backoff) - reconcileStaleWorkflowRuns now retires proxy keys for completed runs - subprocess activity timeout only tracks stdout (stderr retry loops no longer prevent hung agent detection) - add oss-spend script (single bulk fetch from OpenRouter) and backfill-proxy-key-usage script Made-with: Cursor * address review: move isActiveKey check inside transaction, remove redundant guard Made-with: Cursor --- agents/opentoad.ts | 2 +- entry | 34 ++++++++++++++++++---------------- utils/subprocess.ts | 35 +++++++++++++++++++---------------- 3 files changed, 38 insertions(+), 33 deletions(-) diff --git a/agents/opentoad.ts b/agents/opentoad.ts index 8705c63..81a1346 100644 --- a/agents/opentoad.ts +++ b/agents/opentoad.ts @@ -502,7 +502,7 @@ async function runOpenCode(params: RunParams): Promise { args: params.args, cwd: params.cwd, env: params.env, - activityTimeout: 0, + activityTimeout: 300_000, stdio: ["ignore", "pipe", "pipe"], onStdout: async (chunk) => { const text = chunk.toString(); diff --git a/entry b/entry index 2f69493..80dd82a 100755 --- a/entry +++ b/entry @@ -145007,20 +145007,19 @@ function installSignalHandler() { }); } async function spawn(options) { - const { cmd, args: args2, env: env2, input, timeout, cwd, stdio, onStdout, onStderr } = options; const activityTimeoutMs = options.activityTimeout ?? DEFAULT_ACTIVITY_TIMEOUT_MS; installSignalHandler(); const startTime = performance3.now(); let stdoutBuffer = ""; let stderrBuffer = ""; return new Promise((resolve2, reject) => { - const child = nodeSpawn(cmd, args2, { - env: env2 || { + const child = nodeSpawn(options.cmd, options.args, { + env: options.env || { PATH: process.env.PATH || "", HOME: process.env.HOME || "" }, - stdio: stdio || ["pipe", "pipe", "pipe"], - cwd: cwd || process.cwd() + stdio: options.stdio || ["pipe", "pipe", "pipe"], + cwd: options.cwd || process.cwd() }); trackChild({ child }); let timeoutId; @@ -145028,7 +145027,7 @@ async function spawn(options) { let isTimedOut = false; let isActivityTimedOut = false; let lastActivityTime = performance3.now(); - if (timeout) { + if (options.timeout) { timeoutId = setTimeout(() => { isTimedOut = true; child.kill("SIGTERM"); @@ -145037,10 +145036,12 @@ async function spawn(options) { child.kill("SIGKILL"); } }, 5e3); - }, timeout); + }, options.timeout); } if (activityTimeoutMs > 0) { - log.debug(`spawn activity timer: pid=${child.pid} cmd=${cmd} timeout=${activityTimeoutMs}ms`); + log.debug( + `spawn activity timer: pid=${child.pid} cmd=${options.cmd} timeout=${activityTimeoutMs}ms` + ); activityCheckIntervalId = setInterval(() => { const idleMs = performance3.now() - lastActivityTime; log.debug( @@ -145049,7 +145050,9 @@ async function spawn(options) { if (idleMs > activityTimeoutMs) { isActivityTimedOut = true; const idleSec = Math.round(idleMs / 1e3); - log.info(`no output for ${idleSec}s from pid=${child.pid} (${cmd}), killing process`); + log.info( + `no output for ${idleSec}s from pid=${child.pid} (${options.cmd}), killing process` + ); child.kill("SIGKILL"); clearInterval(activityCheckIntervalId); } @@ -145063,15 +145066,14 @@ async function spawn(options) { updateActivity(); const chunk = data.toString(); stdoutBuffer += chunk; - onStdout?.(chunk); + options.onStdout?.(chunk); }); } if (child.stderr) { child.stderr.on("data", (data) => { - updateActivity(); const chunk = data.toString(); stderrBuffer += chunk; - onStderr?.(chunk); + options.onStderr?.(chunk); }); } child.on("close", (exitCode) => { @@ -145080,7 +145082,7 @@ async function spawn(options) { if (timeoutId) clearTimeout(timeoutId); if (activityCheckIntervalId) clearInterval(activityCheckIntervalId); if (isTimedOut) { - reject(new Error(`process timed out after ${timeout}ms`)); + reject(new Error(`process timed out after ${options.timeout}ms`)); return; } if (isActivityTimedOut) { @@ -145108,8 +145110,8 @@ async function spawn(options) { durationMs }); }); - if (input && child.stdin && stdio?.[0] !== "ignore") { - child.stdin.write(input); + if (options.input && child.stdin && options.stdio?.[0] !== "ignore") { + child.stdin.write(options.input); child.stdin.end(); } }); @@ -149649,7 +149651,7 @@ async function runOpenCode(params) { args: params.args, cwd: params.cwd, env: params.env, - activityTimeout: 0, + activityTimeout: 3e5, stdio: ["ignore", "pipe", "pipe"], onStdout: async (chunk) => { const text = chunk.toString(); diff --git a/utils/subprocess.ts b/utils/subprocess.ts index fc0d0a4..c547f97 100644 --- a/utils/subprocess.ts +++ b/utils/subprocess.ts @@ -76,7 +76,8 @@ export interface SpawnOptions { env?: NodeJS.ProcessEnv; input?: string; timeout?: number; - // activity timeout: kill process if no stdout/stderr for this many ms (default: 30s, 0 to disable) + // activity timeout: kill process if no stdout for this many ms (default: 30s, 0 to disable). + // only stdout resets the timer — stderr (e.g. provider error retries) does not count as progress. activityTimeout?: number; cwd?: string; stdio?: ("pipe" | "ignore" | "inherit")[]; @@ -95,7 +96,6 @@ export interface SpawnResult { * Spawn a subprocess with streaming callbacks and buffered results */ export async function spawn(options: SpawnOptions): Promise { - const { cmd, args, env, input, timeout, cwd, stdio, onStdout, onStderr } = options; const activityTimeoutMs = options.activityTimeout ?? DEFAULT_ACTIVITY_TIMEOUT_MS; installSignalHandler(); @@ -106,13 +106,13 @@ export async function spawn(options: SpawnOptions): Promise { return new Promise((resolve, reject) => { // security: caller must provide complete env object, not merged with process.env - const child = nodeSpawn(cmd, args, { - env: env || { + const child = nodeSpawn(options.cmd, options.args, { + env: options.env || { PATH: process.env.PATH || "", HOME: process.env.HOME || "", }, - stdio: stdio || ["pipe", "pipe", "pipe"], - cwd: cwd || process.cwd(), + stdio: options.stdio || ["pipe", "pipe", "pipe"], + cwd: options.cwd || process.cwd(), }); // track child for cleanup on Ctrl+C @@ -125,7 +125,7 @@ export async function spawn(options: SpawnOptions): Promise { let lastActivityTime = performance.now(); // overall timeout - if (timeout) { + if (options.timeout) { timeoutId = setTimeout(() => { isTimedOut = true; child.kill("SIGTERM"); @@ -135,12 +135,14 @@ export async function spawn(options: SpawnOptions): Promise { child.kill("SIGKILL"); } }, 5000); - }, timeout); + }, options.timeout); } // activity timeout: kill if no output for too long if (activityTimeoutMs > 0) { - log.debug(`spawn activity timer: pid=${child.pid} cmd=${cmd} timeout=${activityTimeoutMs}ms`); + log.debug( + `spawn activity timer: pid=${child.pid} cmd=${options.cmd} timeout=${activityTimeoutMs}ms` + ); activityCheckIntervalId = setInterval(() => { const idleMs = performance.now() - lastActivityTime; log.debug( @@ -149,7 +151,9 @@ export async function spawn(options: SpawnOptions): Promise { if (idleMs > activityTimeoutMs) { isActivityTimedOut = true; const idleSec = Math.round(idleMs / 1000); - log.info(`no output for ${idleSec}s from pid=${child.pid} (${cmd}), killing process`); + log.info( + `no output for ${idleSec}s from pid=${child.pid} (${options.cmd}), killing process` + ); child.kill("SIGKILL"); clearInterval(activityCheckIntervalId); } @@ -165,16 +169,15 @@ export async function spawn(options: SpawnOptions): Promise { updateActivity(); const chunk = data.toString(); stdoutBuffer += chunk; - onStdout?.(chunk); + options.onStdout?.(chunk); }); } if (child.stderr) { child.stderr.on("data", (data: Buffer) => { - updateActivity(); const chunk = data.toString(); stderrBuffer += chunk; - onStderr?.(chunk); + options.onStderr?.(chunk); }); } @@ -186,7 +189,7 @@ export async function spawn(options: SpawnOptions): Promise { if (activityCheckIntervalId) clearInterval(activityCheckIntervalId); if (isTimedOut) { - reject(new Error(`process timed out after ${timeout}ms`)); + reject(new Error(`process timed out after ${options.timeout}ms`)); return; } @@ -222,8 +225,8 @@ export async function spawn(options: SpawnOptions): Promise { }); }); - if (input && child.stdin && stdio?.[0] !== "ignore") { - child.stdin.write(input); + if (options.input && child.stdin && options.stdio?.[0] !== "ignore") { + child.stdin.write(options.input); child.stdin.end(); } });