mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 09:07:46 +02:00
Merge accounting review fixes into dashboard stack
Co-Authored-By: Paperclip <noreply@paperclip.ing> # Conflicts: # ui/src/components/AccountQuotaPanels.tsx # ui/src/components/ProviderQuotaCard.tsx # ui/src/pages/Costs.production.tsx # ui/src/pages/Costs.tsx
This commit is contained in:
commit
85c8b8f72a
39 files changed
+1828
-50023
No files matched your search
@@ -155,3 +155,20 @@ Responses carry an opaque account identity and successful capture timestamp. The
|
||||
Both Costs page variants retain loaded reports and expanded rows during background refresh failures, with a small stale-data notice. Initial failures remain visible. Company/date changes isolate cached reports. The run-count label is `runs: 0 api · 11 sub`: distinct recorded runs, not tasks or individual model requests. Input totals include cache reads, with the cached portion shown explicitly.
|
||||
|
||||
Receipt identity and supplemental charges: `heartbeat:` idempotency keys are reserved for internally generated run receipts. Additional API-reported charges may link to the same run with their own keys. They contribute to monthly and acknowledged-run lifetime totals, while receipt-integrity checks compare only the original provider receipt. Recovery isolates each budget scope so a deleted target or failed scope does not prevent recovery elsewhere.
|
||||
|
||||
### Review hardening
|
||||
|
||||
Native restart recovery reuses a held reservation only when the coordinator's
|
||||
current, unexpired lease owns the same run and project. Ordinary duplicate
|
||||
provider dispatch remains rejected. A process adapter with no tokens, provider,
|
||||
price, or explicit unknown-price marker does not create an unpriced charge;
|
||||
provider adapters with missing prices still do.
|
||||
|
||||
Budget policy updates may omit unchanged settings, including the amount for an
|
||||
existing policy. Omitted settings are read under the accounting transaction lock.
|
||||
Creating a policy still requires an amount. Generic agent and company budget
|
||||
edits deliver hard-stop cancellation after their transaction commits.
|
||||
|
||||
Status-card update costs retain the ledger's fractional-cent storage precision.
|
||||
The mutation gate retains partial results and per-run reports/logs under
|
||||
`coverage/accounting/` when its baseline fails or a mutation survives.
|
||||
@@ -1452,6 +1452,59 @@ describe("shared ACPX engine runtime behavior", () => {
|
||||
expect(result.billingType).toBe("unknown");
|
||||
});
|
||||
|
||||
it.each(["failed", "cancelled"] as const)("keeps %s in-stream usage provisional without a final receipt", async (status) => {
|
||||
const root = await makeTempRoot();
|
||||
const stateDir = path.join(root, "state");
|
||||
const execute = createAcpxEngineExecutor({
|
||||
createRuntime: () => ({
|
||||
ensureSession: async () => ({
|
||||
backendSessionId: "backend-session",
|
||||
agentSessionId: "agent-session",
|
||||
runtimeSessionName: "runtime-session",
|
||||
}),
|
||||
startTurn: () => ({
|
||||
events: (async function* () {
|
||||
yield {
|
||||
type: "status",
|
||||
text: "usage",
|
||||
tag: "usage_update",
|
||||
cost: { amount: 0.31, currency: "USD" },
|
||||
breakdown: { inputTokens: 40, outputTokens: 700, cachedReadTokens: 60 },
|
||||
};
|
||||
|
||||
})(),
|
||||
result: Promise.resolve({ status, error: { code: "INTERNAL", message: "fixture failure" } }),
|
||||
cancel: async () => {},
|
||||
}),
|
||||
close: async () => {},
|
||||
}) as never,
|
||||
});
|
||||
|
||||
const onUsage = vi.fn();
|
||||
const result = await execute({
|
||||
runId: "run-usage-event-fallback",
|
||||
agent: {
|
||||
id: "agent-1",
|
||||
companyId: "company-1",
|
||||
},
|
||||
runtime: {},
|
||||
config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir },
|
||||
context: {},
|
||||
onLog: async () => {},
|
||||
onMeta: async () => {},
|
||||
onUsage,
|
||||
} as never);
|
||||
|
||||
expect(result.exitCode).not.toBe(0);
|
||||
expect(result.usageComplete).toBe(false);
|
||||
expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: false, costUsd: 0.31 }));
|
||||
expect(result.usage).toEqual({ inputTokens: 40, outputTokens: 700, cachedInputTokens: 60 });
|
||||
expect(result.usageBasis).toBe("per_run");
|
||||
expect(result.costUsd).toBeCloseTo(0.31);
|
||||
expect(result.provider).toBe("acpx");
|
||||
expect(result.billingType).toBe("unknown");
|
||||
});
|
||||
|
||||
it.skipIf(process.platform === "win32")("materializes ACPX Claude skills without symlinked descendants", async () => {
|
||||
const root = await makeTempRoot();
|
||||
const skillRoot = path.join(root, "skills");
|
||||
|
||||
@@ -4995,8 +4995,11 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
||||
eventBreakdown,
|
||||
eventCostUsd,
|
||||
});
|
||||
// An interrupted stream is provisional unless the runtime supplies its
|
||||
// post-terminal receipt. An in-stream usage_update alone cannot close it.
|
||||
const usageComplete = !channelLost && (turnSucceeded || postTurnStatus?.usage != null);
|
||||
await ctx.onUsage?.({ ...billingFields, usage: turnUsage.usage ?? undefined, costUsd: turnUsage.costUsd,
|
||||
model: prepared.requestedModel, usageBasis: "per_run", complete: !channelLost });
|
||||
model: prepared.requestedModel, usageBasis: "per_run", complete: usageComplete });
|
||||
const failedTurn = terminal.status === "failed" || terminal.status === "cancelled" || timedOut;
|
||||
// ACPX can defer session/load until runTurn. Forget an unavailable
|
||||
// session so the next bounded turn receives the full task conversation.
|
||||
@@ -5082,7 +5085,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
||||
model: prepared.requestedModel || null,
|
||||
...(turnUsage.usage ? { usage: turnUsage.usage, usageBasis: "per_run" as const } : {}),
|
||||
costUsd: turnUsage.costUsd,
|
||||
usageComplete: !channelLost,
|
||||
usageComplete,
|
||||
resultJson: {
|
||||
status: channelLost ? "failed" : terminal.status,
|
||||
...activityDiagnostics,
|
||||
|
||||
@@ -138,6 +138,36 @@ function createFreshLeaseSandboxRunner(options: {
|
||||
}
|
||||
|
||||
describe("cursor execute", () => {
|
||||
it.each([0, 7])("settles legacy step usage only after a clean process exit (%s)", async (exitCode) => {
|
||||
setPrepareCursorSandboxCommand.mockReset();
|
||||
setPrepareCursorSandboxCommand.mockImplementation(async (input) => ({
|
||||
command: input.command, env: input.env, remoteSystemHomeDir: null,
|
||||
addedPathEntry: null, preferredCommandPath: null,
|
||||
}));
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-cursor-legacy-"));
|
||||
const command = path.join(root, "agent.sh");
|
||||
await fs.writeFile(command, `#!/bin/sh
|
||||
cat >/dev/null
|
||||
printf '%s\\n' '{"type":"step_finish","part":{"tokens":{"input":20,"output":5},"cost":0.01}}'
|
||||
exit ${exitCode}
|
||||
`, { mode: 0o755 });
|
||||
const onUsage = vi.fn();
|
||||
try {
|
||||
const result = await execute({
|
||||
runId: "run-legacy", agent: { id: "agent-1", companyId: "company-1", name: "Cursor", adapterType: "cursor", adapterConfig: {} },
|
||||
runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
|
||||
config: { command, cwd: root }, context: createPromptContextFixture(),
|
||||
authToken: "fixture-run-token", onLog: async () => {}, onUsage,
|
||||
});
|
||||
expect(result.usage).toMatchObject({ inputTokens: 20, outputTokens: 5 });
|
||||
expect(result.costUsd).toBe(0.01);
|
||||
expect(result.usageComplete).toBe(exitCode === 0);
|
||||
expect(onUsage).toHaveBeenCalledWith(expect.objectContaining({ complete: false, costUsd: 0.01 }));
|
||||
} finally {
|
||||
await fs.rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ detail: "Authentication failed", structured: "", expected: "Authentication failed" },
|
||||
{ detail: "", structured: "", expected: "Cursor exited with code 7" },
|
||||
|
||||
@@ -701,12 +701,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
},
|
||||
clearSessionOnMissingSession = false,
|
||||
): AdapterExecutionResult => {
|
||||
const usageComplete = attempt.parsed.sawResult || (attempt.parsed.sawLegacyStep
|
||||
&& attempt.proc.exitCode === 0 && !attempt.proc.signal && !attempt.proc.timedOut
|
||||
&& !attempt.parsed.errorMessage);
|
||||
if (attempt.proc.timedOut) {
|
||||
return {
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.sawResult,
|
||||
usageComplete,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: providerFromModel,
|
||||
@@ -746,7 +749,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: attempt.parsed.sawResult,
|
||||
usageComplete,
|
||||
usageBasis: "per_run",
|
||||
errorMessage:
|
||||
(attempt.proc.exitCode ?? 0) === 0
|
||||
|
||||
@@ -68,6 +68,7 @@ export function parseCursorJsonl(stdout: string) {
|
||||
export function createCursorJsonlParser() {
|
||||
let sessionId: string | null = null;
|
||||
let sawResult = false;
|
||||
let sawLegacyStep = false;
|
||||
const messages: string[] = [];
|
||||
let errorMessage: string | null = null;
|
||||
let totalCostUsd = 0;
|
||||
@@ -156,6 +157,7 @@ export function createCursorJsonlParser() {
|
||||
}
|
||||
|
||||
if (type === "step_finish") {
|
||||
sawLegacyStep = true;
|
||||
const part = parseObject(event.part);
|
||||
const tokens = parseObject(part.tokens);
|
||||
const cache = parseObject(tokens.cache);
|
||||
@@ -169,6 +171,7 @@ export function createCursorJsonlParser() {
|
||||
|
||||
return {
|
||||
sawResult,
|
||||
sawLegacyStep,
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage: { ...usage },
|
||||
|
||||
@@ -326,6 +326,9 @@ describe("kimi_local execute", () => {
|
||||
const result = await execute(makeContext(root, { config: { cwd: root, timeoutSec: 5 } }));
|
||||
|
||||
expect(result.timedOut).toBe(true);
|
||||
expect(result.usageComplete).toBe(true);
|
||||
expect(result.costStatus).toBe("unpriced");
|
||||
expect(result.costUsd).toBeNull();
|
||||
expect(result.errorMessage).toContain("5s");
|
||||
});
|
||||
|
||||
|
||||
@@ -648,7 +648,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: false,
|
||||
// This stream protocol has no usage receipt to recover after exit.
|
||||
// Close collection with an explicitly unknown price, never a free run.
|
||||
usageComplete: true,
|
||||
costStatus: "unpriced",
|
||||
usageBasis: "per_run",
|
||||
provider: "moonshot",
|
||||
biller: "moonshot",
|
||||
@@ -711,6 +714,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: true,
|
||||
costStatus: "unpriced",
|
||||
usageBasis: "per_run",
|
||||
errorMessage: failed ? fallbackErrorMessage : null,
|
||||
// Forward the transport-level error code from the run-disposition seam
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
-- Preserve the ledger precision in status-card update snapshots. Reapplying this type widening is safe.
|
||||
ALTER TABLE "status_card_updates" ALTER COLUMN "cost_cents" SET DATA TYPE numeric(24, 7);
|
||||
-49948
File diff suppressed because it is too large.
Load diff
+1392
-12
File diff suppressed because it is too large.
Load diff
+7
@@ -2059,6 +2059,13 @@
|
||||
"when": 1790983662974,
|
||||
"tag": "0295_lean_baron_strucker",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 296,
|
||||
"version": "7",
|
||||
"when": 1791086889296,
|
||||
"tag": "0296_perpetual_black_cat",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import { boolean, index, integer, jsonb, pgTable, text, timestamp, uuid } from "drizzle-orm/pg-core";
|
||||
import { boolean, index, integer, numeric, jsonb, pgTable, text, timestamp, uuid } from "drizzle-orm/pg-core";
|
||||
import type { CompanySearchQuery, StatusCardRefreshPolicy } from "@paperclipai/shared";
|
||||
import { agents } from "./agents.js";
|
||||
import { companies } from "./companies.js";
|
||||
@@ -81,7 +81,7 @@ export const statusCardUpdates = pgTable(
|
||||
changes: jsonb("changes").$type<StatusCardUpdateChange[]>().notNull().default(sql`'[]'::jsonb`),
|
||||
inputTokens: integer("input_tokens").notNull().default(0),
|
||||
outputTokens: integer("output_tokens").notNull().default(0),
|
||||
costCents: integer("cost_cents").notNull().default(0),
|
||||
costCents: numeric("cost_cents", { precision: 24, scale: 7, mode: "number" }).notNull().default(0),
|
||||
model: text("model"),
|
||||
queryVersion: integer("query_version"),
|
||||
changeSummary: text("change_summary"),
|
||||
|
||||
@@ -92,7 +92,7 @@ export interface BudgetPolicyUpsertInput {
|
||||
scopeId: string;
|
||||
metric?: BudgetMetric;
|
||||
windowKind?: BudgetWindowKind;
|
||||
amount: number;
|
||||
amount?: number;
|
||||
reservationCents?: string | number;
|
||||
warnPercent?: number;
|
||||
hardStopEnabled?: boolean;
|
||||
|
||||
@@ -12,13 +12,13 @@ export const upsertBudgetPolicySchema = z.object({
|
||||
scopeId: z.string().guid(),
|
||||
metric: z.enum(BUDGET_METRICS).optional().default("billed_cents"),
|
||||
windowKind: z.enum(BUDGET_WINDOW_KINDS).optional().default("calendar_month_utc"),
|
||||
amount: z.number().int().nonnegative().max(2_147_483_647),
|
||||
amount: z.number().int().nonnegative().max(2_147_483_647).optional(),
|
||||
reservationCents: exactCentsSchema.optional(),
|
||||
warnPercent: z.number().int().min(1).max(99).optional().default(80),
|
||||
hardStopEnabled: z.boolean().optional().default(true),
|
||||
warnPercent: z.number().int().min(1).max(99).optional(),
|
||||
hardStopEnabled: z.boolean().optional(),
|
||||
unpricedUsagePolicy: z.enum(["block", "allow"]).optional(),
|
||||
notifyEnabled: z.boolean().optional().default(true),
|
||||
isActive: z.boolean().optional().default(true),
|
||||
notifyEnabled: z.boolean().optional(),
|
||||
isActive: z.boolean().optional(),
|
||||
});
|
||||
|
||||
export type UpsertBudgetPolicy = z.infer<typeof upsertBudgetPolicySchema>;
|
||||
|
||||
@@ -320,6 +320,8 @@ function runVitest(args, label, testShard = null) {
|
||||
const env = {
|
||||
...process.env,
|
||||
NODE_ENV: "test",
|
||||
PAPERCLIP_TEST_HOST_HOME: process.env.PAPERCLIP_TEST_HOST_HOME
|
||||
?? (process.env.PAPERCLIP_HOME?.trim() || path.join(os.homedir(), ".paperclip")),
|
||||
PAPERCLIP_HOME: path.join(testRoot, "h"),
|
||||
// Config discovery otherwise prefers the checkout's .paperclip/config.json
|
||||
// over PAPERCLIP_HOME, importing preview scheduling policy into unit tests.
|
||||
|
||||
@@ -1,34 +1,47 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { mkdtemp, readFile, rm, writeFile, mkdir } from "node:fs/promises";
|
||||
import { mkdtemp, readFile, rm, writeFile, mkdir, copyFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
const root = fileURLToPath(new URL("..", import.meta.url));
|
||||
const directory = await mkdtemp(path.join(tmpdir(), "accounting-mutations-"));
|
||||
const artifactDirectory = path.join(root, "coverage/accounting/mutations");
|
||||
const targets = { deduplication: "deduplicates a retried receipt", company: "isolates company reporting", projection: "conserves agent projections", threshold: "stops at the exact budget boundary" };
|
||||
const evidence = [];
|
||||
const persist = () => writeFile(path.join(root, "coverage/accounting/mutations.json"), JSON.stringify({ generatedAt: new Date().toISOString(), evidence }, null, 2));
|
||||
await mkdir(artifactDirectory, { recursive: true });
|
||||
try {
|
||||
for (const name of ["baseline", ...Object.keys(targets)]) {
|
||||
const report = path.join(directory, `${name}.json`);
|
||||
const result = await new Promise((resolve, reject) => {
|
||||
const entry = { mutation: name, result: "error", assertions: [] };
|
||||
evidence.push(entry);
|
||||
await persist();
|
||||
const result = await new Promise((resolve) => {
|
||||
const child = spawn(process.execPath, ["node_modules/vitest/vitest.mjs", "run", "--config", "server/vitest.accounting-mutations.config.ts", "--reporter=json", `--outputFile=${report}`], {
|
||||
cwd: root, env: { ...process.env, PAPERCLIP_ACCOUNTING_MUTATION: name === "baseline" ? "" : name }, stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
let log = ""; const capture = data => { log += data; };
|
||||
child.stdout.on("data", capture); child.stderr.on("data", capture);
|
||||
child.once("error", reject); child.once("exit", code => resolve({ code, log }));
|
||||
child.once("error", error => resolve({ code: null, log: `${log}\n${error.message}` }));
|
||||
child.once("close", code => resolve({ code, log }));
|
||||
});
|
||||
await writeFile(path.join(artifactDirectory, `${name}.log`), result.log);
|
||||
await copyFile(report, path.join(artifactDirectory, `${name}.json`)).catch(() => {});
|
||||
const json = JSON.parse(await readFile(report, "utf8").catch(() => { throw new Error(`${name} produced no test evidence: ${result.log}`); }));
|
||||
const assertions = json.testResults.flatMap(file => file.assertionResults);
|
||||
entry.assertions = assertions.map(({ title, status }) => ({ title, status }));
|
||||
if (name === "baseline") {
|
||||
if (result.code !== 0 || assertions.length !== 4 || assertions.some(test => test.status !== "passed")) throw new Error(`Mutation baseline failed or skipped: ${result.log}`);
|
||||
entry.result = result.code === 0 && assertions.length === 4 && assertions.every(test => test.status === "passed") ? "passed" : "baseline_failed";
|
||||
if (entry.result !== "passed") throw new Error(`Mutation baseline failed or skipped: ${result.log}`);
|
||||
} else {
|
||||
const sentinel = assertions.find(test => test.title === targets[name]);
|
||||
if (result.code === 0 || !result.log.includes(`ACCOUNTING_MUTATION_APPLIED ${name}`) || sentinel?.status !== "failed") throw new Error(`Mutation ${name} survived or failed without its sentinel: ${result.log}`);
|
||||
entry.result = result.code !== 0 && result.log.includes(`ACCOUNTING_MUTATION_APPLIED ${name}`) && sentinel?.status === "failed" ? "killed" : "survived_or_invalid";
|
||||
if (entry.result !== "killed") throw new Error(`Mutation ${name} survived or failed without its sentinel: ${result.log}`);
|
||||
}
|
||||
evidence.push({ mutation: name, result: name === "baseline" ? "passed" : "killed", assertions: assertions.map(({ title, status }) => ({ title, status })) });
|
||||
process.stdout.write(`${name}: ${name === "baseline" ? "passed" : "killed by expected assertion"}\n`);
|
||||
await persist();
|
||||
process.stdout.write(`${name}: ${entry.result}\n`);
|
||||
}
|
||||
await mkdir(path.join(root, "coverage/accounting"), { recursive: true });
|
||||
await writeFile(path.join(root, "coverage/accounting/mutations.json"), JSON.stringify({ generatedAt: new Date().toISOString(), evidence }, null, 2));
|
||||
} finally { await rm(directory, { recursive: true, force: true }); }
|
||||
} finally {
|
||||
// Failed baselines, surviving mutants, and missing reports are evidence too.
|
||||
try { await persist(); } finally { await rm(directory, { recursive: true, force: true }); }
|
||||
}
|
||||
@@ -99,7 +99,8 @@ describe("CLI adapter accounting on timeout", () => {
|
||||
context: {}, onLog: async () => {},
|
||||
});
|
||||
expect(result.timedOut).toBe(true);
|
||||
expect(result.usageComplete).toBe(["codex", "cursor", "gemini"].includes(fixture.name));
|
||||
expect(result.usageComplete).toBe(["codex", "cursor", "gemini", "kimi"].includes(fixture.name));
|
||||
if (fixture.name === "kimi") expect(result.costStatus).toBe("unpriced");
|
||||
expect(result.usageBasis).toBe("per_run");
|
||||
expect(result.provider).toBeTruthy();
|
||||
expect(result.billingType).toBeTruthy();
|
||||
|
||||
@@ -84,6 +84,33 @@ describe("managed AI connections", () => {
|
||||
} finally { fetchSpy.mockRestore(); }
|
||||
});
|
||||
|
||||
it("serializes quota credential refresh, persists rotation, and defers while the provider is active", async () => {
|
||||
const owner = "quota-refresh-owner";
|
||||
await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
|
||||
const original = JSON.stringify({ tokens: { access_token: "expired-access", refresh_token: "single-use", id_token: "identity", account_id: null } });
|
||||
const account = await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: "Refresh account", loginSessionId: "fixture", agentIds: [], allAgents: true }, original);
|
||||
const [row] = await service.quotaAccounts(companyId, owner);
|
||||
const request = vi.spyOn(globalThis, "fetch").mockImplementation(async () => new Response(JSON.stringify({ access_token: "fresh-access", refresh_token: "rotated-once" })));
|
||||
const signal = new AbortController().signal;
|
||||
try {
|
||||
const activeId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: activeId, companyId, agentId, invocationSource: "on_demand", status: "running", contextSnapshot: { aiConnection: { grantId: account.grantId } } });
|
||||
await expect(service.refreshQuotaCredential(row, original, signal)).rejects.toThrow("provider_unavailable");
|
||||
expect(request).not.toHaveBeenCalled();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, activeId));
|
||||
const [first, second] = await Promise.all([
|
||||
service.refreshQuotaCredential(row, original, signal),
|
||||
service.refreshQuotaCredential(row, original, signal),
|
||||
]);
|
||||
expect(first).toBe(second);
|
||||
expect(request).toHaveBeenCalledTimes(1);
|
||||
expect(JSON.parse(await service.credential(row))).toMatchObject({ tokens: { access_token: "fresh-access", refresh_token: "rotated-once", account_id: null } });
|
||||
await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, account.grantId));
|
||||
await expect(service.refreshQuotaCredential(row, first, signal)).rejects.toThrow("credentials_unavailable");
|
||||
expect(request).toHaveBeenCalledTimes(1);
|
||||
} finally { request.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" });
|
||||
|
||||
@@ -3,6 +3,7 @@ import type { Db } from "@paperclipai/db";
|
||||
const mocks = vi.hoisted(() => ({
|
||||
accounts: vi.fn(),
|
||||
credential: vi.fn(),
|
||||
refresh: vi.fn(),
|
||||
codex: vi.fn(),
|
||||
claude: vi.fn(),
|
||||
adapters: vi.fn(),
|
||||
@@ -11,6 +12,7 @@ vi.mock("../services/ai-connections.js", () => ({
|
||||
aiConnectionService: () => ({
|
||||
quotaAccounts: mocks.accounts,
|
||||
credential: mocks.credential,
|
||||
refreshQuotaCredential: mocks.refresh,
|
||||
}),
|
||||
}));
|
||||
vi.mock("../adapters/registry.js", () => ({
|
||||
@@ -56,6 +58,7 @@ describe("connected account quotas", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
revision = 1;
|
||||
mocks.refresh.mockRejectedValue(new Error("authentication_required"));
|
||||
mocks.credential.mockResolvedValue(
|
||||
JSON.stringify({
|
||||
tokens: { access_token: "private-token", account_id: "actual-account" },
|
||||
@@ -135,17 +138,34 @@ describe("connected account quotas", () => {
|
||||
new Error("private credential store failure"),
|
||||
);
|
||||
const [failed] = await fetchCompanyQuotaWindows(db, "company-6", "user-6");
|
||||
expect(failed.errorFamily).toBe("authentication_required");
|
||||
expect(failed.errorFamily).toBe("provider_unavailable");
|
||||
expect(JSON.stringify(failed)).not.toContain("private credential");
|
||||
});
|
||||
it("rejects an incomplete account identity without querying a different account", async () => {
|
||||
it("reads quota for a verified login without an optional account ID", async () => {
|
||||
mocks.accounts.mockResolvedValue([account("missing-identity")]);
|
||||
mocks.credential.mockResolvedValue(
|
||||
JSON.stringify({ accessToken: "private" }),
|
||||
);
|
||||
const [result] = await fetchCompanyQuotaWindows(db, "company-7", "user-7");
|
||||
expect(result.errorFamily).toBe("credentials_unavailable");
|
||||
expect(mocks.codex).not.toHaveBeenCalled();
|
||||
expect(result.ok).toBe(true);
|
||||
expect(mocks.codex).toHaveBeenCalledWith("private", null, expect.any(AbortSignal));
|
||||
});
|
||||
it("distinguishes permission errors from revoked credentials", async () => {
|
||||
mocks.accounts.mockResolvedValue([account("permission-error")]);
|
||||
mocks.codex.mockRejectedValueOnce(new Error("chatgpt wham api returned 403"));
|
||||
const [result] = await fetchCompanyQuotaWindows(db, "permission-company", "user");
|
||||
expect(result.errorFamily).toBe("permission_denied");
|
||||
expect(mocks.refresh).not.toHaveBeenCalled();
|
||||
});
|
||||
it("refreshes an expired selected credential once before retrying quota", async () => {
|
||||
mocks.accounts.mockResolvedValue([account("refresh-account")]);
|
||||
mocks.codex.mockRejectedValueOnce(new Error("chatgpt wham api returned 401"));
|
||||
mocks.refresh.mockResolvedValueOnce(JSON.stringify({ tokens: { access_token: "new-token", account_id: "same-account" } }));
|
||||
const [result] = await fetchCompanyQuotaWindows(db, "refresh-company", "user");
|
||||
expect(result.ok).toBe(true);
|
||||
expect(mocks.refresh).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.codex).toHaveBeenLastCalledWith("new-token", "same-account", expect.any(AbortSignal));
|
||||
expect(JSON.stringify(result)).not.toContain("new-token");
|
||||
});
|
||||
it("bounds a stalled credential lookup and preserves its account identity", async () => {
|
||||
vi.useFakeTimers();
|
||||
|
||||
@@ -4,7 +4,7 @@ import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { agents, agentRuntimeState, budgetReservations, companies, costEvents, costAdjustments, createDb, heartbeatRuns, runUsageReceipts } from "@paperclipai/db";
|
||||
import { agents, agentRuntimeState, budgetReservations, companies, costEvents, costAdjustments, createDb, heartbeatRuns, runUsageReceipts, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { accountingIntegrityService } from "../services/accounting-integrity.js";
|
||||
import { billingReconciliationService } from "../services/billing-reconciliation.js";
|
||||
import { budgetService } from "../services/budgets.js";
|
||||
@@ -65,6 +65,37 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
expect((await accountingIntegrityService(db).inspect(f.company.id)).findings).toEqual([]);
|
||||
});
|
||||
|
||||
it("reuses a held native reservation only for its current recovery lease", async () => {
|
||||
const f = await fixture(); const run = await runFor(f);
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Recover" }).returning();
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, run.id));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: f.company.id, runId: run.id, issueId: issue.id,
|
||||
phase: "running", leaseOwner: "successor", leaseExpiresAt: new Date(Date.now() + 60_000) });
|
||||
await budgetService(db).upsertPolicy(f.company.id, { scopeType: "company", scopeId: f.company.id, amount: 10, reservationCents: "6" }, "board");
|
||||
const original = await reserveRunBudget(db, f.company.id, run.id, null);
|
||||
await expect(reserveRunBudget(db, f.company.id, run.id, null)).rejects.toThrow("already reserved");
|
||||
await expect(reserveRunBudget(db, f.company.id, run.id, null, {}, "stale-owner")).rejects.toThrow("already reserved");
|
||||
expect(await reserveRunBudget(db, f.company.id, run.id, null, {}, "successor")).toEqual(original);
|
||||
expect((await accountingIntegrityService(db).health(f.company.id)).heldReservationCents).toBe("6.0000000");
|
||||
await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
await expect(reserveRunBudget(db, f.company.id, run.id, null, {}, "successor")).rejects.toThrow("already reserved");
|
||||
});
|
||||
|
||||
it.each([false, true])("settles absent usage without hiding explicit unknown pricing (%s)", async (unpriced) => {
|
||||
const f = await fixture(); const run = await runFor(f);
|
||||
await budgetService(db).upsertPolicy(f.company.id, { scopeType: "agent", scopeId: f.agent.id, amount: 100, reservationCents: "10" }, "board");
|
||||
await reserveRunBudget(db, f.company.id, run.id, null);
|
||||
const recorder = await createRunUsageRecorder(db, { companyId: f.company.id, runId: run.id, adapterType: "process" }, directory);
|
||||
await recorder.complete({ exitCode: 0, signal: null, timedOut: false, ...(unpriced ? { provider: "moonshot", costStatus: "unpriced" as const, costUsd: null } : {}) });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await accountRunCost(db, run.id)).toBe(true);
|
||||
expect((await accountingIntegrityService(db).health(f.company.id)).heldReservationCents).toBe("0.0000000");
|
||||
const summary = await costService(db).summary(f.company.id);
|
||||
expect(summary.unpricedEventCount).toBe(unpriced ? 1 : 0);
|
||||
const block = await budgetService(db).getInvocationBlock(f.company.id, f.agent.id);
|
||||
if (unpriced) expect(block).not.toBeNull(); else expect(block).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps an interrupted partial receipt pending and settles its reservation when evidence becomes complete", async () => {
|
||||
const f = await fixture(); const run = await runFor(f);
|
||||
await budgetService(db).upsertPolicy(f.company.id, { scopeType: "agent", scopeId: f.agent.id, amount: 100, reservationCents: "10" }, "board");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { upsertBudgetPolicySchema } from "@paperclipai/shared";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
@@ -288,6 +289,31 @@ databaseDescribe("cost accounting reliability (PostgreSQL)", () => {
|
||||
expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).toMatchObject({ scopeType: "company" });
|
||||
});
|
||||
|
||||
it("delivers cancellation after generic budget updates commit", async () => {
|
||||
const f = await fixture();
|
||||
await costService(db).createEvent(f.company.id, f.receipt);
|
||||
const cancelWorkForScope = vi.fn(async (scope) => {
|
||||
const policies = await budgetService(db).listPolicies(f.company.id);
|
||||
expect(policies.some(policy => policy.scopeType === scope.scopeType && policy.amount < 100)).toBe(true);
|
||||
});
|
||||
await services.value.agentService(db, { cancelWorkForScope }).update(f.agent.id, { budgetMonthlyCents: 50 });
|
||||
await services.value.companyService(db, { cancelWorkForScope }).update(f.company.id, { budgetMonthlyCents: 75 });
|
||||
expect(cancelWorkForScope).toHaveBeenCalledWith(expect.objectContaining({ scopeType: "agent", scopeId: f.agent.id }));
|
||||
expect(cancelWorkForScope).toHaveBeenCalledWith(expect.objectContaining({ scopeType: "company", scopeId: f.company.id }));
|
||||
const rows = await db.select().from(budgetPolicies).where(eq(budgetPolicies.companyId, f.company.id));
|
||||
expect(rows.every(row => row.enforcementVersion === row.enforcementDeliveredVersion)).toBe(true);
|
||||
});
|
||||
|
||||
it("merges partial policy edits with current settings under the accounting lock", async () => {
|
||||
const f = await fixture(); const budgets = budgetService(db);
|
||||
const identity = { scopeType: "agent" as const, scopeId: f.agent.id };
|
||||
await budgets.upsertPolicy(f.company.id, { ...identity, amount: 100, hardStopEnabled: true }, "board");
|
||||
await budgets.upsertPolicy(f.company.id, { ...identity, amount: 200, hardStopEnabled: false, warnPercent: 60, notifyEnabled: false }, "other-operator");
|
||||
const saved = await budgets.upsertPolicy(f.company.id, upsertBudgetPolicySchema.parse({ ...identity, reservationCents: "10" }), "board");
|
||||
expect(saved).toMatchObject({ amount: 200, hardStopEnabled: false, warnPercent: 60, notifyEnabled: false, reservationCents: "10.0000000" });
|
||||
await expect(budgets.upsertPolicy(f.company.id, { scopeType: "project", scopeId: f.project.id, reservationCents: "1" }, "board")).rejects.toThrow("Amount is required");
|
||||
});
|
||||
|
||||
it("conserves both spend and tokens across complete per-model receipts", async () => {
|
||||
const f = await fixture();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), costAccountingPending: true, usageJson: {
|
||||
|
||||
@@ -21,6 +21,9 @@ type SupertestTestConstructor = {
|
||||
};
|
||||
};
|
||||
|
||||
// Keep the original home available only for read-only host port-lease checks.
|
||||
process.env.PAPERCLIP_TEST_HOST_HOME ??= process.env.PAPERCLIP_HOME?.trim() || path.join(os.homedir(), ".paperclip");
|
||||
|
||||
// Receipt-spool recovery must never scan a developer's instance during tests.
|
||||
const paperclipTestHome = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-vitest-home-"));
|
||||
process.env.PAPERCLIP_HOME = paperclipTestHome;
|
||||
|
||||
@@ -1078,7 +1078,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
model: "gpt-5.4",
|
||||
inputTokens: 5200,
|
||||
outputTokens: 980,
|
||||
costCents: 2,
|
||||
costCents: 0.1234567,
|
||||
occurredAt: new Date(),
|
||||
});
|
||||
const summaryWrite = await request(writerApp).put(`/api/status-cards/${created.body.id}/summary`).send({
|
||||
@@ -1111,7 +1111,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
summaryBody: "**Decide:** unblock launch approval.\n\n**Recent work:** launch review is waiting.",
|
||||
watchedIssueCount: 1,
|
||||
todayTokens: 6180,
|
||||
todayCostCents: 2,
|
||||
todayCostCents: 0.1234567,
|
||||
};
|
||||
const detail = await request(boardApp).get(`/api/status-cards/${created.body.id}`);
|
||||
expect(detail.status).toBe(200);
|
||||
@@ -1143,7 +1143,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
model: "gpt-5.4",
|
||||
inputTokens: 1300,
|
||||
outputTokens: 410,
|
||||
costCents: 1,
|
||||
costCents: 3_000_000_000.25,
|
||||
occurredAt: new Date(),
|
||||
});
|
||||
const incrementalWrite = await request(createApp(db, agentActor(company.id, summarizer.id, updateRun.id)))
|
||||
@@ -1155,7 +1155,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
model: "gpt-5.4",
|
||||
});
|
||||
expect(incrementalWrite.status).toBe(200);
|
||||
expect(await db.select().from(statusCardUpdates).then((rows) => rows.find((row) => row.kind === "incremental"))).toMatchObject({ inputTokens: 1300, outputTokens: 410 });
|
||||
expect(await db.select().from(statusCardUpdates).then((rows) => rows.find((row) => row.kind === "incremental"))).toMatchObject({ inputTokens: 1300, outputTokens: 410, costCents: 3_000_000_000.25 });
|
||||
const revisions = await request(boardApp).get(`/api/status-cards/${created.body.id}/summary-revisions`);
|
||||
expect(revisions.status).toBe(200);
|
||||
expect(revisions.body.map((row: { revisionNumber: number }) => row.revisionNumber)).toEqual([2, 1]);
|
||||
|
||||
@@ -101,7 +101,9 @@ let NEXT_HMR_PORT: number;
|
||||
* must be read before that happens. Mirrors the default in
|
||||
* `resolvePaperclipHomeDir`.
|
||||
*/
|
||||
const REAL_PAPERCLIP_HOME = process.env.PAPERCLIP_HOME?.trim() || path.join(os.homedir(), ".paperclip");
|
||||
const REAL_PAPERCLIP_HOME = process.env.PAPERCLIP_TEST_HOST_HOME?.trim()
|
||||
|| process.env.PAPERCLIP_HOME?.trim()
|
||||
|| path.join(os.homedir(), ".paperclip");
|
||||
|
||||
/**
|
||||
* Every port a local Paperclip instance's on-disk service registry currently
|
||||
|
||||
@@ -539,7 +539,7 @@ export function agentRoutes(
|
||||
const KNOWN_INSTRUCTIONS_BUNDLE_KEY_SET: ReadonlySet<string> = new Set(KNOWN_INSTRUCTIONS_BUNDLE_KEYS);
|
||||
|
||||
const router = Router();
|
||||
const svc = agentService(db);
|
||||
const svc = agentService(db, { cancelWorkForScope: scope => heartbeat.cancelBudgetScopeWork(scope) });
|
||||
const access = accessService(db);
|
||||
const approvalsSvc = approvalService(db);
|
||||
const budgets = budgetService(db);
|
||||
|
||||
@@ -55,6 +55,7 @@ import {
|
||||
accessService,
|
||||
agentService,
|
||||
budgetService,
|
||||
heartbeatService,
|
||||
buildExportFidelityReport,
|
||||
collectExportFidelityCounts,
|
||||
companyArtifactsService,
|
||||
@@ -278,7 +279,8 @@ export function companyRoutes(db: Db, storage?: StorageService, options?: Compan
|
||||
const router = Router();
|
||||
const importTransferSpoolRoot =
|
||||
options?.importTransferSpoolRoot ?? resolveDefaultImportTransferSpoolRoot();
|
||||
const svc = companyService(db);
|
||||
const heartbeat = heartbeatService(db);
|
||||
const svc = companyService(db, { cancelWorkForScope: heartbeat.cancelBudgetScopeWork });
|
||||
const agents = agentService(db);
|
||||
const portability = companyPortabilityService(db, storage);
|
||||
const access = accessService(db);
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { budgetServiceInTransaction } from "./budgets.js";
|
||||
import { budgetServiceInTransaction, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js";
|
||||
import { withAccountingTransaction } from "./accounting-transaction.js";
|
||||
import type { ActivityPublication } from "./activity-log.js";
|
||||
import { agentAppearanceSchema, randomAgentAppearance, resolveAgentAppearance, agentAvatarUrl } from "@paperclipai/shared";
|
||||
@@ -341,7 +341,7 @@ export function deduplicateAgentName(
|
||||
return `${candidateName} ${Date.now()}`;
|
||||
}
|
||||
|
||||
export function agentService(db: Db) {
|
||||
export function agentService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
|
||||
const secretsSvc = secretService(db);
|
||||
|
||||
function currentUtcMonthWindow(now = new Date()) {
|
||||
@@ -860,7 +860,11 @@ export function agentService(db: Db) {
|
||||
return normalizedUpdated;
|
||||
};
|
||||
|
||||
if (normalizedPatch.budgetMonthlyCents !== undefined) return withAccountingTransaction(db, existing.companyId, applyUpdate);
|
||||
if (normalizedPatch.budgetMonthlyCents !== undefined) {
|
||||
const result = await withAccountingTransaction(db, existing.companyId, applyUpdate);
|
||||
await deliverBudgetEnforcement(db, budgetHooks, existing.companyId);
|
||||
return result;
|
||||
}
|
||||
|
||||
const transaction = (db as unknown as {
|
||||
transaction?: (callback: (tx: unknown) => Promise<AgentUpdateResult>) => Promise<AgentUpdateResult>;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { syncConnectionCredentialBindings } from "./connection-credential-bindings.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { and, eq, inArray, or } from "drizzle-orm";
|
||||
import { and, eq, inArray, or, sql } from "drizzle-orm";
|
||||
import {
|
||||
type Db,
|
||||
authUsers,
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
aiConnectionDefaults,
|
||||
aiProviderDefaults,
|
||||
agents,
|
||||
heartbeatRuns,
|
||||
companyMemberships,
|
||||
companySecrets,
|
||||
userSecretDefinitions,
|
||||
@@ -848,6 +849,63 @@ export function aiConnectionService(db: Db) {
|
||||
});
|
||||
});
|
||||
}
|
||||
/** Refresh only the selected vaulted identity, serializing rotation and reconnect. */
|
||||
async function refreshQuotaCredential(row: Parameters<typeof credential>[0], failedValue: string, signal: AbortSignal) {
|
||||
return db.transaction(async (tx) => {
|
||||
const companyId = row.connection.companyId;
|
||||
const [grant] = await tx.select().from(connectionGrants).where(and(
|
||||
eq(connectionGrants.id, row.grant.id), eq(connectionGrants.companyId, companyId),
|
||||
eq(connectionGrants.connectionId, row.connection.id),
|
||||
)).for("update");
|
||||
if (!grant || grant.status !== "active") throw new Error("credentials_unavailable");
|
||||
const ref = grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential");
|
||||
if (!ref) throw new Error("credentials_unavailable");
|
||||
await tx.select({ id: companySecrets.id }).from(companySecrets).where(and(
|
||||
eq(companySecrets.id, ref.secretId), eq(companySecrets.companyId, companyId),
|
||||
)).for("update");
|
||||
const current = await aiConnectionService(tx as unknown as Db).credential({ connection: row.connection, grant });
|
||||
signal.throwIfAborted();
|
||||
// Another poll, an agent, or reconnect already supplied a newer credential.
|
||||
if (current !== failedValue) return current;
|
||||
// An active provider may rotate this single-use token itself. Defer polling
|
||||
// until its credential write-back instead of invalidating its live copy.
|
||||
const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
|
||||
.innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, companyId))).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
|
||||
or(sql`${heartbeatRuns.contextSnapshot}->'aiConnection'->>'grantId' = ${grant.id}`,
|
||||
sql`${agents.runtimeConfig}->'aiConnection'->>'provider' = 'openai'`, eq(agents.adapterType, "codex_local")),
|
||||
)).limit(1);
|
||||
if (active) throw new Error("provider_unavailable");
|
||||
const auth = JSON.parse(current);
|
||||
const refreshToken = auth.tokens?.refresh_token;
|
||||
if (typeof refreshToken !== "string" || !refreshToken) throw new Error("authentication_required");
|
||||
// Matches the Codex OAuth client (openai/codex, login/src/auth/manager.rs).
|
||||
const response = await fetch("https://auth.openai.com/oauth/token", {
|
||||
method: "POST", redirect: "error", signal,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ client_id: "app_EMoamEEZ73f0CkXaXp7hrann", grant_type: "refresh_token", refresh_token: refreshToken }),
|
||||
});
|
||||
if (!response.ok) {
|
||||
await response.body?.cancel();
|
||||
throw new Error(response.status === 400 || response.status === 401 ? "authentication_required" : "provider_unavailable");
|
||||
}
|
||||
const tokens = await response.json() as Record<string, unknown>;
|
||||
if (typeof tokens.access_token !== "string" || !tokens.access_token) throw new Error("provider_unavailable");
|
||||
const refreshed = JSON.stringify({ ...auth, tokens: { ...auth.tokens,
|
||||
access_token: tokens.access_token,
|
||||
...(typeof tokens.refresh_token === "string" && tokens.refresh_token ? { refresh_token: tokens.refresh_token } : {}),
|
||||
...(typeof tokens.id_token === "string" && tokens.id_token ? { id_token: tokens.id_token } : {}),
|
||||
}, last_refresh: new Date().toISOString() });
|
||||
// Persist the rotated token even if the quota deadline expires after the
|
||||
// successful exchange; losing it would strand all subsequent executions.
|
||||
await secretService(tx).rotate(ref.secretId, { value: refreshed }, { userId: grant.subjectUserId });
|
||||
await tx.update(connectionGrants).set({ updatedAt: new Date() }).where(eq(connectionGrants.id, grant.id));
|
||||
await logActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "quota",
|
||||
action: "ai_connection.credential_refreshed", entityType: "tool_connection", entityId: row.connection.id,
|
||||
details: { provider: "openai", grantId: grant.id } });
|
||||
return refreshed;
|
||||
});
|
||||
}
|
||||
// A quota read uses the same credential audience as execution. Operator status
|
||||
// alone never grants access to another member's personal subscription.
|
||||
async function quotaAccounts(companyId: string, userId: string) {
|
||||
@@ -860,5 +918,5 @@ export function aiConnectionService(db: Db) {
|
||||
return [{ ...row, summary }];
|
||||
});
|
||||
}
|
||||
return { list, quotaAccounts, select, credential, probeUsage, save, setDefault, membership, markAuthenticationFailed };
|
||||
return { list, quotaAccounts, refreshQuotaCredential, select, credential, probeUsage, save, setDefault, membership, markAuthenticationFailed };
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { budgetPolicies, budgetReservations, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { budgetPolicies, budgetReservations, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { centsToUnits, unitsToCents } from "@paperclipai/shared";
|
||||
import { conflict, notFound } from "../errors.js";
|
||||
import { withAccountingTransaction } from "./accounting-transaction.js";
|
||||
@@ -8,13 +8,23 @@ import { budgetServiceInTransaction, computeObservedSpend } from "./budgets.js";
|
||||
/** Reserve before dispatch, under the same company lock as charges and policy
|
||||
* changes. Estimates constrain admission; they cannot cap a provider's bill.
|
||||
* A reservation survives timeouts and restarts until accounting proves closure. */
|
||||
export async function reserveRunBudget(db: Db, companyId: string, runId: string, projectId: string | null, ledgerScope: Record<string, unknown> = {}) {
|
||||
export async function reserveRunBudget(db: Db, companyId: string, runId: string, projectId: string | null, ledgerScope: Record<string, unknown> = {}, recoveryLeaseOwner?: string) {
|
||||
return withAccountingTransaction(db, companyId, async (tx, publications) => {
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).for("update");
|
||||
if (!run) throw notFound("Run not found");
|
||||
if (run.costAccountedAt || !["queued", "running"].includes(run.status)) throw conflict("Run can no longer start provider work");
|
||||
const [existing] = await tx.select().from(budgetReservations).where(and(eq(budgetReservations.companyId, companyId), eq(budgetReservations.runId, runId)));
|
||||
if (existing) throw conflict("Provider dispatch has already reserved this run");
|
||||
if (existing) {
|
||||
// The native coordinator fences recovery ownership. Reuse the original
|
||||
// hold only for that live owner, never for an ordinary duplicate dispatch.
|
||||
const [owner] = recoveryLeaseOwner && run.runtimeMode === "native" && existing.state === "held"
|
||||
&& existing.projectId === projectId ? await tx.select({ runId: nativeRunFinalizations.runId })
|
||||
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId),
|
||||
eq(nativeRunFinalizations.runId, runId), eq(nativeRunFinalizations.leaseOwner, recoveryLeaseOwner),
|
||||
sql`${nativeRunFinalizations.leaseExpiresAt} > now()`)).for("update") : [];
|
||||
if (owner) return existing;
|
||||
throw conflict("Provider dispatch has already reserved this run");
|
||||
}
|
||||
const block = await budgetServiceInTransaction(tx, publications).getInvocationBlock(companyId, run.agentId, { projectId });
|
||||
if (block) throw conflict(block.reason);
|
||||
const candidates = await tx.select().from(budgetPolicies).where(and(eq(budgetPolicies.companyId, companyId), eq(budgetPolicies.isActive, true), eq(budgetPolicies.hardStopEnabled, true)));
|
||||
|
||||
@@ -611,8 +611,6 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
if (input.reservationCents !== undefined && compareCents(input.reservationCents, 0) < 0) throw unprocessable("Reservation cannot be negative");
|
||||
const metric = input.metric ?? "billed_cents";
|
||||
const windowKind = input.windowKind ?? (input.scopeType === "project" ? "lifetime" : "calendar_month_utc");
|
||||
const amount = Math.max(0, Math.floor(input.amount));
|
||||
const nextIsActive = amount > 0 && (input.isActive ?? true);
|
||||
const existing = await db
|
||||
.select()
|
||||
.from(budgetPolicies)
|
||||
@@ -627,6 +625,9 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (!existing && input.amount === undefined) throw unprocessable("Amount is required for a new budget policy");
|
||||
const amount = Math.max(0, Math.floor(input.amount ?? existing!.amount));
|
||||
const nextIsActive = amount > 0 && (input.isActive ?? (input.amount !== undefined ? true : existing?.isActive ?? true));
|
||||
const now = new Date();
|
||||
const row = existing
|
||||
? await db
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { publishAccountingActivities } from "./accounting-transaction.js";
|
||||
import { budgetServiceInTransaction } from "./budgets.js";
|
||||
import { budgetServiceInTransaction, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js";
|
||||
import { and, count, eq, gte, inArray, isNull, lt, notInArray, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
@@ -68,7 +68,7 @@ const SYSTEM_COMPANY_ACTOR: CompanyActivityActor = {
|
||||
runId: null,
|
||||
};
|
||||
|
||||
export function companyService(db: Db) {
|
||||
export function companyService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
|
||||
const environmentsSvc = environmentService(db);
|
||||
const heartbeat = heartbeatService(db);
|
||||
const builtInAgents = builtInAgentService(db);
|
||||
@@ -462,6 +462,7 @@ export function companyService(db: Db) {
|
||||
});
|
||||
if (!result) return null;
|
||||
publishAccountingActivities(id, budgetPublications);
|
||||
if (data.budgetMonthlyCents !== undefined) await deliverBudgetEnforcement(db, budgetHooks, id);
|
||||
// Post-commit, fire-and-forget, and BEFORE any finalization that
|
||||
// could throw: a Cloud-pinned primary company that crossed the
|
||||
// archived boundary (either direction) rings the harness so the
|
||||
|
||||
@@ -24657,7 +24657,7 @@ export function heartbeatService(
|
||||
const guardedDispatch =
|
||||
await dispatchResolvedInteractionContinuationWithAtomicGate(
|
||||
async (markDispatchStarted) => {
|
||||
await reserveRunBudget(db, run.companyId, run.id, readNonEmptyString(runLedgerScope.projectId), runLedgerScope);
|
||||
await reserveRunBudget(db, run.companyId, run.id, readNonEmptyString(runLedgerScope.projectId), runLedgerScope, runOptions.nativeLeaseOwner);
|
||||
return executePaperclipNativeSession({
|
||||
db,
|
||||
execution: nativeExecution,
|
||||
|
||||
@@ -35,7 +35,8 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId
|
||||
const base = { provider: row.summary.provider, accountKey, accountLabel: row.summary.name, source: "managed-connection" };
|
||||
const definitionUnavailable = row.grant.credentialSecretRefs.some(ref => {
|
||||
const revision = revisions.find(value => value.id === ref.secretId);
|
||||
return revision?.definitionId && (revision.definitionStatus !== "active" || revision.definitionDeletedAt !== null);
|
||||
return revision && (revision.status !== "active" || revision.versionStatus === "disabled" || revision.revokedAt != null
|
||||
|| (revision.definitionId && (revision.definitionStatus !== "active" || revision.definitionDeletedAt !== null)));
|
||||
});
|
||||
if (row.summary.status !== "connected" || definitionUnavailable) {
|
||||
return { ...base, ok: false, errorFamily: "credentials_unavailable", error: publicError, windows: [] };
|
||||
@@ -52,23 +53,34 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId
|
||||
credentialResolved = true;
|
||||
let windows;
|
||||
if (base.provider === "openai") {
|
||||
const auth = JSON.parse(value);
|
||||
let auth;
|
||||
try { auth = JSON.parse(value); } catch { throw new Error("credentials_unavailable"); }
|
||||
const token = auth.tokens?.access_token ?? auth.accessToken;
|
||||
const accountId = auth.tokens?.account_id ?? auth.accountId;
|
||||
if (typeof token !== "string" || !token || typeof accountId !== "string" || !accountId) {
|
||||
if (typeof token !== "string" || !token) {
|
||||
return { ...base, ok: false, errorFamily: "credentials_unavailable", error: publicError, windows: [] };
|
||||
}
|
||||
windows = await fetchCodexQuota(token, accountId, controller.signal);
|
||||
const read = (auth: { tokens?: { access_token?: string; account_id?: string | null }; accessToken?: string; accountId?: string | null }) =>
|
||||
fetchCodexQuota(auth.tokens?.access_token ?? auth.accessToken!, auth.tokens?.account_id ?? auth.accountId ?? null, controller.signal);
|
||||
try { windows = await read(auth); }
|
||||
catch (error) {
|
||||
if (!(error instanceof Error) || !/\b401\b/.test(error.message)) throw error;
|
||||
auth = JSON.parse(await service.refreshQuotaCredential(row, value, controller.signal));
|
||||
controller.signal.throwIfAborted();
|
||||
windows = await read(auth);
|
||||
}
|
||||
} else {
|
||||
windows = await fetchClaudeQuota(value, controller.signal);
|
||||
}
|
||||
return { ...base, ok: true, windows, capturedAt: new Date().toISOString() };
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : "Quota unavailable";
|
||||
const invalid = !credentialResolved || /401|403|refresh_token_(?:reused|expired|invalidated)/i.test(message);
|
||||
const missing = message === "credentials_unavailable";
|
||||
const invalid = message === "authentication_required" || (credentialResolved && /\b401\b|refresh_token_(?:reused|expired|invalidated)/i.test(message));
|
||||
const errorFamily = missing ? "credentials_unavailable" : invalid ? "authentication_required"
|
||||
: credentialResolved && /\b403\b/.test(message) ? "permission_denied" : "provider_unavailable";
|
||||
logger.warn({ companyId, accountKey, provider: base.provider, authenticationFailed: invalid }, "Connected account quota unavailable");
|
||||
return { ...base, ok: false, windows: [], error: publicError,
|
||||
...(invalid ? { errorFamily: "authentication_required" } : {}) };
|
||||
errorFamily };
|
||||
}
|
||||
})();
|
||||
const bounded = withQuotaTimeout(base.provider, result, () => controller.abort()).then(quota => ({ ...base, ...quota, ...(quota.ok ? {} : { error: publicError }) }));
|
||||
|
||||
@@ -50,6 +50,10 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
|
||||
const billingType = text(usage.billingType) ?? "unknown";
|
||||
const costUnits = billingType === "subscription_included" ? 0n : usdToUnits(costUsd ?? 0);
|
||||
const costCents = unitsToCents(costUnits);
|
||||
const [agent] = await tx.select({ adapterType: agents.adapterType }).from(agents).where(eq(agents.id, run.agentId));
|
||||
const hasBillableEvidence = agent?.adapterType !== "process" || costUsd !== null || usage.costStatus === "unpriced"
|
||||
|| ["inputTokens", "cachedInputTokens", "outputTokens"].some(key => (amount(usage[key]) ?? 0) > 0)
|
||||
|| (text(usage.provider) !== null && usage.provider !== "unknown");
|
||||
const receipt = {
|
||||
idempotencyKey: `heartbeat:${run.id}:final`,
|
||||
agentId: run.agentId, heartbeatRunId: run.id,
|
||||
@@ -58,7 +62,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
|
||||
providerRequestId: text(usage.providerRequestId),
|
||||
pricingProvenance: object(usage.pricingProvenance),
|
||||
model: text(usage.model) ?? "unknown", billingType,
|
||||
costStatus: costUsd === null && billingType !== "subscription_included" ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported",
|
||||
costStatus: costUsd === null && billingType !== "subscription_included" && hasBillableEvidence ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported",
|
||||
inputTokens: amount(usage.inputTokens) ?? 0, cachedInputTokens: amount(usage.cachedInputTokens) ?? 0, outputTokens: amount(usage.outputTokens) ?? 0,
|
||||
costCents, occurredAt: run.finishedAt ?? run.createdAt,
|
||||
};
|
||||
@@ -93,7 +97,6 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
|
||||
outputTokens: sql<string>`coalesce(sum(${costEvents.outputTokens}),0)::text`,
|
||||
costCents: sql<string>`coalesce(sum(${costEvents.costCents}),0)::text`,
|
||||
}).from(costEvents).where(and(eq(costEvents.companyId, run.companyId), eq(costEvents.heartbeatRunId, run.id)));
|
||||
const [agent] = await tx.select({ adapterType: agents.adapterType }).from(agents).where(eq(agents.id, run.agentId));
|
||||
await tx.insert(agentRuntimeState).values({ agentId: run.agentId, companyId: run.companyId, adapterType: agent.adapterType }).onConflictDoNothing();
|
||||
await tx.execute(sql`insert into accounting_runtime_baselines (agent_id,company_id,cost_cents,input_tokens,cached_input_tokens,output_tokens)
|
||||
select agent_id,company_id,total_cost_cents,total_input_tokens,total_cached_input_tokens,total_output_tokens
|
||||
|
||||
@@ -179,7 +179,7 @@ export function statusCardService(
|
||||
: Promise.resolve(null),
|
||||
db.select({
|
||||
tokens: sql<number>`coalesce(sum(coalesce(${statusCardUpdates.inputTokens}, 0) + coalesce(${statusCardUpdates.outputTokens}, 0)), 0)::int`,
|
||||
costCents: sql<number>`coalesce(sum(${statusCardUpdates.costCents}), 0)::int`,
|
||||
costCents: sql<number>`coalesce(sum(${statusCardUpdates.costCents}), 0)::double precision`,
|
||||
})
|
||||
.from(statusCardUpdates)
|
||||
.where(and(eq(statusCardUpdates.cardId, card.id), gte(statusCardUpdates.startedAt, dayStart)))
|
||||
@@ -877,7 +877,7 @@ export function statusCardService(
|
||||
? await tx.select({
|
||||
inputTokens: sql<number>`coalesce(sum(${costEvents.inputTokens}), 0)::int`,
|
||||
outputTokens: sql<number>`coalesce(sum(${costEvents.outputTokens}), 0)::int`,
|
||||
costCents: sql<number>`coalesce(sum(${costEvents.costCents}), 0)::double precision`,
|
||||
costCents: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
||||
}).from(costEvents).where(eq(costEvents.heartbeatRunId, actor.runId))
|
||||
: [];
|
||||
const existingUpdate = await tx.select().from(statusCardUpdates)
|
||||
@@ -892,7 +892,7 @@ export function statusCardService(
|
||||
changeSummary: input.changeSummary,
|
||||
inputTokens: Number(usage[0]?.inputTokens ?? 0),
|
||||
outputTokens: Number(usage[0]?.outputTokens ?? 0),
|
||||
costCents: Number(usage[0]?.costCents ?? 0),
|
||||
costCents: sql`${usage[0]?.costCents ?? "0"}::numeric`,
|
||||
};
|
||||
if (existingUpdate) {
|
||||
await tx.update(statusCardUpdates).set(updateValues).where(eq(statusCardUpdates.id, existingUpdate.id));
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import { renderToStaticMarkup } from "react-dom/server";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { AccountQuotaPanels } from "./AccountQuotaPanels";
|
||||
|
||||
describe("account quota panels", () => {
|
||||
it("renders both accounts for one provider without merging their windows", () => {
|
||||
const html = renderToStaticMarkup(<AccountQuotaPanels accounts={[
|
||||
{ provider: "openai", accountKey: "a", accountLabel: "Personal", ok: true, windows: [{ label: "Personal limit", usedPercent: 25, resetsAt: null, valueLabel: null }] },
|
||||
{ provider: "openai", accountKey: "b", accountLabel: "Shared", ok: true, windows: [{ label: "Shared limit", usedPercent: 75, resetsAt: null, valueLabel: null }] },
|
||||
]} />);
|
||||
expect(html).toContain("Personal limit");
|
||||
expect(html).toContain("Shared limit");
|
||||
expect(html).toContain("25% used");
|
||||
expect(html).toContain("75% used");
|
||||
});
|
||||
it("displays unknown utilization without inventing zero percent", () => {
|
||||
const html = renderToStaticMarkup(<AccountQuotaPanels accounts={[
|
||||
{ provider: "openai", accountKey: "a", ok: true, windows: [{ label: "Credits", usedPercent: null, resetsAt: null, valueLabel: "$5 remaining" }] },
|
||||
]} />);
|
||||
expect(html).toContain("$5 remaining");
|
||||
expect(html).not.toContain("0%");
|
||||
expect(html).not.toContain('role="progressbar"');
|
||||
});
|
||||
});
|
||||
@@ -26,11 +26,16 @@ export function AccountQuotaPanels({
|
||||
: quotaUnavailableMessage(account.windows.length > 0)}
|
||||
</p>
|
||||
)}
|
||||
{account.windows.map((window, index) => (
|
||||
{account.windows.map((window, index) => window.usedPercent == null ? (
|
||||
<div key={`${window.label}:${index}`} className="flex justify-between gap-3 text-sm">
|
||||
<span>{window.label}</span>
|
||||
<span className="text-muted-foreground">{window.valueLabel ?? "Usage not reported"}</span>
|
||||
</div>
|
||||
) : (
|
||||
<QuotaBar
|
||||
key={`${window.label}:${index}`}
|
||||
label={window.label}
|
||||
percentUsed={window.usedPercent ?? 0}
|
||||
percentUsed={window.usedPercent}
|
||||
leftLabel={
|
||||
window.valueLabel ??
|
||||
(window.usedPercent === null
|
||||
|
||||
@@ -37,6 +37,7 @@ interface ProviderQuotaCardProps {
|
||||
quotaError?: string | null;
|
||||
quotaSource?: string | null;
|
||||
quotaLoading?: boolean;
|
||||
quotaAccounts?: ProviderQuotaResult[];
|
||||
}
|
||||
|
||||
export function ProviderQuotaCard({
|
||||
@@ -53,6 +54,7 @@ export function ProviderQuotaCard({
|
||||
quotaError = null,
|
||||
quotaSource = null,
|
||||
quotaLoading = false,
|
||||
quotaAccounts,
|
||||
}: ProviderQuotaCardProps) {
|
||||
// single-pass aggregation over rows — memoized so the 8 derived values are not
|
||||
// recomputed on every parent render tick (providers tab polls every 30s, and each
|
||||
|
||||
Reference in new issue
Block a user