fix: scope quota reports to authorized subscription accounts

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip committed 2026-10-06 11:28:22 -07:00
1 parent a85fa3af31
commit e85b02f99e
24 files changed
+1983 -182

No files matched your search

@@ -265,14 +265,15 @@ export async function fetchWithTimeout(url: string, init: RequestInit, ms = 8000
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), ms);
try {
return await fetch(url, { ...init, signal: controller.signal });
return await fetch(url, { ...init, signal: init.signal ? AbortSignal.any([init.signal, controller.signal]) : controller.signal });
} finally {
clearTimeout(timer);
}
}
export async function fetchClaudeQuota(token: string): Promise<QuotaWindow[]> {
export async function fetchClaudeQuota(token: string, signal?: AbortSignal): Promise<QuotaWindow[]> {
const resp = await fetchWithTimeout("https://api.anthropic.com/api/oauth/usage", {
signal,
headers: {
Authorization: `Bearer ${token}`,
"anthropic-beta": "oauth-2025-04-20",
@@ -463,13 +463,15 @@ describe("codex auth cache store", () => {
releaseFirstCaller = resolve;
});
const entered = deferredSignal();
const firstCall = withAccountHomeSecretMutationLock(env, "company-shared", async () => {
events.push("first-enter");
entered.resolve();
await firstCallerGate;
events.push("first-exit");
return "first";
});
await new Promise((resolve) => setTimeout(resolve, 20));
await entered.promise;
const secondCall = withAccountHomeSecretMutationLock(env, "company-shared", async () => {
events.push("second-enter");
events.push("second-exit");
@@ -497,12 +499,14 @@ describe("codex auth cache store", () => {
releaseCompanyA = resolve;
});
const entered = deferredSignal();
const companyACall = withAccountHomeSecretMutationLock(env, "company-a", async () => {
events.push("a-enter");
entered.resolve();
await companyAGate;
events.push("a-exit");
});
await new Promise((resolve) => setTimeout(resolve, 20));
await entered.promise;
const companyBCall = withAccountHomeSecretMutationLock(env, "company-b", async () => {
events.push("b-enter");
events.push("b-exit");
@@ -536,6 +540,28 @@ describe("codex auth cache store", () => {
expect(events).toEqual(["promotion-enter", "mutation-enter", "mutation-exit", "promotion-exit"]);
});
it("allows awaited nested writes but expires inherited ownership after release", async () => {
const env = envFor(await makeInstanceRoot());
const detachedGate = deferredSignal();
let detached!: Promise<void>;
let enteredDetached = false;
await withAccountHomeSecretMutationLock(env, "nested", async () => {
await expect(withAccountHomeSecretMutationLock(env, "nested", async () => "nested write")).resolves.toBe("nested write");
detached = detachedGate.promise.then(() => withAccountHomeSecretMutationLock(env, "nested", async () => { enteredDetached = true; }));
});
const holding = deferredSignal();
const release = deferredSignal();
const other = withAccountHomeSecretMutationLock(env, "nested", async () => { holding.resolve(); await release.promise; });
await holding.promise;
detachedGate.resolve();
try {
await new Promise(resolve => setTimeout(resolve, 30));
expect(enteredDetached).toBe(false);
} finally { release.resolve(); }
await Promise.all([other, detached]);
expect(enteredDetached).toBe(true);
});
});
describe("Phase 8: account-home directory validity before a secret write commits", () => {
@@ -581,3 +607,9 @@ describe("codex auth cache store", () => {
});
});
});
function deferredSignal() {
let resolve!: () => void;
const promise = new Promise<void>(done => { resolve = done; });
return { promise, resolve };
}
@@ -1,4 +1,5 @@
import { lstat, mkdir, open, readFile, rename, rm } from "node:fs/promises";
import { AsyncLocalStorage } from "node:async_hooks";
import { lstat, mkdir, open, readFile, realpath, rename, rm } from "node:fs/promises";
import path from "node:path";
import { randomUUID } from "node:crypto";
import { resolvePaperclipInstanceRootForAdapter } from "@paperclipai/adapter-utils/server-utils";
@@ -334,6 +335,11 @@ export async function withCodexAccountHomePromotionLock<T>(
* create) can still call into a secrets-service write that takes this lock
* without deadlocking on itself.
*/
// An outer credential transaction acquires this lock before database row locks.
// Nested secret-service writes must share that ownership, not acquire it again.
// Expire ownership on return so detached work cannot inherit a released lock.
const secretMutationOwnership = new AsyncLocalStorage<Map<string, { active: boolean }>>();
export async function withAccountHomeSecretMutationLock<T>(
env: NodeJS.ProcessEnv = process.env,
companyId: string,
@@ -341,7 +347,16 @@ export async function withAccountHomeSecretMutationLock<T>(
): Promise<T> {
const lockDir = resolveCodexAuthCacheNamedLockDir(env, companyId, "codex-account-home-secret-mutation-lock");
await ensurePrivateDir(lockDir);
return withDirectoryMergeLock(lockDir, fn, env);
const key = await realpath(lockDir);
const inherited = secretMutationOwnership.getStore();
if (inherited?.get(key)?.active) return fn();
return withDirectoryMergeLock(lockDir, async () => {
const ownership = { active: true };
const scope = new Map(inherited);
scope.set(key, ownership);
try { return await secretMutationOwnership.run(scope, fn); }
finally { ownership.active = false; }
}, env);
}
/**
@@ -17,7 +17,7 @@ vi.mock("node:child_process", async (importOriginal) => {
};
});
import { fetchCodexQuota, getQuotaWindows } from "./quota.js";
import { fetchCodexQuota, fetchCodexRpcQuota, getQuotaWindows } from "./quota.js";
function createChildThatErrorsOnMicrotask(err: Error): ChildProcess {
const child = new EventEmitter() as ChildProcess;
@@ -27,7 +27,7 @@ function createChildThatErrorsOnMicrotask(err: Error): ChildProcess {
Object.assign(child, {
stdout: stream,
stderr: Object.assign(new EventEmitter(), { setEncoding: () => {} }),
stdin: { write: vi.fn(), end: vi.fn() },
stdin: Object.assign(new EventEmitter(), { write: vi.fn(), end: vi.fn() }),
kill: vi.fn(),
});
queueMicrotask(() => {
@@ -36,6 +36,31 @@ function createChildThatErrorsOnMicrotask(err: Error): ChildProcess {
return child;
}
type RpcRequest = { id?: number; method: string };
function createRpcChild(response?: (request: RpcRequest) => unknown) {
const child = new EventEmitter() as ChildProcess;
const stdout = Object.assign(new EventEmitter(), { setEncoding: () => {} });
const stderr = Object.assign(new EventEmitter(), { setEncoding: () => {} });
const stdin = Object.assign(new EventEmitter(), {
write: vi.fn((line: string) => {
const request = JSON.parse(line) as RpcRequest;
if (request.id == null) return;
const message = response ? response(request) : { id: request.id, result: {} };
if (message !== undefined) {
queueMicrotask(() => stdout.emit("data", JSON.stringify(message) + "\n"));
}
}),
end: vi.fn(),
});
const kill = vi.fn((_signal?: string | number) => {
queueMicrotask(() => child.emit("close", 0));
return true;
});
Object.assign(child, { pid: 12345, stdout, stderr, stdin, kill });
return { child, stdout, stderr, stdin, kill };
}
describe("CodexRpcClient spawn failures", () => {
let previousCodexHome: string | undefined;
let isolatedCodexHome: string | undefined;
@@ -51,6 +76,7 @@ describe("CodexRpcClient spawn failures", () => {
});
afterEach(() => {
vi.useRealTimers();
vi.unstubAllGlobals();
if (isolatedCodexHome) {
try {
@@ -67,6 +93,159 @@ describe("CodexRpcClient spawn failures", () => {
}
});
it("reads account quota with a supported approval policy and never starts a model turn", async () => {
const { child, stdin, kill } = createRpcChild((request) => {
const result = request.method === "account/rateLimits/read"
? { rateLimits: { primary: { usedPercent: 37, windowDurationMins: 300 } } }
: {};
return { id: request.id, result };
});
mockSpawn.mockImplementation((_command, args: string[]) => {
if (args.includes("untrusted")) return createChildThatErrorsOnMicrotask(new Error("invalid approval policy"));
return child;
});
const result = await fetchCodexRpcQuota();
expect(result.windows).toEqual([expect.objectContaining({ usedPercent: 37 })]);
expect(mockSpawn).toHaveBeenCalledWith("codex", ["-s", "read-only", "-a", "on-request", "app-server"], expect.any(Object));
expect(stdin.write.mock.calls.map(([line]) => JSON.parse(line).method)).toEqual([
"initialize", "initialized", "account/rateLimits/read", "account/read",
]);
expect(stdin.end).toHaveBeenCalledOnce();
expect(kill).toHaveBeenCalledWith("SIGTERM");
expect(kill).not.toHaveBeenCalledWith("SIGKILL");
});
it("rejects RPC error envelopes and preserves their authentication classification", async () => {
const { child, kill } = createRpcChild((request) => request.method === "account/rateLimits/read"
? { id: request.id, error: { code: -32000, message: "OAuth failed: refresh token has expired" } }
: { id: request.id, result: {} });
mockSpawn.mockReturnValue(child);
const result = await getQuotaWindows();
expect(result).toMatchObject({ ok: false, source: "codex-rpc", errorFamily: "refresh_token_expired", windows: [] });
expect(kill).toHaveBeenCalledWith("SIGTERM");
});
it.each([null, [], "invalid", 17, undefined])("rejects an invalid result envelope (%s)", async (result) => {
const { child } = createRpcChild((request) => ({ id: request.id, result }));
mockSpawn.mockReturnValue(child);
await expect(fetchCodexRpcQuota()).rejects.toThrow("invalid quota response");
});
it("tolerates non-object lines and parses fragmented, out-of-order replies", async () => {
const probe = createRpcChild();
mockSpawn.mockReturnValue(probe.child);
probe.stdin.write.mockImplementation((line) => {
const request = JSON.parse(line) as RpcRequest;
if (request.method === "initialize") {
queueMicrotask(() => probe.stdout.emit("data", 'null\n[]\nnot-json\n{"id":1,"result":{}}\n'));
} else if (request.method === "account/read") {
queueMicrotask(() => {
probe.stdout.emit("data", '{"id":3,"result":{"account":{"email":"test@example.com"}}}\n{"id":2,');
probe.stdout.emit("data", '"result":{"rateLimits":{"primary":{"usedPercent":37}}}}\n');
});
}
});
const result = await fetchCodexRpcQuota();
expect(result.email).toBe("test@example.com");
expect(result.windows[0].usedPercent).toBe(37);
});
it.each([false, true])("bounds stdout buffering with and without a completed frame (newline: %s)", async (newline) => {
const probe = createRpcChild(() => undefined);
mockSpawn.mockReturnValue(probe.child);
const result = fetchCodexRpcQuota();
probe.stdout.emit("data", "x".repeat(64 * 1024));
probe.stdout.emit("data", "x" + (newline ? "\n" : ""));
await expect(result).rejects.toThrow("response exceeded the quota probe limit");
expect(probe.kill).toHaveBeenCalledWith("SIGTERM");
});
it("bounds stderr diagnostics when the process exits", async () => {
const probe = createRpcChild(() => undefined);
mockSpawn.mockReturnValue(probe.child);
const result = fetchCodexRpcQuota().catch((error: Error) => error);
probe.stderr.emit("data", "x".repeat(128 * 1024));
probe.stderr.emit("data", " end of diagnostic");
probe.child.emit("exit", 1);
const error = await result as Error;
expect(error.message.length).toBeLessThanOrEqual(4_000);
expect(error.message).toMatch(/end of diagnostic$/);
});
it("handles a broken stdin without an uncaught stream error", async () => {
const probe = createRpcChild(() => undefined);
mockSpawn.mockReturnValue(probe.child);
const result = fetchCodexRpcQuota();
probe.stdin.emit("error", new Error("write EPIPE"));
await expect(result).rejects.toThrow("write EPIPE");
expect(probe.kill).toHaveBeenCalledWith("SIGTERM");
});
it("clears other pending requests when a probe times out and kills a child that ignores SIGTERM", async () => {
vi.useFakeTimers();
const probe = createRpcChild((request) => request.method === "initialize"
? { id: request.id, result: {} }
: undefined);
probe.kill.mockImplementation((signal) => {
if (signal === "SIGKILL") queueMicrotask(() => probe.child.emit("close", null));
return true;
});
mockSpawn.mockReturnValue(probe.child);
const result = fetchCodexRpcQuota();
const rejected = expect(result).rejects.toThrow("timed out on account/rateLimits/read");
await vi.advanceTimersByTimeAsync(6_000);
expect(probe.stdin.end).toHaveBeenCalledOnce();
expect(probe.kill.mock.calls).toEqual([["SIGTERM"]]);
await vi.advanceTimersByTimeAsync(250);
await rejected;
expect(probe.kill.mock.calls).toEqual([["SIGTERM"], ["SIGKILL"]]);
expect(vi.getTimerCount()).toBe(0);
expect(probe.child.listenerCount("close")).toBe(1);
});
it("bounds shutdown even if a killed child never emits close", async () => {
vi.useFakeTimers();
const probe = createRpcChild();
probe.kill.mockImplementation(() => true);
mockSpawn.mockReturnValue(probe.child);
const result = fetchCodexRpcQuota();
await vi.advanceTimersByTimeAsync(500);
await expect(result).resolves.toMatchObject({ windows: [] });
expect(probe.kill.mock.calls).toEqual([["SIGTERM"], ["SIGKILL"]]);
expect(vi.getTimerCount()).toBe(0);
expect(probe.child.listenerCount("close")).toBe(1);
});
it("reaps a real subprocess that ignores SIGTERM", async () => {
const { spawn: spawnReal } = await vi.importActual<typeof import("node:child_process")>("node:child_process");
const child = spawnReal(process.execPath, ["--input-type=commonjs", "-e", `
process.on("SIGTERM", () => {});
setInterval(() => {}, 60_000);
require("node:readline").createInterface({ input: process.stdin }).on("line", (line) => {
const request = JSON.parse(line);
if (request.id == null) return;
const result = request.method === "account/rateLimits/read"
? { rateLimits: { primary: { usedPercent: 37 } } }
: {};
process.stdout.write(JSON.stringify({ id: request.id, result }) + "\\n");
});
`], { stdio: ["pipe", "pipe", "pipe"] });
mockSpawn.mockReturnValue(child);
try {
const result = await fetchCodexRpcQuota();
expect(result.windows[0].usedPercent).toBe(37);
expect(child.signalCode).toBe("SIGKILL");
expect(child.stdin.destroyed).toBe(true);
expect(child.stdout.destroyed).toBe(true);
expect(child.stderr.destroyed).toBe(true);
} finally {
if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
}
});
it("classifies app-server refresh-token failures as quota probe auth errors", async () => {
mockSpawn.mockImplementation(() => createChildThatErrorsOnMicrotask(new Error("OAuth failed: refresh token has expired")));
@@ -110,7 +289,7 @@ describe("CodexRpcClient spawn failures", () => {
expect(result.windows).toEqual([
expect.objectContaining({
label: "5h limit",
usedPercent: 50,
usedPercent: 0.5,
resetsAt: "2024-03-22T12:38:31.000Z",
}),
]);
@@ -11,6 +11,8 @@ import {
const CODEX_USAGE_SOURCE_RPC = "codex-rpc";
const CODEX_USAGE_SOURCE_WHAM = "codex-wham";
const MAX_QUOTA_ERROR_BODY_BYTES = 4_000;
const MAX_RPC_FRAME_BYTES = 64 * 1024;
const RPC_SHUTDOWN_GRACE_MS = 250;
export function codexHomeDir(): string {
const fromEnv = process.env.CODEX_HOME;
@@ -227,7 +229,7 @@ export async function fetchWithTimeout(
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), ms);
try {
return await fetch(url, { ...init, signal: controller.signal });
return await fetch(url, { ...init, signal: init.signal ? AbortSignal.any([init.signal, controller.signal]) : controller.signal });
} finally {
clearTimeout(timer);
}
@@ -272,20 +274,22 @@ async function readResponseTextPrefix(
}
function normalizeCodexUsedPercent(rawPct: number | null | undefined): number | null {
if (rawPct == null) return null;
return Math.min(100, Math.round(rawPct < 1 ? rawPct * 100 : rawPct));
if (typeof rawPct !== "number" || !Number.isFinite(rawPct)) return null;
// Both RPC and WHAM report percentage points, including values below 1%.
return Math.max(0, Math.min(100, rawPct));
}
export async function fetchCodexQuota(
token: string,
accountId: string | null,
signal?: AbortSignal,
): Promise<QuotaWindow[]> {
const headers: Record<string, string> = {
Authorization: `Bearer ${token}`,
};
if (accountId) headers["ChatGPT-Account-Id"] = accountId;
const resp = await fetchWithTimeout("https://chatgpt.com/backend-api/wham/usage", { headers });
const resp = await fetchWithTimeout("https://chatgpt.com/backend-api/wham/usage", { headers, signal });
if (!resp.ok) {
const message = `chatgpt wham api returned ${resp.status}`;
const responseText = await readResponseTextPrefix(resp);
@@ -470,7 +474,9 @@ type PendingRequest = {
class CodexRpcClient {
private proc = spawn(
"codex",
["-s", "read-only", "-a", "untrusted", "app-server"],
// Only account-read RPCs are sent; no agent turn is started. Keep the
// read-only sandbox and use the approval policy supported by current CLIs.
["-s", "read-only", "-a", "on-request", "app-server"],
{ stdio: ["pipe", "pipe", "pipe"], env: process.env },
);
@@ -478,69 +484,101 @@ class CodexRpcClient {
private buffer = "";
private pending = new Map<number, PendingRequest>();
private stderr = "";
private failure: Error | null = null;
private closed = false;
constructor() {
this.proc.stdout.setEncoding("utf8");
this.proc.stderr.setEncoding("utf8");
this.proc.stdout.on("data", (chunk: string) => this.onStdout(chunk));
this.proc.stderr.on("data", (chunk: string) => {
this.stderr += chunk;
this.stderr = (this.stderr + chunk).slice(-MAX_QUOTA_ERROR_BODY_BYTES);
});
this.proc.on("exit", () => {
for (const request of this.pending.values()) {
clearTimeout(request.timer);
request.reject(new Error(this.stderr.trim() || "codex app-server closed unexpectedly"));
}
this.pending.clear();
this.fail(new Error(this.stderr.trim() || "codex app-server closed unexpectedly"));
});
this.proc.on("error", (err: Error) => {
for (const request of this.pending.values()) {
clearTimeout(request.timer);
request.reject(err);
}
this.pending.clear();
this.proc.on("close", () => {
this.closed = true;
});
this.proc.on("error", (err: Error) => this.fail(err));
// A child may close its input between writing a request and exiting.
this.proc.stdin.on("error", (err: Error) => this.fail(err));
}
private fail(error: Error) {
this.failure ??= error;
this.buffer = "";
for (const request of this.pending.values()) {
clearTimeout(request.timer);
request.reject(this.failure);
}
this.pending.clear();
}
private onStdout(chunk: string) {
if (this.failure) return;
this.buffer += chunk;
while (true) {
const newlineIndex = this.buffer.indexOf("\n");
const frame = newlineIndex < 0 ? this.buffer : this.buffer.slice(0, newlineIndex);
if (Buffer.byteLength(frame, "utf8") > MAX_RPC_FRAME_BYTES) {
this.fail(new Error("codex app-server response exceeded the quota probe limit"));
return;
}
if (newlineIndex < 0) break;
const line = this.buffer.slice(0, newlineIndex).trim();
this.buffer = this.buffer.slice(newlineIndex + 1);
if (!line) continue;
let parsed: Record<string, unknown>;
let value: unknown;
try {
parsed = JSON.parse(line) as Record<string, unknown>;
value = JSON.parse(line);
} catch {
continue;
}
if (!value || typeof value !== "object" || Array.isArray(value)) continue;
const parsed = value as Record<string, unknown>;
const id = typeof parsed.id === "number" ? parsed.id : null;
if (id == null) continue;
const pending = this.pending.get(id);
if (!pending) continue;
this.pending.delete(id);
clearTimeout(pending.timer);
pending.resolve(parsed);
if (parsed.error != null) {
const error = parsed.error as { message?: unknown };
pending.reject(new Error(typeof error.message === "string"
? error.message.slice(0, MAX_QUOTA_ERROR_BODY_BYTES)
: "codex app-server rejected the quota request"));
} else if (!parsed.result || typeof parsed.result !== "object" || Array.isArray(parsed.result)) {
pending.reject(new Error("codex app-server returned an invalid quota response"));
} else {
pending.resolve(parsed);
}
}
}
private request(method: string, params: Record<string, unknown> = {}, timeoutMs = 6_000): Promise<Record<string, unknown>> {
if (this.failure) return Promise.reject(this.failure);
const id = this.nextId++;
const payload = JSON.stringify({ id, method, params }) + "\n";
return new Promise<Record<string, unknown>>((resolve, reject) => {
const timer = setTimeout(() => {
this.pending.delete(id);
reject(new Error(`codex app-server timed out on ${method}`));
this.fail(new Error(`codex app-server timed out on ${method}`));
}, timeoutMs);
this.pending.set(id, { resolve, reject, timer });
this.proc.stdin.write(payload);
this.write(payload);
});
}
private write(payload: string) {
try {
this.proc.stdin.write(payload);
} catch (error) {
this.fail(error instanceof Error ? error : new Error("codex app-server input failed"));
}
}
private notify(method: string, params: Record<string, unknown> = {}) {
this.proc.stdin.write(JSON.stringify({ method, params }) + "\n");
this.write(JSON.stringify({ method, params }) + "\n");
}
async initialize() {
@@ -568,7 +606,30 @@ class CodexRpcClient {
}
async shutdown() {
this.fail(new Error("codex app-server quota probe closed"));
this.proc.stdin.end();
// A failed spawn has no process to signal. Never leave a live probe behind
// when it ignores SIGTERM, including after a request or transport failure.
if (this.closed || this.proc.pid == null) return;
this.proc.kill("SIGTERM");
await this.waitForClose();
if (!this.closed) {
this.proc.kill("SIGKILL");
await this.waitForClose();
}
}
private async waitForClose(): Promise<void> {
if (this.closed) return;
await new Promise<void>((resolve) => {
const done = () => {
clearTimeout(timer);
this.proc.removeListener("close", done);
resolve();
};
const timer = setTimeout(done, RPC_SHUTDOWN_GRACE_MS);
this.proc.once("close", done);
});
}
}
+4
View File
@@ -16,6 +16,10 @@ export interface QuotaWindow {
export interface ProviderQuotaResult {
/** provider slug, e.g. "anthropic", "openai" */
provider: string;
/** Connection, grant and credential revision; never a credential value. */
accountKey?: string;
accountLabel?: string;
capturedAt?: string;
/** source label when the provider reports where the quota data came from */
source?: string | null;
/** true when the fetch succeeded and windows is populated */