mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix: carry quota review fixes into durable accounting
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
commit
5997745d77
7 files changed
+130
-15
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",
|
||||
|
||||
@@ -229,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);
|
||||
}
|
||||
@@ -282,13 +282,14 @@ function normalizeCodexUsedPercent(rawPct: number | null | undefined): number |
|
||||
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);
|
||||
|
||||
@@ -8,8 +8,9 @@ import { mkdtemp, rm, access, readFile, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { createDb, companies, agents, heartbeatRuns, companyMemberships, connectionGrants, connectionGrantDelegations, connectionGrantMembers, toolConnections, toolConnectionInstalls, aiConnectionDefaults, aiProviderDefaults, adapterAuthSessions, environments, issues, issueThreadInteractions, issueRecoveryActions, connectionIntentDeliveries, agentWakeupRequests, companySecrets, principalPermissionGrants } from "@paperclipai/db";
|
||||
import { createDb, companies, agents, heartbeatRuns, companyMemberships, connectionGrants, connectionGrantDelegations, connectionGrantMembers, toolConnections, toolConnectionInstalls, aiConnectionDefaults, aiProviderDefaults, adapterAuthSessions, environments, issues, issueThreadInteractions, issueRecoveryActions, connectionIntentDeliveries, agentWakeupRequests, companySecrets, companySecretVersions, principalPermissionGrants } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "@paperclipai/db/test-embedded-postgres";
|
||||
import { fetchCompanyQuotaWindows } from "../services/quota-windows.js";
|
||||
import { aiConnectionService } from "../services/ai-connections.js";
|
||||
import * as executionTarget from "@paperclipai/adapter-utils/execution-target";
|
||||
import { prepareManagedAiRuntime, assertManagedAiProjectAuth } from "../services/ai-connection-runtime.js";
|
||||
@@ -46,6 +47,36 @@ beforeAll(async () => {
|
||||
afterAll(async () => { await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); });
|
||||
|
||||
describe("managed AI connections", () => {
|
||||
it("caches quota after a real secret read but invalidates credential rotation and revocation", async () => {
|
||||
const owner = "quota-cache-owner";
|
||||
await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
|
||||
const account = await service.save(companyId, owner, { provider: "anthropic", method: "subscription", ownership: "personal", name: "Quota cache account", loginSessionId: "fixture", agentIds: [], allAgents: true }, "quota-cache-original");
|
||||
const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, account.grantId));
|
||||
const secretId = grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!.secretId;
|
||||
const [before] = await db.select().from(companySecrets).where(eq(companySecrets.id, secretId));
|
||||
const fetchSpy = vi.spyOn(globalThis, "fetch").mockImplementation(async () => new Response(JSON.stringify({ five_hour: { utilization: 42 } })));
|
||||
try {
|
||||
const first = await fetchCompanyQuotaWindows(db, companyId, owner);
|
||||
const [after] = await db.select().from(companySecrets).where(eq(companySecrets.id, secretId));
|
||||
expect(after.lastResolvedAt).not.toBeNull();
|
||||
expect(after.updatedAt.getTime()).toBeGreaterThan(before.updatedAt.getTime());
|
||||
expect(first[0].ok).toBe(true);
|
||||
expect(await fetchCompanyQuotaWindows(db, companyId, owner)).toEqual(first);
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(1);
|
||||
await secretService(db).rotate(secretId, { value: "quota-cache-rotated" });
|
||||
const rotated = await fetchCompanyQuotaWindows(db, companyId, owner);
|
||||
expect(rotated[0].ok).toBe(true);
|
||||
expect(rotated[0].accountKey).not.toBe(first[0].accountKey);
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(2);
|
||||
expect(fetchSpy).toHaveBeenLastCalledWith(expect.any(String), expect.objectContaining({ headers: expect.objectContaining({ Authorization: "Bearer quota-cache-rotated" }) }));
|
||||
await db.update(companySecretVersions).set({ status: "disabled", revokedAt: new Date() }).where(and(eq(companySecretVersions.secretId, secretId), eq(companySecretVersions.version, 2)));
|
||||
const revoked = await fetchCompanyQuotaWindows(db, companyId, owner);
|
||||
expect(revoked[0].ok).toBe(false);
|
||||
expect(revoked[0].windows).toEqual([]);
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(2);
|
||||
} finally { fetchSpy.mockRestore(); }
|
||||
});
|
||||
|
||||
it.each([false, true])("reports the authoritative connection-manager capability for custom grants (manager: %s)", async (manager) => {
|
||||
const userId = `custom-manager-${manager}`;
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" });
|
||||
|
||||
@@ -370,7 +370,11 @@ function listGuidanceFiles(rootDir = repoRoot): string[] {
|
||||
const found: string[] = [];
|
||||
// Include tracked files and new authored files, while respecting gitignored
|
||||
// local settings. A tracked file remains scanned even if an ignore matches.
|
||||
const authored = rootDir === repoRoot ? new Set(execFileSync("git", ["ls-files", "--cached", "--others", "--exclude-standard", "-z"], { cwd: rootDir, encoding: "utf8" }).split("\0")) : null;
|
||||
const authored = new Set(execFileSync(
|
||||
"git",
|
||||
["ls-files", "--cached", "--others", "--exclude-standard", "-z"],
|
||||
{ cwd: rootDir, encoding: "utf8" },
|
||||
).split("\0"));
|
||||
|
||||
function walk(absDir: string, relDir: string): void {
|
||||
for (const entry of readdirSync(absDir, { withFileTypes: true })) {
|
||||
@@ -386,7 +390,7 @@ function listGuidanceFiles(rootDir = repoRoot): string[] {
|
||||
if (!SCAN_EXTENSIONS.has(path.extname(entry.name))) continue;
|
||||
if (isTestFile(relPath)) continue;
|
||||
if (SKIP_PATH_PREFIXES.some((prefix) => relPath.startsWith(prefix))) continue;
|
||||
if (!authored || authored.has(relPath)) found.push(relPath);
|
||||
if (authored.has(relPath)) found.push(relPath);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -490,6 +494,7 @@ describe("paperclipai CLI invocation safety", () => {
|
||||
"tests/runner-e2e/catalog.ts",
|
||||
];
|
||||
try {
|
||||
execFileSync("git", ["init", "--quiet"], { cwd: root });
|
||||
for (const relPath of [
|
||||
...sourcePaths,
|
||||
"tests/runner-e2e/results/campaign/attempt-1/snapshots/api-state.json",
|
||||
@@ -507,6 +512,7 @@ describe("paperclipai CLI invocation safety", () => {
|
||||
it("excludes root runtime recordings but still scans unsafe docs and source guidance", () => {
|
||||
const fixtureRoot = mkdtempSync(path.join(os.tmpdir(), "paperclip-cli-guidance-"));
|
||||
try {
|
||||
execFileSync("git", ["init", "--quiet"], { cwd: fixtureRoot });
|
||||
const fixtures = [
|
||||
"doc/CLI.md",
|
||||
"src/guidance.ts",
|
||||
@@ -532,6 +538,26 @@ describe("paperclipai CLI invocation safety", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("scans tracked and ordinary untracked guidance but excludes ignored untracked files", () => {
|
||||
const root = mkdtempSync(path.join(os.tmpdir(), "paperclip-cli-guidance-"));
|
||||
try {
|
||||
execFileSync("git", ["init", "--quiet"], { cwd: root });
|
||||
writeFileSync(path.join(root, ".gitignore"), "ignored.md\ntracked.md\n");
|
||||
for (const name of ["ignored.md", "untracked.md", "tracked.md"]) {
|
||||
writeFileSync(path.join(root, name), "pnpm paperclipai issue get issue-1\n");
|
||||
}
|
||||
execFileSync("git", ["add", "--force", "tracked.md"], { cwd: root });
|
||||
|
||||
const files = listGuidanceFiles(root).sort();
|
||||
expect(files).toEqual(["tracked.md", "untracked.md"]);
|
||||
for (const name of files) {
|
||||
expect(scanText(name, readFileSync(path.join(root, name), "utf8"))).toHaveLength(1);
|
||||
}
|
||||
} finally {
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("allows only exact-allowlist pnpm paperclipai commands on every guidance surface", () => {
|
||||
const offenders = scanForOffenders();
|
||||
expect(
|
||||
|
||||
@@ -28,6 +28,7 @@ let revision = 1;
|
||||
const db = {
|
||||
select: () => ({
|
||||
from: () => ({
|
||||
leftJoin: () => ({
|
||||
where: async () => [
|
||||
{
|
||||
id: "secret",
|
||||
@@ -36,6 +37,7 @@ const db = {
|
||||
status: "active",
|
||||
},
|
||||
],
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
} as unknown as Db;
|
||||
@@ -66,7 +68,7 @@ describe("connected account quotas", () => {
|
||||
mocks.accounts.mockResolvedValue([account("remote-codex")]);
|
||||
const result = await fetchCompanyQuotaWindows(db, "company-1", "user-1");
|
||||
expect(mocks.accounts).toHaveBeenCalledWith("company-1", "user-1");
|
||||
expect(mocks.codex).toHaveBeenCalledWith("private-token", "actual-account");
|
||||
expect(mocks.codex).toHaveBeenCalledWith("private-token", "actual-account", expect.any(AbortSignal));
|
||||
expect(mocks.adapters).not.toHaveBeenCalled();
|
||||
expect(result[0]).toMatchObject({
|
||||
ok: true,
|
||||
@@ -125,7 +127,7 @@ describe("connected account quotas", () => {
|
||||
]);
|
||||
const [result] = await fetchCompanyQuotaWindows(db, "company-6", "user-6");
|
||||
expect(result.ok).toBe(true);
|
||||
expect(mocks.claude).toHaveBeenCalledWith("claude-private-oauth");
|
||||
expect(mocks.claude).toHaveBeenCalledWith("claude-private-oauth", expect.any(AbortSignal));
|
||||
mocks.accounts.mockResolvedValue([account("secret-failure")]);
|
||||
mocks.credential.mockRejectedValue(
|
||||
new Error("private credential store failure"),
|
||||
@@ -158,6 +160,33 @@ describe("connected account quotas", () => {
|
||||
expect(result.accountKey).toBeTruthy();
|
||||
expect(result.error).not.toContain("timed out");
|
||||
});
|
||||
it("does not start a provider request when a credential resolves after its deadline", async () => {
|
||||
vi.useFakeTimers();
|
||||
mocks.accounts.mockResolvedValue([account("late-credential")]);
|
||||
let resolveCredential!: (value: string) => void;
|
||||
mocks.credential.mockReturnValue(new Promise<string>((resolve) => { resolveCredential = resolve; }));
|
||||
const pending = fetchCompanyQuotaWindows(db, "late-company", "late-user");
|
||||
await vi.advanceTimersByTimeAsync(20_001);
|
||||
expect((await pending)[0].ok).toBe(false);
|
||||
resolveCredential(JSON.stringify({ accessToken: "private", accountId: "actual" }));
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(mocks.codex).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("cancels an active provider read when the account deadline expires", async () => {
|
||||
vi.useFakeTimers();
|
||||
mocks.accounts.mockResolvedValue([account("slow-provider")]);
|
||||
let signal!: AbortSignal;
|
||||
mocks.codex.mockImplementation((_token, _id, abortSignal) => {
|
||||
signal = abortSignal;
|
||||
return new Promise((_resolve, reject) => signal.addEventListener("abort", () => reject(signal.reason), { once: true }));
|
||||
});
|
||||
const pending = fetchCompanyQuotaWindows(db, "slow-company", "slow-user");
|
||||
await vi.advanceTimersByTimeAsync(20_001);
|
||||
expect((await pending)[0].ok).toBe(false);
|
||||
expect(signal.aborted).toBe(true);
|
||||
});
|
||||
|
||||
it("limits account probes to four concurrent requests and preserves account order", async () => {
|
||||
mocks.accounts.mockResolvedValue(
|
||||
Array.from({ length: 9 }, (_, i) => account(`parallel-${i}`)),
|
||||
|
||||
@@ -829,6 +829,29 @@ describe("mapCodexRpcQuota", () => {
|
||||
// fetchWithTimeout — abort on timeout
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
describe("managed quota cancellation", () => {
|
||||
afterEach(() => vi.unstubAllGlobals());
|
||||
|
||||
it.each([
|
||||
["Claude", (signal: AbortSignal) => fetchClaudeQuota("token", signal)],
|
||||
["Codex", (signal: AbortSignal) => fetchCodexQuota("token", "account", signal)],
|
||||
] as const)("forwards cancellation to the %s provider", async (_provider, read) => {
|
||||
const controller = new AbortController();
|
||||
let requestSignal!: AbortSignal;
|
||||
vi.stubGlobal("fetch", vi.fn(async (_url: string, init: RequestInit) => {
|
||||
requestSignal = init.signal!;
|
||||
return new Promise<Response>((_resolve, reject) => {
|
||||
requestSignal.addEventListener("abort", () => reject(requestSignal.reason), { once: true });
|
||||
});
|
||||
}));
|
||||
const pending = read(controller.signal);
|
||||
const rejected = expect(pending).rejects.toThrow("account deadline");
|
||||
controller.abort(new Error("account deadline"));
|
||||
await rejected;
|
||||
expect(requestSignal.aborted).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("fetchWithTimeout", () => {
|
||||
afterEach(() => {
|
||||
vi.unstubAllGlobals();
|
||||
|
||||
@@ -4,7 +4,7 @@ import { listServerAdapters } from "../adapters/registry.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { eq, inArray, and } from "drizzle-orm";
|
||||
import { companySecrets, type Db } from "@paperclipai/db";
|
||||
import { companySecrets, companySecretVersions, type Db } from "@paperclipai/db";
|
||||
import { fetchCodexQuota } from "@paperclipai/adapter-codex-local/server";
|
||||
import { fetchClaudeQuota } from "@paperclipai/adapter-claude-local/server";
|
||||
import { aiConnectionService } from "./ai-connections.js";
|
||||
@@ -25,8 +25,8 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId
|
||||
// Resolve revision metadata before consulting the cache, so rotation and
|
||||
// revocation cannot serve windows associated with an old credential.
|
||||
const secretIds = accounts.flatMap(row => row.grant.credentialSecretRefs.map(ref => ref.secretId));
|
||||
const revisions = secretIds.length ? await db.select({ id: companySecrets.id, updatedAt: companySecrets.updatedAt, latestVersion: companySecrets.latestVersion, status: companySecrets.status })
|
||||
.from(companySecrets).where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIds))) : [];
|
||||
const revisions = secretIds.length ? await db.select({ id: companySecrets.id, latestVersion: companySecrets.latestVersion, status: companySecrets.status, versionStatus: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt })
|
||||
.from(companySecrets).leftJoin(companySecretVersions, and(eq(companySecretVersions.secretId, companySecrets.id), eq(companySecretVersions.version, companySecrets.latestVersion))).where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIds))) : [];
|
||||
async function readAccount(row: (typeof accounts)[number]): Promise<ProviderQuotaResult> {
|
||||
const accountKey = createHash("sha256").update(JSON.stringify([companyId, row.connection.id, row.grant.id,
|
||||
row.connection.updatedAt, row.grant.updatedAt,
|
||||
@@ -39,10 +39,12 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId
|
||||
const key = `${userId}:${accountKey}`;
|
||||
let pending = accountRequests.get(key);
|
||||
if (!pending || pending.expires <= Date.now()) {
|
||||
const controller = new AbortController();
|
||||
const result = (async (): Promise<ProviderQuotaResult> => {
|
||||
let credentialResolved = false;
|
||||
try {
|
||||
const value = await service.credential(row);
|
||||
controller.signal.throwIfAborted();
|
||||
credentialResolved = true;
|
||||
let windows;
|
||||
if (base.provider === "openai") {
|
||||
@@ -52,9 +54,9 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId
|
||||
if (typeof token !== "string" || !token || typeof accountId !== "string" || !accountId) {
|
||||
return { ...base, ok: false, errorFamily: "credentials_unavailable", error: publicError, windows: [] };
|
||||
}
|
||||
windows = await fetchCodexQuota(token, accountId);
|
||||
windows = await fetchCodexQuota(token, accountId, controller.signal);
|
||||
} else {
|
||||
windows = await fetchClaudeQuota(value);
|
||||
windows = await fetchClaudeQuota(value, controller.signal);
|
||||
}
|
||||
return { ...base, ok: true, windows, capturedAt: new Date().toISOString() };
|
||||
} catch (error) {
|
||||
@@ -65,7 +67,7 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId
|
||||
...(invalid ? { errorFamily: "authentication_required" } : {}) };
|
||||
}
|
||||
})();
|
||||
const bounded = withQuotaTimeout(base.provider, result).then(quota => ({ ...base, ...quota, ...(quota.ok ? {} : { error: publicError }) }));
|
||||
const bounded = withQuotaTimeout(base.provider, result, () => controller.abort()).then(quota => ({ ...base, ...quota, ...(quota.ok ? {} : { error: publicError }) }));
|
||||
pending = { result: bounded, expires: Date.now() + 30_000 };
|
||||
if (accountRequests.size >= 500) accountRequests.delete(accountRequests.keys().next().value!);
|
||||
accountRequests.set(key, pending);
|
||||
@@ -130,6 +132,7 @@ export async function fetchAllQuotaWindows(): Promise<ProviderQuotaResult[]> {
|
||||
async function withQuotaTimeout(
|
||||
adapterType: string,
|
||||
task: Promise<ProviderQuotaResult>,
|
||||
onTimeout?: () => void,
|
||||
): Promise<ProviderQuotaResult> {
|
||||
let timeoutId: NodeJS.Timeout | null = null;
|
||||
try {
|
||||
@@ -137,6 +140,7 @@ async function withQuotaTimeout(
|
||||
task,
|
||||
new Promise<ProviderQuotaResult>((resolve) => {
|
||||
timeoutId = setTimeout(() => {
|
||||
onTimeout?.();
|
||||
resolve({
|
||||
provider: providerSlugForAdapterType(adapterType),
|
||||
ok: false,
|
||||
|
||||
Reference in new issue
Block a user