ShockVPN/shockbot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix proxy key usage tracking and add OSS spend reporting (#500)

Browse Source 
* 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
This commit is contained in:
Colin McDonnell
2026-03-31 02:18:08 +00:00
committed by pullfrog[bot]
parent 6c5d228c04
commit 9f566d20e4
3 changed files with 38 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/opentoad.ts
+1 -1
View File
@@ -502,7 +502,7 @@ async function runOpenCode(params: RunParams): Promise<AgentResult> {
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();
entry
+18 -16
View File
@@ -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();
utils/subprocess.ts
+19 -16
View File
@@ -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<SpawnResult> {
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<SpawnResult> {
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<SpawnResult> {
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<SpawnResult> {
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<SpawnResult> {
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<SpawnResult> {
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<SpawnResult> {
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<SpawnResult> {
});
});
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();
}
});