From e8f389beb6911229c477b8e73d022dbf5341f727 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 16:52:55 -0700 Subject: [PATCH 1/9] test: exercise authored-file discovery through a real Git index Co-Authored-By: Paperclip --- .../__tests__/cli-invocation-safety.test.ts | 30 +++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/server/src/__tests__/cli-invocation-safety.test.ts b/server/src/__tests__/cli-invocation-safety.test.ts index abd8a68fe3b..e465e7a57e9 100644 --- a/server/src/__tests__/cli-invocation-safety.test.ts +++ b/server/src/__tests__/cli-invocation-safety.test.ts @@ -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( From 5825f92c62ad61161a2ef21ddf9d31f677558173 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:01:49 -0700 Subject: [PATCH 2/9] fix: retain quota caches across reads and cancel expired probes Co-Authored-By: Paperclip --- .../adapters/claude-local/src/server/quota.ts | 5 +-- .../adapters/codex-local/src/server/quota.ts | 5 +-- server/src/__tests__/ai-connections.test.ts | 33 ++++++++++++++++++- .../__tests__/company-quota-windows.test.ts | 33 +++++++++++++++++-- server/src/__tests__/quota-windows.test.ts | 23 +++++++++++++ server/src/services/quota-windows.ts | 16 +++++---- 6 files changed, 102 insertions(+), 13 deletions(-) diff --git a/packages/adapters/claude-local/src/server/quota.ts b/packages/adapters/claude-local/src/server/quota.ts index 44d4cc633f6..4774acbd710 100644 --- a/packages/adapters/claude-local/src/server/quota.ts +++ b/packages/adapters/claude-local/src/server/quota.ts @@ -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 { +export async function fetchClaudeQuota(token: string, signal?: AbortSignal): Promise { const resp = await fetchWithTimeout("https://api.anthropic.com/api/oauth/usage", { + signal, headers: { Authorization: `Bearer ${token}`, "anthropic-beta": "oauth-2025-04-20", diff --git a/packages/adapters/codex-local/src/server/quota.ts b/packages/adapters/codex-local/src/server/quota.ts index a81995801d6..cb2407057ff 100644 --- a/packages/adapters/codex-local/src/server/quota.ts +++ b/packages/adapters/codex-local/src/server/quota.ts @@ -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 { const headers: Record = { 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); diff --git a/server/src/__tests__/ai-connections.test.ts b/server/src/__tests__/ai-connections.test.ts index 93417627188..8be04f43fba 100644 --- a/server/src/__tests__/ai-connections.test.ts +++ b/server/src/__tests__/ai-connections.test.ts @@ -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" }); diff --git a/server/src/__tests__/company-quota-windows.test.ts b/server/src/__tests__/company-quota-windows.test.ts index d73bb3c5cff..dd4da0a056c 100644 --- a/server/src/__tests__/company-quota-windows.test.ts +++ b/server/src/__tests__/company-quota-windows.test.ts @@ -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((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}`)), diff --git a/server/src/__tests__/quota-windows.test.ts b/server/src/__tests__/quota-windows.test.ts index 31d5e70ec0c..b56a99f7321 100644 --- a/server/src/__tests__/quota-windows.test.ts +++ b/server/src/__tests__/quota-windows.test.ts @@ -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((_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(); diff --git a/server/src/services/quota-windows.ts b/server/src/services/quota-windows.ts index ef3d9e697b8..a897f59904e 100644 --- a/server/src/services/quota-windows.ts +++ b/server/src/services/quota-windows.ts @@ -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 { 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 => { 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 { async function withQuotaTimeout( adapterType: string, task: Promise, + onTimeout?: () => void, ): Promise { let timeoutId: NodeJS.Timeout | null = null; try { @@ -137,6 +140,7 @@ async function withQuotaTimeout( task, new Promise((resolve) => { timeoutId = setTimeout(() => { + onTimeout?.(); resolve({ provider: providerSlugForAdapterType(adapterType), ok: false, From 3ac5772c66bceaa5115c2b04c31d3f5afe5db486 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:15:39 -0700 Subject: [PATCH 3/9] fix: isolate budget recovery and account supplemental run charges Co-Authored-By: Paperclip --- cli/src/__tests__/worktree.test.ts | 12 +++---- doc/COST-ACCOUNTING.md | 2 ++ .../cost-accounting-operators.test.ts | 34 +++++++++++++++++++ .../cost-accounting-reliability.test.ts | 29 ++++++++++++++++ server/src/services/accounting-integrity.ts | 4 +-- server/src/services/budgets.ts | 27 ++++++++------- server/src/services/costs.ts | 17 +++++++++- server/src/services/recovery/service.ts | 6 ++++ server/src/services/run-cost-accounting.ts | 23 +++++++------ 9 files changed, 121 insertions(+), 33 deletions(-) diff --git a/cli/src/__tests__/worktree.test.ts b/cli/src/__tests__/worktree.test.ts index 0fb6b2aee32..a37449a774c 100644 --- a/cli/src/__tests__/worktree.test.ts +++ b/cli/src/__tests__/worktree.test.ts @@ -157,12 +157,12 @@ async function seedValidWorktreeSource( userId, role: "instance_admin", }); - await db.insert(companies).values({ - id: companyId, - name: "Seed Source", - issuePrefix: "SEED", - requireBoardApprovalForNewAgents: false, - }); + // Seed only columns present in the historical source schema. The current + // model includes accounting columns that the worktree migration adds later. + await db.$client` + insert into companies (id, name, issue_prefix, require_board_approval_for_new_agents) + values (${companyId}, 'Seed Source', 'SEED', false) + `; await db.insert(companyMemberships).values({ companyId, principalType: "user", diff --git a/doc/COST-ACCOUNTING.md b/doc/COST-ACCOUNTING.md index 085824748c9..172bfd9abde 100644 --- a/doc/COST-ACCOUNTING.md +++ b/doc/COST-ACCOUNTING.md @@ -153,3 +153,5 @@ Company quota requests resolve the caller's accessible subscription AI connectio Responses carry an opaque account identity and successful capture timestamp. The 30-second cache is scoped to company, user, connection, grant and credential revision; authorization and revision metadata are checked before reading cached results. Transient failures retain the last successful UI observation with its original timestamp. Rotation, revocation and explicit authentication rejection clear obsolete windows. Public messages never contain provider subprocess commands or raw diagnostics. 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. diff --git a/server/src/__tests__/cost-accounting-operators.test.ts b/server/src/__tests__/cost-accounting-operators.test.ts index b3c32714706..e47565f1894 100644 --- a/server/src/__tests__/cost-accounting-operators.test.ts +++ b/server/src/__tests__/cost-accounting-operators.test.ts @@ -86,6 +86,40 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await service.inspect(f.company.id)).findings).toEqual([]); }); + it.each(["before", "after", "concurrent"])("accounts separate run charges %s the provider receipt without false integrity findings", async (timing) => { + const f = await fixture(); const run = await runFor(f, "succeeded"); + await db.update(heartbeatRuns).set({ costAccountingPending: true, usageJson: { costUsdExact: "0.01", inputTokens: 7, cachedInputTokens: 3, outputTokens: 11 } }).where(eq(heartbeatRuns.id, run.id)); + const extra = { agentId: f.agent.id, heartbeatRunId: run.id, provider: "test", model: "tool", costCents: "0.1234567", inputTokens: 2, cachedInputTokens: 4, outputTokens: 6, occurredAt: new Date(), idempotencyKey: "extra-charge" }; + const costs = costService(db); + if (timing === "before") await costs.createEvent(f.company.id, extra); + if (timing === "concurrent") await Promise.all([costs.createEvent(f.company.id, extra), accountRunCost(db, run.id)]); + else await accountRunCost(db, run.id); + const event = await costs.createEvent(f.company.id, extra); + // Replaying either source cannot count it twice. + await costs.createEvent(f.company.id, extra); + expect(await accountRunCost(db, run.id)).toBe(false); + const [runtime] = await db.select().from(agentRuntimeState).where(eq(agentRuntimeState.agentId, f.agent.id)); + expect(runtime).toMatchObject({ totalCostCents: 1.1234567, totalInputTokens: 9, totalCachedInputTokens: 7, totalOutputTokens: 17 }); + const integrity = accountingIntegrityService(db); + expect((await integrity.inspect(f.company.id)).findings).toEqual([]); + await billingReconciliationService(db).adjust(f.company.id, event.id, { idempotencyKey: "extra-correction", expectedCents: "0.1234567", correctedCents: "0.2234567", reason: "Provider correction", pricing: { source: "provider_invoice", evidence: "external charge" } }, "board"); + expect((await costs.summary(f.company.id)).spendCentsExact).toBe("1.2234567"); + expect((await integrity.inspect(f.company.id)).findings).toEqual([]); + await db.update(agentRuntimeState).set({ totalCostCents: 999 }).where(eq(agentRuntimeState.agentId, f.agent.id)); + const review = await integrity.inspect(f.company.id); + expect(review.findings.map(f => f.kind)).toEqual(["runtime_projection"]); + expect((await integrity.repair(f.company.id, review.fingerprint, "Fix drift", "board")).findings).toEqual([]); + // An extra charge cannot conceal a missing original provider receipt. + await db.delete(costEvents).where(and(eq(costEvents.heartbeatRunId, run.id), eq(costEvents.idempotencyKey, `heartbeat:${run.id}:final`))); + expect((await integrity.inspect(f.company.id)).findings.some(f => f.kind === "missing_receipt")).toBe(true); + }); + + it("rejects public charges that impersonate the provider receipt namespace", async () => { + const f = await fixture(); + await expect(costService(db).createEvent(f.company.id, { agentId: f.agent.id, provider: "test", model: "test", costCents: 1, occurredAt: new Date(), idempotencyKey: "heartbeat:fake:final" })).rejects.toThrow("reserved"); + expect(await db.select().from(costEvents).where(eq(costEvents.companyId, f.company.id))).toEqual([]); + }); + it("imports invoices idempotently and appends reviewed corrections without losing the provider receipt", async () => { const f = await fixture(); const originalPricing = { source: "rate_card" as const, version: "2026-09-01", inputCentsPerMillion: "12.3456789" }; diff --git a/server/src/__tests__/cost-accounting-reliability.test.ts b/server/src/__tests__/cost-accounting-reliability.test.ts index e3a1153d422..022de3e1948 100644 --- a/server/src/__tests__/cost-accounting-reliability.test.ts +++ b/server/src/__tests__/cost-accounting-reliability.test.ts @@ -125,6 +125,35 @@ databaseDescribe("cost accounting reliability (PostgreSQL)", () => { expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).toMatchObject({ scopeType: "agent" }); }); + it("recovers live scopes despite deleted agent and project policies and an isolated transaction failure", async () => { + const f = await fixture(); const other = await fixture(); const budgets = budgetService(db); + const [removedAgent] = await db.insert(agents).values({ companyId: f.company.id, name: "Removed", adapterType: "process" }).returning(); + const [removedProject] = await db.insert(projects).values({ companyId: f.company.id, name: "Removed" }).returning(); + await budgets.upsertPolicy(f.company.id, { scopeType: "agent", scopeId: removedAgent.id, amount: 100 }, "board"); + await budgets.upsertPolicy(f.company.id, { scopeType: "project", scopeId: removedProject.id, amount: 100 }, "board"); + await db.delete(agents).where(eq(agents.id, removedAgent.id)); + await db.delete(projects).where(eq(projects.id, removedProject.id)); + for (const live of [f, other]) { + await budgets.upsertPolicy(live.company.id, { scopeType: "agent", scopeId: live.agent.id, amount: 100 }, "board"); + await db.update(agents).set({ status: "paused", pauseReason: "budget" }).where(eq(agents.id, live.agent.id)); + } + // Restrict the scanned scope list so the injected first transaction fails + // on the stale scope, ahead of both live policies in this real database. + const query = vi.spyOn(db, "selectDistinct").mockReturnValueOnce({ from: () => ({ orderBy: async () => [ + { companyId: f.company.id, scopeType: "project", scopeId: removedProject.id }, + { companyId: f.company.id, scopeType: "agent", scopeId: removedAgent.id }, + ...[f, other].map(live => ({ companyId: live.company.id, scopeType: "agent", scopeId: live.agent.id })), + ] }) } as never); + const transaction = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("Injected scope recovery failure")); + try { await budgets.reconcilePolicies(); } finally { query.mockRestore(); transaction.mockRestore(); } + for (const live of [f, other]) { + const [agent] = await db.select().from(agents).where(eq(agents.id, live.agent.id)); + expect(agent).toMatchObject({ status: "idle", pauseReason: null }); + } + // A subsequent sweep uses actual policy discovery and skips both deleted scopes. + await expect(budgets.reconcilePolicies()).resolves.toBeUndefined(); + }); + it("recovers terminal run accounting exactly once across simultaneous live and recovery workers", async () => { const f = await fixture(); await db.update(heartbeatRuns).set({ status: "timed_out", finishedAt: new Date(), costAccountingPending: true, diff --git a/server/src/services/accounting-integrity.ts b/server/src/services/accounting-integrity.ts index c5b1709e71e..2e03fdd8e53 100644 --- a/server/src/services/accounting-integrity.ts +++ b/server/src/services/accounting-integrity.ts @@ -38,7 +38,7 @@ async function inspect(tx: Db, companyId: string): Promise } const incomplete = await tx.execute<{ id: string; acknowledged: boolean; events: number }>(sql` select r.id, r.cost_accounted_at is not null as acknowledged, count(e.id)::int as events from heartbeat_runs r - left join cost_events e on e.heartbeat_run_id = r.id and e.company_id = r.company_id + left join cost_events e on e.heartbeat_run_id = r.id and e.company_id = r.company_id and e.idempotency_key like ('heartbeat:' || r.id::text || ':%') where r.company_id = ${companyId} and r.status in ('succeeded','failed','timed_out','cancelled') and (r.cost_accounting_pending or r.accounting_projection_version = 'v2') and r.result_json->'executionRecovery'->>'providerWorkStarted' is distinct from 'false' @@ -52,7 +52,7 @@ async function inspect(tx: Db, companyId: string): Promise const receipts = await tx.execute<{ id: string; usage: Record; actual: Record }>(sql` select r.id, r.usage_json as usage, jsonb_build_object('cents', sum(coalesce(e.reported_cost_cents,e.cost_cents))::text, 'input', sum(e.input_tokens)::text, 'cached', sum(e.cached_input_tokens)::text, 'output', sum(e.output_tokens)::text) as actual - from heartbeat_runs r join cost_events e on e.heartbeat_run_id = r.id and e.company_id = r.company_id + from heartbeat_runs r join cost_events e on e.heartbeat_run_id = r.id and e.company_id = r.company_id and e.idempotency_key like ('heartbeat:' || r.id::text || ':%') where r.company_id = ${companyId} and r.accounting_projection_version = 'v2' and r.cost_accounted_at is not null group by r.id order by r.id`); for (const row of receipts) { const usage = row.usage ?? {}; diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 4767a99b519..b26ffe092e2 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -24,7 +24,7 @@ import type { BudgetThresholdType, BudgetWindowKind, } from "@paperclipai/shared"; -import { notFound, unprocessable } from "../errors.js"; +import { HttpError, notFound, unprocessable } from "../errors.js"; import { logger } from "../middleware/logger.js"; import { withAccountingTransaction } from "./accounting-transaction.js"; import { logActivity, type LogActivityInput, type ActivityPublication } from "./activity-log.js"; @@ -586,16 +586,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } return { - reconcileCompanyScopes: async (companyId: string) => { - const policies = await listPolicyRows(companyId); - const seen = new Set(); - for (const policy of policies) { - const key = `${policy.scopeType}:${policy.scopeId}`; - if (seen.has(key)) continue; - seen.add(key); - await reconcileScope(companyId, policy.scopeType as BudgetScopeType, policy.scopeId); - } - }, + reconcileScope, listPolicies: async (companyId: string): Promise => { const rows = await listPolicyRows(companyId); return rows.map((row) => ({ @@ -982,8 +973,18 @@ export function budgetService(db: Db, hooks: BudgetServiceHooks = {}) { ...reads, deliverPendingEnforcement: (companyId?: string) => deliverBudgetEnforcement(db, hooks, companyId), reconcilePolicies: async () => { - const rows = await db.selectDistinct({ companyId: budgetPolicies.companyId }).from(budgetPolicies); - for (const row of rows) await mutate(row.companyId, (service) => service.reconcileCompanyScopes(row.companyId)); + const scopes = await db.selectDistinct({ companyId: budgetPolicies.companyId, scopeType: budgetPolicies.scopeType, scopeId: budgetPolicies.scopeId }) + .from(budgetPolicies).orderBy(budgetPolicies.companyId, budgetPolicies.scopeType, budgetPolicies.scopeId); + // Each scope commits independently: deleted targets and a failed scope + // must not roll back recovery or cancellation delivery for other work. + for (const scope of scopes) { + try { + await mutate(scope.companyId, (service) => service.reconcileScope(scope.companyId, scope.scopeType as BudgetScopeType, scope.scopeId)); + } catch (error) { + if (error instanceof HttpError && error.status === 404) continue; + logger.warn({ err: error, ...scope }, "Budget scope recovery pending; continuing remaining scopes"); + } + } }, upsertPolicy: (companyId: string, input: BudgetPolicyUpsertInput, actorUserId: string | null) => mutate(companyId, (service) => service.upsertPolicy(companyId, input, actorUserId)), diff --git a/server/src/services/costs.ts b/server/src/services/costs.ts index fe1a583900d..431dd5a136a 100644 --- a/server/src/services/costs.ts +++ b/server/src/services/costs.ts @@ -3,7 +3,7 @@ import { agentAvatarUrl, resolveAgentAppearance, createCostEventSchema, normaliz import { and, desc, eq, gte, inArray, isNotNull, isNull, lt, lte, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; -import { activityLog, agents, companies, costEvents, heartbeatRuns, issues, projects, goals } from "@paperclipai/db"; +import { activityLog, agents, agentRuntimeState, companies, costEvents, heartbeatRuns, issues, projects, goals } from "@paperclipai/db"; import { notFound, unprocessable, conflict } from "../errors.js"; import { budgetService, budgetServiceInTransaction, type BudgetServiceHooks } from "./budgets.js"; import { logActivity, type LogActivityInput, type ActivityPublication } from "./activity-log.js"; @@ -119,6 +119,20 @@ export async function createCostEventInTransaction(db: Db, companyId: string, da } const [event] = await db.insert(costEvents).values({ ...values, costCents: sql`${values.costCents}::numeric`, reportedCostCents: values.costCents, id: data.id, companyId, receiptHash }).returning(); await updateMonthlySpendProjections(db, companyId, event.agentId, values.costCents, event.occurredAt); + // Separately reported charges linked to an already-accounted run contribute + // to lifetime totals too. Before acknowledgement, accountRunCost includes + // all of the run's events atomically when it initializes the projection. + if (event.heartbeatRunId) { + const [run] = await db.select({ acknowledged: heartbeatRuns.costAccountedAt, version: heartbeatRuns.accountingProjectionVersion }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, event.heartbeatRunId), eq(heartbeatRuns.companyId, companyId))); + if (run.acknowledged && run.version === "v2") await db.update(agentRuntimeState).set({ + totalCostCents: sql`${agentRuntimeState.totalCostCents} + ${values.costCents}::numeric`, + totalInputTokens: sql`${agentRuntimeState.totalInputTokens} + ${event.inputTokens}`, + totalCachedInputTokens: sql`${agentRuntimeState.totalCachedInputTokens} + ${event.cachedInputTokens}`, + totalOutputTokens: sql`${agentRuntimeState.totalOutputTokens} + ${event.outputTokens}`, + updatedAt: new Date(), + }).where(and(eq(agentRuntimeState.agentId, event.agentId), eq(agentRuntimeState.companyId, companyId))); + } await budgetServiceInTransaction(db, publications).evaluateCostEvent(event); await logActivity(db, { companyId, actorType: actor?.actorType ?? "system", actorId: actor?.actorId ?? "cost_accounting", agentId: actor?.agentId ?? event.agentId, @@ -132,6 +146,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { const budgets = budgetService(db, budgetHooks); return { createEvent: async (companyId: string, data: Omit & { costCents: MoneyInput }, actor?: Pick) => { + if (data.idempotencyKey?.startsWith("heartbeat:")) throw unprocessable("The heartbeat idempotency namespace is reserved for run receipts"); const event = await withAccountingTransaction(db, companyId, (tx, publications) => createCostEventInTransaction(tx, companyId, data, publications, actor)); await budgets.deliverPendingEnforcement(companyId); return event; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 113ea5ce1cd..27bb40f1c24 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -3827,6 +3827,12 @@ export function recoveryService( current.companyId, current.id, ); + // Budget admission may wait behind another recovery worker. Read the + // execution paths after the persisted counter so a newly reserved + // successor cannot be mistaken for permission to schedule the next one. + const currentState = await collectDispositionRepairSourceState(db, { issue: current }); + if (currentState.hasActiveExecutionPath || currentState.hasDurableWaitingPath) return "skipped"; + if (!episode && currentState.fingerprint !== state.fingerprint) return "skipped"; const runAttempt = previousAttempt?.fingerprint === state.fingerprint ? previousAttempt.attempt diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index e4d643f92d5..84b7b2d2f05 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -1,6 +1,6 @@ -import { usdToUnits, unitsToCents, centsToUnits } from "@paperclipai/shared"; +import { usdToUnits, unitsToCents } from "@paperclipai/shared"; import { and, asc, eq, inArray, sql } from "drizzle-orm"; -import { accountingRuntimeBaselines, budgetReservations, agentRuntimeState, agents, heartbeatRuns, issues, projects, type Db } from "@paperclipai/db"; +import { accountingRuntimeBaselines, budgetReservations, agentRuntimeState, agents, costEvents, heartbeatRuns, issues, projects, type Db } from "@paperclipai/db"; import { withAccountingTransaction } from "./accounting-transaction.js"; import { createCostEventInTransaction } from "./costs.js"; import { budgetService, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js"; @@ -69,7 +69,6 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService const validParts = parts.length > 0 && tokensMatch && costMatches && parts.every((part) => text(part.model) && amount(part.costUsd) !== null) && new Set(parts.map((part) => part.model)).size === parts.length; - const events = []; if (validParts) { let allocated = 0n; for (const [index, part] of parts.entries()) { @@ -79,19 +78,21 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService const partUnits = billingType === "subscription_included" ? 0n : index === parts.length - 1 ? remainder : reported < remainder ? reported : remainder; allocated += partUnits; const partCents = unitsToCents(partUnits); - events.push(await createCostEventInTransaction(tx, run.companyId, { + await createCostEventInTransaction(tx, run.companyId, { ...receipt, idempotencyKey: `heartbeat:${run.id}:model:${index}`, model: text(part.model)!, inputTokens: amount(partUsage.inputTokens) ?? 0, cachedInputTokens: amount(partUsage.cachedInputTokens) ?? 0, outputTokens: amount(partUsage.outputTokens) ?? 0, costCents: partCents, - }, publications)); + }, publications); } } else { - events.push(await createCostEventInTransaction(tx, run.companyId, receipt, publications)); + await createCostEventInTransaction(tx, run.companyId, receipt, publications); } - const event = events.reduce((total, part) => ({ - inputTokens: total.inputTokens + part.inputTokens, cachedInputTokens: total.cachedInputTokens + part.cachedInputTokens, - outputTokens: total.outputTokens + part.outputTokens, costCents: total.costCents + centsToUnits(part.costCentsExact), - }), { inputTokens: 0, cachedInputTokens: 0, outputTokens: 0, costCents: 0n }); + const [event] = await tx.select({ + inputTokens: sql`coalesce(sum(${costEvents.inputTokens}),0)::text`, + cachedInputTokens: sql`coalesce(sum(${costEvents.cachedInputTokens}),0)::text`, + outputTokens: sql`coalesce(sum(${costEvents.outputTokens}),0)::text`, + costCents: sql`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) @@ -101,7 +102,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService totalInputTokens: sql`${agentRuntimeState.totalInputTokens} + ${event.inputTokens}`, totalCachedInputTokens: sql`${agentRuntimeState.totalCachedInputTokens} + ${event.cachedInputTokens}`, totalOutputTokens: sql`${agentRuntimeState.totalOutputTokens} + ${event.outputTokens}`, - totalCostCents: sql`${agentRuntimeState.totalCostCents} + ${unitsToCents(event.costCents)}::numeric`, + totalCostCents: sql`${agentRuntimeState.totalCostCents} + ${event.costCents}::numeric`, updatedAt: new Date(), }).where(eq(agentRuntimeState.agentId, run.agentId)); await tx.update(heartbeatRuns).set({ costAccountingPending: false, costAccountedAt: new Date(), accountingProjectionVersion: "v2", accountingLastError: null, accountingLastAttemptAt: new Date() }).where(eq(heartbeatRuns.id, run.id)); From f4100b256e15c4ad78f6d08279185cb9e670efe0 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:27:02 -0700 Subject: [PATCH 4/9] fix: flush attempt receipts and surface checkpoint failures Co-Authored-By: Paperclip --- .../src/usage-checkpoint.test.ts | 56 +++++++++++++++ .../adapter-utils/src/usage-checkpoint.ts | 69 +++++++++++++------ .../claude-local/src/server/execute.ts | 22 +++--- .../codex-local/src/server/execute.ts | 2 + .../cursor-local/src/server/execute.ts | 4 +- .../adapters/cursor-local/src/server/parse.ts | 3 + .../gemini-local/src/server/execute.ts | 11 +-- .../adapters/kimi-local/src/server/execute.ts | 1 + .../opencode-local/src/server/execute.ts | 13 ++-- .../adapters/pi-local/src/server/execute.ts | 11 +-- .../adapters/pi-local/src/server/parse.ts | 3 + .../adapter-cost-failure-paths.test.ts | 46 +++++++++++++ 12 files changed, 198 insertions(+), 43 deletions(-) create mode 100644 packages/adapter-utils/src/usage-checkpoint.test.ts diff --git a/packages/adapter-utils/src/usage-checkpoint.test.ts b/packages/adapter-utils/src/usage-checkpoint.test.ts new file mode 100644 index 00000000000..79d8167460c --- /dev/null +++ b/packages/adapter-utils/src/usage-checkpoint.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it, vi } from "vitest"; +import { createUsageCheckpointLog } from "./usage-checkpoint.js"; +import { runChildProcess } from "./server-utils.js"; +import type { AdapterUsageCheckpoint } from "./types.js"; + +const parse = (stdout: string): AdapterUsageCheckpoint => { + const events = stdout.trim().split("\n").map(line => JSON.parse(line)); + return { costUsd: events.reduce((sum, event) => sum + event.usage.costUsd, 0), complete: false }; +}; +const event = JSON.stringify({ type: "usage", usage: { inputTokens: 2, costUsd: 0.25 }, text: "private content" }); + +describe("usage checkpoint stream", () => { + it("flushes an unterminated Pi price and closes each attempt independently", async () => { + const saved = vi.fn(); const parser = vi.fn(parse); + const first = createUsageCheckpointLog(vi.fn(), saved, parser); + await first("stdout", event); + expect(saved).not.toHaveBeenCalled(); + await first.flush({ complete: true }); + const second = createUsageCheckpointLog(vi.fn(), saved, parse); + await second("stdout", event + "\n"); + await second.flush({ complete: true }); + expect(saved.mock.calls[0][0]).toMatchObject({ costUsd: 0.25, complete: true }); + expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: true }); + expect(saved.mock.calls.at(-1)![0].attemptId).not.toBe(saved.mock.calls[0][0].attemptId); + expect(parser.mock.calls[0][0]).not.toContain("private content"); + }); + + it("surfaces a persistence failure outside the local process's best-effort log path", async () => { + const failure = new Error("Receipt storage unavailable"); + const output = vi.fn(); const saved = vi.fn().mockRejectedValue(failure); + const log = createUsageCheckpointLog(output, saved, parse); + const result = await runChildProcess("checkpoint-test", process.execPath, ["-e", `process.stdout.write(${JSON.stringify(event + "\n")})`], { + cwd: process.cwd(), env: {}, timeoutSec: 5, graceSec: 1, onLog: log, + }); + expect(result.exitCode).toBe(0); + expect(output).toHaveBeenCalledWith("stdout", event + "\n"); + await expect(log.flush()).rejects.toBe(failure); + }); + + it("does not promote partial usage to complete merely because the stream was flushed", async () => { + const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, parse); + await log("stdout", event); await log.flush(); + expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: false }); + }); + + it("bounds replay work by stream size while retaining the exact final total", async () => { + const parser = vi.fn(parse); const saved = vi.fn(); + const log = createUsageCheckpointLog(vi.fn(), saved, parser); + for (let i = 0; i < 1024; i++) await log("stdout", event + "\n"); + await log.flush({ complete: true }); + expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 256, complete: true }); + expect(parser.mock.calls.length).toBeLessThanOrEqual(12); + const finalSize = parser.mock.calls.at(-1)![0].length; + expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThan(3 * finalSize); + }); +}); diff --git a/packages/adapter-utils/src/usage-checkpoint.ts b/packages/adapter-utils/src/usage-checkpoint.ts index 39b05c68a5f..f7ebe701203 100644 --- a/packages/adapter-utils/src/usage-checkpoint.ts +++ b/packages/adapter-utils/src/usage-checkpoint.ts @@ -9,7 +9,7 @@ const fields = new Set([ "cache_read_input_tokens", "cache_creation_input_tokens", "inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens", "cachedInputTokens", "promptTokenCount", "candidatesTokenCount", "cachedContentTokenCount", "reasoning", "cacheRead", "cacheWrite", "prompt", "candidates", "totalTokenCount", "total_tokens", "toolUsePromptTokenCount", "messages", - "thoughtsTokenCount", "inputTokens", "totalTokens", "cached", "thoughts", "total_cost_usd", "cost_usd", "costUSD", + "thoughtsTokenCount", "inputTokens", "totalTokens", "cached", "thoughts", "total_cost_usd", "cost_usd", "costUSD", "costUsd", ]); const stringFields = new Set(["type", "subtype", "role", "id", "session_id", "sessionID", "thread_id", "model", "modelID", "providerID"]); function project(value: unknown, depth = 0, field = ""): unknown { @@ -24,36 +24,63 @@ function project(value: unknown, depth = 0, field = ""): unknown { : project(entry, depth + 1, key)])); } -/** A fresh instance belongs to one CLI attempt. Retries must not concatenate - * cumulative usage from different attempts. Callback durability is awaited - * before ordinary log publication. */ +/** A fresh instance belongs to one CLI attempt. Call flush() after its process + * exits, before retrying or final result handling. Checkpoint failures are + * retained outside runChildProcess's best-effort onLog error handler. */ export function createUsageCheckpointLog( onLog: (stream: "stdout" | "stderr", chunk: string) => Promise, onUsage: ((receipt: AdapterUsageCheckpoint) => Promise) | undefined, parse: (stdout: string) => AdapterUsageCheckpoint | null, ) { - if (!onUsage) return onLog; const attemptId = randomUUID(); let remainder = "", accounting = "", previous = ""; - return async (stream: "stdout" | "stderr", chunk: string) => { - if (stream === "stdout") { - remainder += chunk; - const lines = remainder.split("\n"); remainder = lines.pop() ?? ""; - if (remainder.length > 8 * 1024 * 1024) throw new Error("Accounting protocol line exceeds 8 MiB"); - for (const line of lines) { - let raw: unknown; - try { raw = JSON.parse(line); } catch { continue; } - const compact = JSON.stringify(project(raw)); - if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"/.test(compact)) continue; - accounting += compact + "\n"; - if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB"); - const receipt = parse(accounting); - if (!receipt) continue; - const serialized = JSON.stringify(receipt); - if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; } + let nextReplaySize = 0; + let failure: unknown; + let failed = false; + function retain(line: string) { + let raw: unknown; + try { raw = JSON.parse(line); } catch { return; } + const compact = JSON.stringify(project(raw)); + if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"|"model"|"modelID"/.test(compact)) return; + accounting += compact + "\n"; + if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB"); + } + async function publish(final = false, complete = false) { + if (!onUsage || !accounting || (!final && accounting.length < nextReplaySize)) return; + const parsed = parse(accounting); + // Geometric replay bounds total parsing work by stream size. The first + // observation is saved immediately; flush always saves the final snapshot. + nextReplaySize = accounting.length * 2; + if (!parsed) return; + if (!parsed.complete && parsed.costUsd == null && parsed.costUsdExact == null && + !Object.values(parsed.usage ?? {}).some(value => typeof value === "number" && value > 0)) return; + const receipt = { ...parsed, complete: parsed.complete || complete }; + const serialized = JSON.stringify(receipt); + if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; } + } + const log = async (stream: "stdout" | "stderr", chunk: string) => { + if (onUsage && !failed && stream === "stdout") { + try { + remainder += chunk; + const lines = remainder.split("\n"); remainder = lines.pop() ?? ""; + if (remainder.length > 8 * 1024 * 1024) throw new Error("Accounting protocol line exceeds 8 MiB"); + for (const line of lines) retain(line); + await publish(); + } catch (error) { + failed = true; failure = error; remainder = ""; accounting = ""; } } + // Receipt persistence and ordinary logs have separate failure contracts. + // A failed receipt must not hide the provider's diagnostic output. await onLog(stream, chunk); }; + return Object.assign(log, { + async flush(options: { complete?: boolean } = {}) { + if (failed) throw failure; + if (!onUsage) return; + if (remainder) { retain(remainder); remainder = ""; } + await publish(true, options.complete ?? false); + }, + }); } import { randomUUID } from "node:crypto"; diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 378917ecadb..2f407d05364 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -964,6 +964,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parseClaudeStreamJson(stdout); + return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd, + provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic", + billingType, model: Object.keys(parseObject(parsed.resultJson?.modelUsage)).length > 1 ? "mixed" : parsed.model || model, + usageByModel: claudeModelReceipts(parsed.resultJson?.modelUsage), complete: parsed.resultJson !== null }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -973,12 +980,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseClaudeStreamJson(stdout); - return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd, - provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic", - billingType, model: parsed.model || model, complete: parsed.resultJson !== null }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, terminalResultCleanup: { @@ -987,6 +989,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise 1 ? "mixed" : parsedStream.model || model, + usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage), billingType, costUsd: parsedStream.costUsd, errorMessage: `Timed out after ${timeoutSec}s`, @@ -1091,7 +1096,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise 1 ? "mixed" : parsedStream.model || model, + usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage), billingType, costUsd: parsedStream.costUsd, errorMessage: fallbackErrorMessage, diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index a302bdbba86..2069e55f957 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -1347,6 +1347,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { const parsed = parseCursorJsonl(stdout); const provider = resolveProviderFromModel(model); - return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: false }; + return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: parsed.sawResult }; }); let stdoutLineBuffer = ""; const emitNormalizedStdoutLine = async (rawLine: string) => { @@ -676,6 +676,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise): string | null { export function parseCursorJsonl(stdout: string) { let sessionId: string | null = null; + let sawResult = false; const messages: string[] = []; let errorMessage: string | null = null; let totalCostUsd = 0; @@ -97,6 +98,7 @@ export function parseCursorJsonl(stdout: string) { } if (type === "result") { + sawResult = true; const usageObj = parseObject(event.usage); usage.inputTokens += asNumber( usageObj.input_tokens, @@ -160,6 +162,7 @@ export function parseCursorJsonl(stdout: string) { } return { + sawResult, sessionId, summary: messages.join("\n\n").trim(), usage, diff --git a/packages/adapters/gemini-local/src/server/execute.ts b/packages/adapters/gemini-local/src/server/execute.ts index 1ba97244a2d..500d395a819 100644 --- a/packages/adapters/gemini-local/src/server/execute.ts +++ b/packages/adapters/gemini-local/src/server/execute.ts @@ -643,6 +643,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parseGeminiJsonl(stdout); + return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -651,13 +655,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseGeminiJsonl(stdout); - return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, }); + await accountingLog.flush({ complete: !proc.timedOut && !proc.signal }); return { proc, parsed: parseGeminiJsonl(proc.stdout), @@ -691,6 +693,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parseOpenCodeJsonl(stdout); + const provider = parseModelProvider(model || null); + return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -655,14 +660,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseOpenCodeJsonl(stdout); - const provider = parseModelProvider(model || null); - return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, }); + await accountingLog.flush({ complete: !proc.timedOut && !proc.signal }); return { proc, rawStderr: proc.stderr, @@ -683,6 +685,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parsePiJsonl(stdout); + return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: parsed.sawAgentEnd }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -745,13 +749,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parsePiJsonl(stdout); - return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, }); + await accountingLog.flush({ complete: !proc.timedOut && !proc.signal }); // Flush any remaining buffer content if (stdoutBuffer) { @@ -778,6 +780,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise> | undefined; if (messages && messages.length > 0) { const lastMessage = messages[messages.length - 1]; diff --git a/server/src/__tests__/adapter-cost-failure-paths.test.ts b/server/src/__tests__/adapter-cost-failure-paths.test.ts index 6355240efbb..0ad2c8d0038 100644 --- a/server/src/__tests__/adapter-cost-failure-paths.test.ts +++ b/server/src/__tests__/adapter-cost-failure-paths.test.ts @@ -31,6 +31,51 @@ const cases: Array<{ name: string; execute: () => Promise; event: unkno describe("CLI adapter accounting on timeout", () => { const directories: string[] = []; afterEach(async () => { vi.clearAllMocks(); for (const dir of directories.splice(0)) await rm(dir, { recursive: true, force: true }); }); + it.each(cases.filter(fixture => fixture.name !== "kimi"))("$name flushes its attempt and surfaces checkpoint persistence failures", async (fixture) => { + const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir); + const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 }); + const stdout = JSON.stringify(fixture.event); + processResult.mockImplementation(async (_run, _target, _command, _args, options) => { + // Local process logging catches callback failures. Finalization must + // independently enforce receipt durability despite that contract. + await options.onLog("stdout", stdout).catch(() => {}); + return { exitCode: 0, signal: null, timedOut: false, stdout, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + const onUsage = vi.fn(); + const context: AdapterExecutionContext = { + runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} }, + runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, + config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [], + env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } }, + context: {}, onLog: async () => {}, onUsage, + }; + const execute = await fixture.execute(); + await execute(context); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete: true }); + if (fixture.name === "claude") { + const mixed = [ + { type: "system", subtype: "init", model: "actual-model" }, + { type: "result", total_cost_usd: 0.003, usage: { input_tokens: 3, output_tokens: 3 }, modelUsage: { + "model-a": { inputTokens: 1, outputTokens: 1, costUSD: 0.001 }, + "model-b": { inputTokens: 2, outputTokens: 2, costUSD: 0.002 }, + } }, + ]; + processResult.mockImplementation(async (_run, _target, _command, _args, options) => { + const output = mixed.map(item => JSON.stringify(item)).join("\n"); + await options.onLog("stdout", output).catch(() => {}); + return { exitCode: 0, signal: null, timedOut: false, stdout: output, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + await execute(context); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "mixed", costUsd: 0.003, complete: true, usageByModel: [ + { model: "model-a", costUsd: 0.001 }, { model: "model-b", costUsd: 0.002 }, + ] }); + delete (mixed[1] as { modelUsage?: unknown }).modelUsage; + await execute(context); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "actual-model", complete: true }); + } + onUsage.mockRejectedValue(new Error("Receipt persistence failed")); + await expect(execute(context)).rejects.toThrow("Receipt persistence failed"); + }); it.each(cases)("$name retains observed accounting when the process times out", async (fixture) => { const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir); const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 }); @@ -44,6 +89,7 @@ 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.usageBasis).toBe("per_run"); expect(result.provider).toBeTruthy(); expect(result.billingType).toBeTruthy(); From e1c61054ccfc85fec3ee775da657f8f39b2f7d02 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:33:13 -0700 Subject: [PATCH 5/9] fix: settle interrupted usage and retain estimate provenance Co-Authored-By: Paperclip --- .../cost-accounting-operators.test.ts | 29 +++++++++++++++++-- server/src/services/accounting-integrity.ts | 8 ++--- server/src/services/billing-reconciliation.ts | 2 +- server/src/services/budgets.ts | 2 +- server/src/services/costs.ts | 2 +- server/src/services/run-cost-accounting.ts | 2 +- 6 files changed, 34 insertions(+), 11 deletions(-) diff --git a/server/src/__tests__/cost-accounting-operators.test.ts b/server/src/__tests__/cost-accounting-operators.test.ts index e47565f1894..52c2e96a57f 100644 --- a/server/src/__tests__/cost-accounting-operators.test.ts +++ b/server/src/__tests__/cost-accounting-operators.test.ts @@ -10,7 +10,7 @@ import { billingReconciliationService } from "../services/billing-reconciliation import { budgetService } from "../services/budgets.js"; import { reserveRunBudget } from "../services/budget-reservations.js"; import { costService } from "../services/costs.js"; -import { accountRunCost } from "../services/run-cost-accounting.js"; +import { accountRunCost, reconcileRunCosts } from "../services/run-cost-accounting.js"; import { createRunUsageRecorder, persistUsageReceipt, replayUsageReceipts, spoolUsageReceipt, type UsageReceiptEnvelope } from "../services/usage-receipts.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -65,6 +65,28 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await accountingIntegrityService(db).inspect(f.company.id)).findings).toEqual([]); }); + 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"); + await reserveRunBudget(db, f.company.id, run.id, null); + const usage = { accountingReceiptReady: false, costUsdExact: "0.01", inputTokens: 5, provider: "test", billingType: "metered_api" }; + await db.update(heartbeatRuns).set({ status: "interrupted", usageJson: usage }).where(eq(heartbeatRuns.id, run.id)); + const integrity = accountingIntegrityService(db); + expect((await integrity.health(f.company.id)).pendingRunCount).toBe(1); + expect((await integrity.health(f.company.id)).items[0]).toMatchObject({ runId: run.id, state: "waiting_for_receipt" }); + expect((await costService(db).summary(f.company.id)).pendingRunCount).toBe(1); + expect((await integrity.inspect(f.company.id)).findings.some(f => f.kind === "missing_acknowledgement")).toBe(true); + expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).not.toBeNull(); + expect(await accountRunCost(db, run.id)).toBe(false); + expect((await integrity.health(f.company.id)).heldReservationCents).toBe("10.0000000"); + await db.update(heartbeatRuns).set({ usageJson: { ...usage, accountingReceiptReady: true } }).where(eq(heartbeatRuns.id, run.id)); + await reconcileRunCosts(db); + expect((await costService(db).summary(f.company.id))).toMatchObject({ pendingRunCount: 0, spendCents: 1 }); + expect((await integrity.health(f.company.id)).heldReservationCents).toBe("0.0000000"); + expect((await integrity.inspect(f.company.id)).findings).toEqual([]); + expect(await accountRunCost(db, run.id)).toBe(false); + }); + it("independently detects and repairs corrupted projections with a stale-review guard and audit", async () => { const f = await fixture(); const run = await runFor(f, "succeeded"); await db.update(heartbeatRuns).set({ costAccountingPending: true, usageJson: { accountingReceiptReady: true, costUsdExact: "0.012345678", inputTokens: 7, cachedInputTokens: 3, outputTokens: 11 } }).where(eq(heartbeatRuns.id, run.id)); @@ -163,7 +185,7 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await db.select().from(costAdjustments).where(eq(costAdjustments.companyId, f.company.id))).toEqual([]); }); - it("prices unknown usage from an invoice and updates runtime, admission and independent verification", async () => { + it.each(["provider_invoice", "rate_card"] as const)("prices unknown usage from %s and preserves its estimate status", async (source) => { const f = await fixture(); const run = await runFor(f, "failed"); await budgetService(db).upsertPolicy(f.company.id, { scopeType: "agent", scopeId: f.agent.id, amount: 10 }, "board"); await db.update(heartbeatRuns).set({ costAccountingPending: true, usageJson: { accountingReceiptReady: true, inputTokens: 9, provider: "test" } }).where(eq(heartbeatRuns.id, run.id)); @@ -171,9 +193,10 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await accountingIntegrityService(db).health(f.company.id)).unpricedEventCount).toBe(1); expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).not.toBeNull(); const [event] = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, run.id)); - await billingReconciliationService(db).adjust(f.company.id, event.id, { idempotencyKey: "pricing", expectedCents: 0, correctedCents: "1.0000001", reason: "Verified provider invoice", pricing: { source: "provider_invoice", evidence: "billing-export" } }, "board"); + await billingReconciliationService(db).adjust(f.company.id, event.id, { idempotencyKey: "pricing", expectedCents: 0, correctedCents: "1.0000001", reason: "Verified provider invoice", pricing: { source, evidence: "billing-export" } }, "board"); expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).toBeNull(); expect((await accountingIntegrityService(db).health(f.company.id)).unpricedEventCount).toBe(0); + expect((await costService(db).summary(f.company.id)).estimatedEventCount).toBe(source === "rate_card" ? 1 : 0); expect((await accountingIntegrityService(db).inspect(f.company.id)).findings).toEqual([]); }); diff --git a/server/src/services/accounting-integrity.ts b/server/src/services/accounting-integrity.ts index 2e03fdd8e53..56c0412169b 100644 --- a/server/src/services/accounting-integrity.ts +++ b/server/src/services/accounting-integrity.ts @@ -39,7 +39,7 @@ async function inspect(tx: Db, companyId: string): Promise const incomplete = await tx.execute<{ id: string; acknowledged: boolean; events: number }>(sql` select r.id, r.cost_accounted_at is not null as acknowledged, count(e.id)::int as events from heartbeat_runs r left join cost_events e on e.heartbeat_run_id = r.id and e.company_id = r.company_id and e.idempotency_key like ('heartbeat:' || r.id::text || ':%') - where r.company_id = ${companyId} and r.status in ('succeeded','failed','timed_out','cancelled') + where r.company_id = ${companyId} and r.status in ('succeeded','failed','timed_out','cancelled','interrupted') and (r.cost_accounting_pending or r.accounting_projection_version = 'v2') and r.result_json->'executionRecovery'->>'providerWorkStarted' is distinct from 'false' and r.usage_json->>'accountingProviderWorkStarted' is distinct from 'false' @@ -100,9 +100,9 @@ export function accountingIntegrityService(db: Db, hooks: BudgetServiceHooks = { }, health: (companyId: string): Promise => withAccountingTransaction(db, companyId, async tx => { const [counts] = await tx.execute<{ pending: number; unpriced: number; oldest: string | null; cancellations: number; reserved: string }>(sql` - select (select count(*)::int from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled')) as pending, + select (select count(*)::int from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled','interrupted')) as pending, (select count(*)::int from cost_events where company_id = ${companyId} and cost_status = 'unpriced' and billing_type <> 'subscription_included') as unpriced, - (select min(coalesce(finished_at,created_at))::text from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled')) as oldest, + (select min(coalesce(finished_at,created_at))::text from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled','interrupted')) as oldest, (select count(*)::int from budget_policies where company_id = ${companyId} and enforcement_version > enforcement_delivered_version) as cancellations, (select coalesce(sum(amount_cents),0)::text from budget_reservations where company_id = ${companyId} and state = 'held') as reserved`); const pending = await tx.execute<{ runId: string; agentId: string; state: "waiting_for_receipt" | "retryable"; lastError: string | null; attempts: number; since: string; lastAttemptAt: string | null }>(sql` @@ -110,7 +110,7 @@ export function accountingIntegrityService(db: Db, hooks: BudgetServiceHooks = { and usage_json->>'accountingProviderWorkStarted' is distinct from 'false' and result_json->'executionRecovery'->>'providerWorkStarted' is distinct from 'false' then 'waiting_for_receipt' else 'retryable' end as state, accounting_last_error as "lastError", accounting_attempt_count as attempts, coalesce(finished_at,created_at)::text as since, accounting_last_attempt_at::text as "lastAttemptAt" - from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled') order by coalesce(finished_at,created_at),id limit 100`); + from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled','interrupted') order by coalesce(finished_at,created_at),id limit 100`); const unpriced = await tx.execute<{ runId: string | null; costEventId: string; agentId: string; since: string }>(sql` select heartbeat_run_id as "runId", id as "costEventId", agent_id as "agentId", occurred_at::text as since from cost_events where company_id = ${companyId} and cost_status = 'unpriced' and billing_type <> 'subscription_included' order by occurred_at,id limit 100`); diff --git a/server/src/services/billing-reconciliation.ts b/server/src/services/billing-reconciliation.ts index 21cec4110ed..99e2af8689c 100644 --- a/server/src/services/billing-reconciliation.ts +++ b/server/src/services/billing-reconciliation.ts @@ -117,7 +117,7 @@ export function billingReconciliationService(db: Db, hooks: BudgetServiceHooks = // Original provider value and fingerprint remain immutable. cost_cents // is the effective valuation, backed by this append-only correction. const [updated] = await tx.update(costEvents).set({ costCents: sql`${input.correctedCents}::numeric`, - reportedCostCents: event.event.reportedCostCents ?? normalizeCents(event.exact), costStatus: "reported", pricingProvenance: input.pricing, + reportedCostCents: event.event.reportedCostCents ?? normalizeCents(event.exact), costStatus: input.pricing.source === "rate_card" ? "estimated" : "reported", pricingProvenance: input.pricing, }).where(eq(costEvents.id, eventId)).returning(); await updateMonthlySpendProjections(tx, companyId, updated.agentId, subtractCents(input.correctedCents, input.expectedCents), updated.occurredAt); if (updated.heartbeatRunId) { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index b26ffe092e2..45236d29c4e 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -170,7 +170,7 @@ export async function computeObservedSpend( .where(and(...conditions)); const pendingConditions = [eq(heartbeatRuns.companyId, policy.companyId), eq(heartbeatRuns.costAccountingPending, true), - inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled"])]; + inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"])]; if (policy.scopeType === "agent") pendingConditions.push(eq(heartbeatRuns.agentId, policy.scopeId)); if (policy.scopeType === "project") pendingConditions.push(sql`${heartbeatRuns.usageJson}->'ledgerScope'->>'projectId' = ${policy.scopeId}`); if (policy.windowKind === "calendar_month_utc") { diff --git a/server/src/services/costs.ts b/server/src/services/costs.ts index 431dd5a136a..b25a3e3bb9d 100644 --- a/server/src/services/costs.ts +++ b/server/src/services/costs.ts @@ -178,7 +178,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { .where(and(...conditions)); const pendingConditions = [eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.costAccountingPending, true), - inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled"])]; + inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"])]; if (range?.from) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) >= ${range.from.toISOString()}::timestamptz`); if (range?.to) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) <= ${range.to.toISOString()}::timestamptz`); const [pending] = await db.select({ count: sql`count(*)::int` }).from(heartbeatRuns).where(and(...pendingConditions)); diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 84b7b2d2f05..a04accf0774 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -6,7 +6,7 @@ import { createCostEventInTransaction } from "./costs.js"; import { budgetService, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js"; import { logger } from "../middleware/logger.js"; -const terminalStatuses = ["succeeded", "failed", "timed_out", "cancelled"]; +const terminalStatuses = ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]; const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; const text = (value: unknown) => typeof value === "string" && value.length > 0 ? value : null; const amount = (value: unknown) => typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : null; From 2a811ea1a9b93ee2996b7cf69eeb191242d805fe Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:33:49 -0700 Subject: [PATCH 6/9] test: forward stdin through the Cursor fixture runner Co-Authored-By: Paperclip --- packages/adapters/cursor-local/src/server/execute.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/adapters/cursor-local/src/server/execute.test.ts b/packages/adapters/cursor-local/src/server/execute.test.ts index 913d34cdc40..6aea2414d3d 100644 --- a/packages/adapters/cursor-local/src/server/execute.test.ts +++ b/packages/adapters/cursor-local/src/server/execute.test.ts @@ -288,7 +288,7 @@ printf '%s\\n' '{"type":"result","subtype":"success","session_id":"cursor-sessio finalPreparedCommand = preferredCommandPath; const runtimeEnv = { ...input.env, - PATH: `${path.join(systemHomeDir, ".local", "bin")}${path.delimiter}${input.env.PATH}`, + PATH: `${path.join(systemHomeDir, ".local", "bin")}${path.delimiter}${input.env.PATH ?? process.env.PATH ?? "/usr/bin:/bin"}`, }; await fs.mkdir(path.dirname(preferredCommandPath), { recursive: true }); await fs.writeFile(preferredCommandPath, preferredAgentScript); @@ -312,13 +312,14 @@ printf '%s\\n' '{"type":"result","subtype":"success","session_id":"cursor-sessio // shell seam faithful to that protocol instead of returning empty stdout // for every shell command. const runner = { - execute: async (input: { command: string; args?: string[]; env?: Record }) => { + execute: async (input: { command: string; args?: string[]; env?: Record; stdin?: string }) => { runnerState.commands.push(input.command); // Exercise actual bounded file reads during managed-home restoration; // reporting empty success for every shell command hides missing bytes. return runChildProcess(`cursor-fresh-lease-${runnerState.commands.length}`, input.command, input.args ?? [], { cwd: remoteWorkspace, env: { ...input.env, PATH: `${input.env?.PATH ?? ""}:/usr/bin:/bin` }, + stdin: input.stdin, timeoutSec: 30, graceSec: 5, onLog: async () => {}, From 32501493050743867eeb5e75ed84b1040c27f8f0 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:37:20 -0700 Subject: [PATCH 7/9] fix: invalidate quota caches when personal credentials are disabled Co-Authored-By: Paperclip --- server/src/__tests__/ai-connections.test.ts | 9 ++++++++- server/src/__tests__/company-quota-windows.test.ts | 2 ++ server/src/services/quota-windows.ts | 12 ++++++++---- 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/ai-connections.test.ts b/server/src/__tests__/ai-connections.test.ts index 8be04f43fba..3cb74ed7039 100644 --- a/server/src/__tests__/ai-connections.test.ts +++ b/server/src/__tests__/ai-connections.test.ts @@ -8,7 +8,7 @@ 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, companySecretVersions, 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, userSecretDefinitions, 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"; @@ -69,6 +69,13 @@ describe("managed AI connections", () => { 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(userSecretDefinitions).set({ status: "disabled" }).where(eq(userSecretDefinitions.id, before.userSecretDefinitionId!)); + const disabled = await fetchCompanyQuotaWindows(db, companyId, owner); + expect(disabled[0]).toMatchObject({ ok: false, errorFamily: "credentials_unavailable", windows: [] }); + expect(disabled[0].accountKey).not.toBe(rotated[0].accountKey); + expect(fetchSpy).toHaveBeenCalledTimes(2); + await db.update(userSecretDefinitions).set({ status: "active" }).where(eq(userSecretDefinitions.id, before.userSecretDefinitionId!)); + expect((await fetchCompanyQuotaWindows(db, companyId, owner))[0].ok).toBe(true); 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); diff --git a/server/src/__tests__/company-quota-windows.test.ts b/server/src/__tests__/company-quota-windows.test.ts index dd4da0a056c..d848986790a 100644 --- a/server/src/__tests__/company-quota-windows.test.ts +++ b/server/src/__tests__/company-quota-windows.test.ts @@ -28,6 +28,7 @@ let revision = 1; const db = { select: () => ({ from: () => ({ + leftJoin: () => ({ leftJoin: () => ({ where: async () => [ { @@ -38,6 +39,7 @@ const db = { }, ], }), + }), }), }), } as unknown as Db; diff --git a/server/src/services/quota-windows.ts b/server/src/services/quota-windows.ts index a897f59904e..c290969c96f 100644 --- a/server/src/services/quota-windows.ts +++ b/server/src/services/quota-windows.ts @@ -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, companySecretVersions, type Db } from "@paperclipai/db"; +import { companySecrets, companySecretVersions, userSecretDefinitions, 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,15 +25,19 @@ 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, 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))) : []; + const revisions = secretIds.length ? await db.select({ id: companySecrets.id, latestVersion: companySecrets.latestVersion, status: companySecrets.status, versionStatus: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt, definitionId: companySecrets.userSecretDefinitionId, definitionStatus: userSecretDefinitions.status, definitionDeletedAt: userSecretDefinitions.deletedAt }) + .from(companySecrets).leftJoin(companySecretVersions, and(eq(companySecretVersions.secretId, companySecrets.id), eq(companySecretVersions.version, companySecrets.latestVersion))).leftJoin(userSecretDefinitions, and(eq(userSecretDefinitions.id, companySecrets.userSecretDefinitionId), eq(userSecretDefinitions.companyId, companyId))).where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIds))) : []; async function readAccount(row: (typeof accounts)[number]): Promise { const accountKey = createHash("sha256").update(JSON.stringify([companyId, row.connection.id, row.grant.id, row.connection.updatedAt, row.grant.updatedAt, row.grant.credentialSecretRefs.map(ref => [ref.secretId, revisions.find(r => r.id === ref.secretId)]), ])).digest("hex"); const base = { provider: row.summary.provider, accountKey, accountLabel: row.summary.name, source: "managed-connection" }; - if (row.summary.status !== "connected") { + 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); + }); + if (row.summary.status !== "connected" || definitionUnavailable) { return { ...base, ok: false, errorFamily: "credentials_unavailable", error: publicError, windows: [] }; } const key = `${userId}:${accountKey}`; From 014810c58da90f3a69063187fc05e4da731b916a Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:45:06 -0700 Subject: [PATCH 8/9] fix: checkpoint every usage update with incremental protocol parsers Use provider terminal evidence for completeness; an OpenCode successful EOF closes its step-based stream. Preserve immutable snapshots and propagate partial completeness through final results. Co-Authored-By: Paperclip --- .../src/usage-checkpoint.test.ts | 43 ++- .../adapter-utils/src/usage-checkpoint.ts | 24 +- .../claude-local/src/server/execute.ts | 9 +- .../adapters/claude-local/src/server/parse.ts | 143 ++++---- .../codex-local/src/server/execute.ts | 11 +- .../adapters/codex-local/src/server/parse.ts | 105 +++--- .../cursor-local/src/server/execute.ts | 8 +- .../adapters/cursor-local/src/server/parse.ts | 159 ++++----- .../gemini-local/src/server/execute.ts | 8 +- .../adapters/gemini-local/src/server/parse.ts | 195 ++++++----- .../opencode-local/src/server/execute.ts | 10 +- .../opencode-local/src/server/parse.ts | 99 +++--- .../adapters/pi-local/src/server/execute.ts | 10 +- .../adapters/pi-local/src/server/parse.ts | 321 +++++++++--------- .../adapter-cost-failure-paths.test.ts | 47 ++- 15 files changed, 656 insertions(+), 536 deletions(-) diff --git a/packages/adapter-utils/src/usage-checkpoint.test.ts b/packages/adapter-utils/src/usage-checkpoint.test.ts index 79d8167460c..2b622ef60f0 100644 --- a/packages/adapter-utils/src/usage-checkpoint.test.ts +++ b/packages/adapter-utils/src/usage-checkpoint.test.ts @@ -3,20 +3,24 @@ import { createUsageCheckpointLog } from "./usage-checkpoint.js"; import { runChildProcess } from "./server-utils.js"; import type { AdapterUsageCheckpoint } from "./types.js"; -const parse = (stdout: string): AdapterUsageCheckpoint => { - const events = stdout.trim().split("\n").map(line => JSON.parse(line)); - return { costUsd: events.reduce((sum, event) => sum + event.usage.costUsd, 0), complete: false }; +const createParser = () => { + let costUsd = 0; + return (stdout: string): AdapterUsageCheckpoint => { + const events = stdout.trim().split("\n").map(line => JSON.parse(line)); + costUsd += events.reduce((sum, event) => sum + event.usage.costUsd, 0); + return { costUsd, complete: events.some(event => event.type === "result") }; + }; }; const event = JSON.stringify({ type: "usage", usage: { inputTokens: 2, costUsd: 0.25 }, text: "private content" }); describe("usage checkpoint stream", () => { it("flushes an unterminated Pi price and closes each attempt independently", async () => { - const saved = vi.fn(); const parser = vi.fn(parse); + const saved = vi.fn(); const parser = vi.fn(createParser()); const first = createUsageCheckpointLog(vi.fn(), saved, parser); await first("stdout", event); expect(saved).not.toHaveBeenCalled(); await first.flush({ complete: true }); - const second = createUsageCheckpointLog(vi.fn(), saved, parse); + const second = createUsageCheckpointLog(vi.fn(), saved, createParser()); await second("stdout", event + "\n"); await second.flush({ complete: true }); expect(saved.mock.calls[0][0]).toMatchObject({ costUsd: 0.25, complete: true }); @@ -28,7 +32,7 @@ describe("usage checkpoint stream", () => { it("surfaces a persistence failure outside the local process's best-effort log path", async () => { const failure = new Error("Receipt storage unavailable"); const output = vi.fn(); const saved = vi.fn().mockRejectedValue(failure); - const log = createUsageCheckpointLog(output, saved, parse); + const log = createUsageCheckpointLog(output, saved, createParser()); const result = await runChildProcess("checkpoint-test", process.execPath, ["-e", `process.stdout.write(${JSON.stringify(event + "\n")})`], { cwd: process.cwd(), env: {}, timeoutSec: 5, graceSec: 1, onLog: log, }); @@ -38,19 +42,32 @@ describe("usage checkpoint stream", () => { }); it("does not promote partial usage to complete merely because the stream was flushed", async () => { - const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, parse); + const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, createParser()); await log("stdout", event); await log.flush(); expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: false }); }); - it("bounds replay work by stream size while retaining the exact final total", async () => { - const parser = vi.fn(parse); const saved = vi.fn(); + it("persists each changed total and the final receipt before publishing its chunk", async () => { + const order: string[] = []; + const saved = vi.fn(async (receipt: AdapterUsageCheckpoint) => { order.push(`saved:${receipt.costUsd}:${receipt.complete}`); }); + const log = createUsageCheckpointLog(async () => { order.push("log"); }, saved, createParser()); + await log("stdout", event + "\n"); + await log("stdout", event + "\n"); + await log("stdout", JSON.stringify({ type: "result", usage: { costUsd: 0 } }) + "\n"); + expect(order).toEqual(["saved:0.25:false", "log", "saved:0.5:false", "log", "saved:0.5:true", "log"]); + }); + + it("saves every new receipt before logging with linear parsing work", async () => { + const parser = vi.fn(createParser()); const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, parser); - for (let i = 0; i < 1024; i++) await log("stdout", event + "\n"); + for (let i = 0; i < 1024; i++) { + await log("stdout", event + "\n"); + expect(saved.mock.calls.at(-1)![0].costUsd).toBe((i + 1) * 0.25); + } await log.flush({ complete: true }); expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 256, complete: true }); - expect(parser.mock.calls.length).toBeLessThanOrEqual(12); - const finalSize = parser.mock.calls.at(-1)![0].length; - expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThan(3 * finalSize); + expect(saved).toHaveBeenCalledTimes(1025); // all updates plus completion + expect(parser).toHaveBeenCalledTimes(1024); + expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThanOrEqual(1024 * (event.length + 1)); }); }); diff --git a/packages/adapter-utils/src/usage-checkpoint.ts b/packages/adapter-utils/src/usage-checkpoint.ts index f7ebe701203..1e0c44bd86e 100644 --- a/packages/adapter-utils/src/usage-checkpoint.ts +++ b/packages/adapter-utils/src/usage-checkpoint.ts @@ -24,17 +24,18 @@ function project(value: unknown, depth = 0, field = ""): unknown { : project(entry, depth + 1, key)])); } -/** A fresh instance belongs to one CLI attempt. Call flush() after its process +/** The consumer incrementally parses only new records (never the full history). + * A fresh instance belongs to one CLI attempt. Call flush() after its process * exits, before retrying or final result handling. Checkpoint failures are * retained outside runChildProcess's best-effort onLog error handler. */ export function createUsageCheckpointLog( onLog: (stream: "stdout" | "stderr", chunk: string) => Promise, onUsage: ((receipt: AdapterUsageCheckpoint) => Promise) | undefined, - parse: (stdout: string) => AdapterUsageCheckpoint | null, + consume: (records: string) => AdapterUsageCheckpoint | null, ) { const attemptId = randomUUID(); let remainder = "", accounting = "", previous = ""; - let nextReplaySize = 0; + let snapshot: AdapterUsageCheckpoint | null = null; let failure: unknown; let failed = false; function retain(line: string) { @@ -43,14 +44,15 @@ export function createUsageCheckpointLog( const compact = JSON.stringify(project(raw)); if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"|"model"|"modelID"/.test(compact)) return; accounting += compact + "\n"; - if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB"); + if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint chunk exceeds 8 MiB"); } - async function publish(final = false, complete = false) { - if (!onUsage || !accounting || (!final && accounting.length < nextReplaySize)) return; - const parsed = parse(accounting); - // Geometric replay bounds total parsing work by stream size. The first - // observation is saved immediately; flush always saves the final snapshot. - nextReplaySize = accounting.length * 2; + async function publish(complete = false) { + if (!onUsage) return; + if (accounting) { + snapshot = consume(accounting); + accounting = ""; + } + const parsed = snapshot; if (!parsed) return; if (!parsed.complete && parsed.costUsd == null && parsed.costUsdExact == null && !Object.values(parsed.usage ?? {}).some(value => typeof value === "number" && value > 0)) return; @@ -79,7 +81,7 @@ export function createUsageCheckpointLog( if (failed) throw failure; if (!onUsage) return; if (remainder) { retain(remainder); remainder = ""; } - await publish(true, options.complete ?? false); + await publish(options.complete ?? false); }, }); } diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 2f407d05364..fedcafe5132 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -65,7 +65,7 @@ import { import { claudeModelUsageTotals, claudeModelReceipts, - parseClaudeStreamJson, + parseClaudeStreamJson, createClaudeStreamParser, describeClaudeFailure, detectClaudeLoginRequired, extractClaudeRetryNotBefore, @@ -964,8 +964,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseClaudeStreamJson(stdout); + const parsed = consumeAccounting(stdout); return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd, provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic", billingType, model: Object.keys(parseObject(parsed.resultJson?.modelUsage)).length > 1 ? "mixed" : parsed.model || model, @@ -989,7 +990,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise | null = null; const assistantTexts: string[] = []; const messageUsage = new Map(); let anonymousMessage = 0; + const observedTotals = { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = rawLine.trim(); - if (!line) continue; - const event = parseJson(line); - if (!event) continue; + return (stdout: string) => { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; + const event = parseJson(line); + if (!event) continue; - const type = asString(event.type, ""); - if (type === "system" && asString(event.subtype, "") === "init") { - sessionId = asString(event.session_id, sessionId ?? "") || sessionId; - model = asString(event.model, model); - continue; - } - - if (type === "assistant") { - sessionId = asString(event.session_id, sessionId ?? "") || sessionId; - const message = parseObject(event.message); - const observed = parseObject(message.usage); - if (Object.keys(observed).length > 0) { - const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`; - const previous = messageUsage.get(key); - messageUsage.set(key, { - inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)), - cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)), - outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)), - }); + const type = asString(event.type, ""); + if (type === "system" && asString(event.subtype, "") === "init") { + sessionId = asString(event.session_id, sessionId ?? "") || sessionId; + model = asString(event.model, model); + continue; } - const content = Array.isArray(message.content) ? message.content : []; - for (const entry of content) { - if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue; - const block = entry as Record; - if (asString(block.type, "") === "text") { - const text = asString(block.text, ""); - if (text) assistantTexts.push(text); + + if (type === "assistant") { + sessionId = asString(event.session_id, sessionId ?? "") || sessionId; + const message = parseObject(event.message); + const observed = parseObject(message.usage); + if (Object.keys(observed).length > 0) { + const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`; + const previous = messageUsage.get(key); + const next = { + inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)), + cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)), + outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)), + }; + observedTotals.inputTokens += next.inputTokens - (previous?.inputTokens ?? 0); + observedTotals.outputTokens += next.outputTokens - (previous?.outputTokens ?? 0); + observedTotals.cachedInputTokens += next.cachedInputTokens - (previous?.cachedInputTokens ?? 0); + messageUsage.set(key, next); } + const content = Array.isArray(message.content) ? message.content : []; + for (const entry of content) { + if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue; + const block = entry as Record; + if (asString(block.type, "") === "text") { + const text = asString(block.text, ""); + if (text) assistantTexts.push(text); + } + } + continue; + } + + if (type === "result") { + finalResult = event; + sessionId = asString(event.session_id, sessionId ?? "") || sessionId; } - continue; } - if (type === "result") { - finalResult = event; - sessionId = asString(event.session_id, sessionId ?? "") || sessionId; + if (!finalResult) { + return { + sessionId, + model, + costUsd: null as number | null, + usage: messageUsage.size ? { ...observedTotals } : null, + usageBasis: "per_run" as const, + summary: assistantTexts.join("\n\n").trim(), + resultJson: null as Record | null, + }; } - } - if (!finalResult) { + const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage); + const usageObj = parseObject(finalResult.usage); + const usage: UsageSummary = modelUsageTotals ?? { + inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0), + cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0), + outputTokens: asNumber(usageObj.output_tokens, 0), + }; + const costRaw = finalResult.total_cost_usd; + const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null; + const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim(); + return { sessionId, model, - costUsd: null as number | null, - usage: messageUsage.size ? [...messageUsage.values()].reduce((total, part) => ({ - inputTokens: total.inputTokens + part.inputTokens, outputTokens: total.outputTokens + part.outputTokens, - cachedInputTokens: (total.cachedInputTokens ?? 0) + (part.cachedInputTokens ?? 0), - }), { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }) : null, + costUsd, + usage, + // modelUsage covers exactly this CLI invocation, so mark it per-run to + // keep the server from applying its session-cumulative delta heuristic. usageBasis: "per_run" as const, - summary: assistantTexts.join("\n\n").trim(), - resultJson: null as Record | null, + summary, + resultJson: finalResult, }; - } - - const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage); - const usageObj = parseObject(finalResult.usage); - const usage: UsageSummary = modelUsageTotals ?? { - inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0), - cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0), - outputTokens: asNumber(usageObj.output_tokens, 0), - }; - const costRaw = finalResult.total_cost_usd; - const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null; - const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim(); - - return { - sessionId, - model, - costUsd, - usage, - // modelUsage covers exactly this CLI invocation, so mark it per-run to - // keep the server from applying its session-cumulative delta heuristic. - usageBasis: "per_run" as const, - summary, - resultJson: finalResult, }; } diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index 2069e55f957..630e82fb7da 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -62,7 +62,7 @@ import { type LocalProcessSandboxOptions, } from "@paperclipai/adapter-utils/local-process-sandbox"; import { - parseCodexJsonl, + parseCodexJsonl, createCodexJsonlParser, classifyCodexAuthRefreshFailure, extractCodexRetryNotBefore, isCodexHarnessCrash, @@ -1319,8 +1319,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseCodexJsonl(stdout); + const parsed = consumeAccounting(stdout); return { usage: parsed.usage, usageBasis: "per_run", provider: "openai", biller: resolveCodexBiller(effectiveEnv, billingType), billingType, model, costUsd: null, complete: parsed.sawProtocolTerminalEvent }; }); try { @@ -1347,7 +1348,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const type = asString(event.type, ""); - if (type) sawProtocolEvent = true; - if (type === "error" || type === "turn.completed" || type === "turn.failed") { - sawProtocolTerminalEvent = true; - } - if (type === "thread.started") { - sessionId = asString(event.thread_id, sessionId ?? "") || sessionId; - continue; - } + const type = asString(event.type, ""); + if (type) sawProtocolEvent = true; + if (type === "error" || type === "turn.completed" || type === "turn.failed") { + sawProtocolTerminalEvent = true; + } + if (type === "thread.started") { + sessionId = asString(event.thread_id, sessionId ?? "") || sessionId; + continue; + } - if (type === "error") { - const msg = asString(event.message, "").trim(); - if (msg) errorMessage = msg; - continue; - } + if (type === "error") { + const msg = asString(event.message, "").trim(); + if (msg) errorMessage = msg; + continue; + } - if (type === "item.completed") { - const item = parseObject(event.item); - if (asString(item.type, "") === "agent_message") { - const text = asString(item.text, ""); - if (text) finalMessage = text; + if (type === "item.completed") { + const item = parseObject(event.item); + if (asString(item.type, "") === "agent_message") { + const text = asString(item.text, ""); + if (text) finalMessage = text; + } + continue; + } + + if (type === "turn.completed") { + const usageObj = parseObject(event.usage); + usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens); + usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens); + usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens); + continue; + } + + if (type === "turn.failed") { + const err = parseObject(event.error); + const msg = asString(err.message, "").trim(); + if (msg) errorMessage = msg; } - continue; } - if (type === "turn.completed") { - const usageObj = parseObject(event.usage); - usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens); - usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens); - usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens); - continue; - } - - if (type === "turn.failed") { - const err = parseObject(event.error); - const msg = asString(err.message, "").trim(); - if (msg) errorMessage = msg; - } - } - - return { - sessionId, - summary: finalMessage?.trim() ?? "", - // Codex includes cache hits in input_tokens; Paperclip stores them separately. - usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) }, - usageBasis: "per_run" as const, - errorMessage, - sawProtocolEvent, - sawProtocolTerminalEvent, + return { + sessionId, + summary: finalMessage?.trim() ?? "", + // Codex includes cache hits in input_tokens; Paperclip stores them separately. + usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) }, + usageBasis: "per_run" as const, + errorMessage, + sawProtocolEvent, + sawProtocolTerminalEvent, + }; }; } diff --git a/packages/adapters/cursor-local/src/server/execute.ts b/packages/adapters/cursor-local/src/server/execute.ts index 2c0b49dbd8c..2c1e6698c6d 100644 --- a/packages/adapters/cursor-local/src/server/execute.ts +++ b/packages/adapters/cursor-local/src/server/execute.ts @@ -54,7 +54,7 @@ import { joinPromptSections, } from "@paperclipai/adapter-utils/server-utils"; import { DEFAULT_CURSOR_LOCAL_MODEL, SANDBOX_INSTALL_COMMAND } from "../index.js"; -import { firstCursorDiagnosticLine, parseCursorJsonl, isCursorUnknownSessionError } from "./parse.js"; +import { firstCursorDiagnosticLine, parseCursorJsonl, createCursorJsonlParser, isCursorUnknownSessionError } from "./parse.js"; import { prepareCursorSandboxCommand } from "./remote-command.js"; import { normalizeCursorStreamLine } from "../shared/stream.js"; import { hasCursorTrustBypassArg } from "../shared/trust.js"; @@ -627,8 +627,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseCursorJsonl(stdout); + const parsed = consumeAccounting(stdout); const provider = resolveProviderFromModel(model); return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: parsed.sawResult }; }); @@ -676,7 +677,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise): string | null { } export function parseCursorJsonl(stdout: string) { + return createCursorJsonlParser()(stdout); +} + +/** Consume complete JSONL records once, retaining protocol accounting state. */ +export function createCursorJsonlParser() { let sessionId: string | null = null; let sawResult = false; const messages: string[] = []; @@ -80,94 +85,96 @@ export function parseCursorJsonl(stdout: string) { outputTokens: 0, }; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = normalizeCursorStreamLine(rawLine).line; - if (!line) continue; + return (stdout: string) => { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = normalizeCursorStreamLine(rawLine).line; + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const foundSession = readSessionId(event); - if (foundSession) sessionId = foundSession; + const foundSession = readSessionId(event); + if (foundSession) sessionId = foundSession; - const type = asString(event.type, "").trim(); + const type = asString(event.type, "").trim(); - if (type === "assistant") { - messages.push(...collectAssistantText(event.message)); - continue; - } - - if (type === "result") { - sawResult = true; - const usageObj = parseObject(event.usage); - usage.inputTokens += asNumber( - usageObj.input_tokens, - asNumber(usageObj.inputTokens, 0), - ); - usage.cachedInputTokens += asNumber( - usageObj.cached_input_tokens, - asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)), - ); - usage.outputTokens += asNumber( - usageObj.output_tokens, - asNumber(usageObj.outputTokens, 0), - ); - addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost); - - const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error"; - const resultText = asString(event.result, "").trim(); - if (resultText && messages.length === 0) { - messages.push(resultText); + if (type === "assistant") { + messages.push(...collectAssistantText(event.message)); + continue; } - if (isError) { - const resultError = asErrorText(event.error ?? event.message ?? event.result).trim(); - if (resultError) errorMessage = resultError; + + if (type === "result") { + sawResult = true; + const usageObj = parseObject(event.usage); + usage.inputTokens += asNumber( + usageObj.input_tokens, + asNumber(usageObj.inputTokens, 0), + ); + usage.cachedInputTokens += asNumber( + usageObj.cached_input_tokens, + asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)), + ); + usage.outputTokens += asNumber( + usageObj.output_tokens, + asNumber(usageObj.outputTokens, 0), + ); + addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost); + + const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error"; + const resultText = asString(event.result, "").trim(); + if (resultText && messages.length === 0) { + messages.push(resultText); + } + if (isError) { + const resultError = asErrorText(event.error ?? event.message ?? event.result).trim(); + if (resultError) errorMessage = resultError; + } + continue; } - continue; - } - if (type === "error") { - const message = asErrorText(event.message ?? event.error ?? event.detail).trim(); - if (message) errorMessage = message; - continue; - } - - if (type === "system") { - const subtype = asString(event.subtype, "").trim().toLowerCase(); - if (subtype === "error") { + if (type === "error") { const message = asErrorText(event.message ?? event.error ?? event.detail).trim(); if (message) errorMessage = message; + continue; + } + + if (type === "system") { + const subtype = asString(event.subtype, "").trim().toLowerCase(); + if (subtype === "error") { + const message = asErrorText(event.message ?? event.error ?? event.detail).trim(); + if (message) errorMessage = message; + } + continue; + } + + // Compatibility with older stream-json shapes. + if (type === "text") { + const part = parseObject(event.part); + const text = asString(part.text, "").trim(); + if (text) messages.push(text); + continue; + } + + if (type === "step_finish") { + const part = parseObject(event.part); + const tokens = parseObject(part.tokens); + const cache = parseObject(tokens.cache); + usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); + usage.cachedInputTokens += asNumber(cache.read, 0); + usage.outputTokens += asNumber(tokens.output, 0); + addCost(part.cost); + continue; } - continue; } - // Compatibility with older stream-json shapes. - if (type === "text") { - const part = parseObject(event.part); - const text = asString(part.text, "").trim(); - if (text) messages.push(text); - continue; - } - - if (type === "step_finish") { - const part = parseObject(event.part); - const tokens = parseObject(part.tokens); - const cache = parseObject(tokens.cache); - usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); - usage.cachedInputTokens += asNumber(cache.read, 0); - usage.outputTokens += asNumber(tokens.output, 0); - addCost(part.cost); - continue; - } - } - - return { - sawResult, - sessionId, - summary: messages.join("\n\n").trim(), - usage, - costUsd: reportedCost && !missingCost ? totalCostUsd : null, - errorMessage, + return { + sawResult, + sessionId, + summary: messages.join("\n\n").trim(), + usage: { ...usage }, + costUsd: reportedCost && !missingCost ? totalCostUsd : null, + errorMessage, + }; }; } diff --git a/packages/adapters/gemini-local/src/server/execute.ts b/packages/adapters/gemini-local/src/server/execute.ts index 500d395a819..3e4b19b9ac3 100644 --- a/packages/adapters/gemini-local/src/server/execute.ts +++ b/packages/adapters/gemini-local/src/server/execute.ts @@ -63,7 +63,7 @@ import { isGeminiTransientNetworkError, isGeminiTurnLimitResult, isGeminiSessionUnrecoverableError, - parseGeminiJsonl, + parseGeminiJsonl, createGeminiJsonlParser, } from "./parse.js"; import { firstNonEmptyLine } from "./utils.js"; import { @@ -643,8 +643,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseGeminiJsonl(stdout); + const parsed = consumeAccounting(stdout); return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null }; }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { @@ -659,7 +660,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const foundSessionId = readSessionId(event); - if (foundSessionId) sessionId = foundSessionId; + const foundSessionId = readSessionId(event); + if (foundSessionId) sessionId = foundSessionId; - const type = asString(event.type, "").trim(); + const type = asString(event.type, "").trim(); - if (type === "assistant") { - messages.push(...collectMessageText(event.message)); - const messageObj = parseObject(event.message); - const content = Array.isArray(messageObj.content) ? messageObj.content : []; - for (const partRaw of content) { - const part = parseObject(partRaw); - if (asString(part.type, "").trim() === "question") { - question = { - prompt: asString(part.prompt, "").trim(), - choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => { - const choice = parseObject(choiceRaw); - return { - key: asString(choice.key, "").trim(), - label: asString(choice.label, "").trim(), - description: asString(choice.description, "").trim() || undefined, - }; - }), - }; - break; // only one question per message + if (type === "assistant") { + messages.push(...collectMessageText(event.message)); + const messageObj = parseObject(event.message); + const content = Array.isArray(messageObj.content) ? messageObj.content : []; + for (const partRaw of content) { + const part = parseObject(partRaw); + if (asString(part.type, "").trim() === "question") { + question = { + prompt: asString(part.prompt, "").trim(), + choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => { + const choice = parseObject(choiceRaw); + return { + key: asString(choice.key, "").trim(), + label: asString(choice.label, "").trim(), + description: asString(choice.description, "").trim() || undefined, + }; + }), + }; + break; // only one question per message + } } + continue; } - continue; - } - // Gemini CLI v0.38+ stream-json schema emits assistant turns as: - // {"type":"message","role":"assistant","content":"...","delta":true} - // These are discrete final messages (one per assistant turn), not - // cumulative streaming tokens, so collecting all of them produces the - // expected concatenated turn-by-turn summary rather than duplicated text. - if (type === "message") { - const role = asString(event.role, "").trim().toLowerCase(); - if (role === "assistant") { - messages.push(...collectMessageText(event.content)); + // Gemini CLI v0.38+ stream-json schema emits assistant turns as: + // {"type":"message","role":"assistant","content":"...","delta":true} + // These are discrete final messages (one per assistant turn), not + // cumulative streaming tokens, so collecting all of them produces the + // expected concatenated turn-by-turn summary rather than duplicated text. + if (type === "message") { + const role = asString(event.role, "").trim().toLowerCase(); + if (role === "assistant") { + messages.push(...collectMessageText(event.content)); + } + continue; } - continue; - } - if (type === "result") { - resultEvent = event; - accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats); - const resultText = - asString(event.result, "").trim() || - asString(event.text, "").trim() || - asString(event.response, "").trim(); - if (resultText && messages.length === 0) messages.push(resultText); - const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); - if (typeof reportedCost === "number") costUsd = reportedCost; - const status = asString(event.status, "").toLowerCase(); - const isError = - event.is_error === true || - asString(event.subtype, "").toLowerCase() === "error" || - status === "error" || - status === "failed"; - if (isError) { - const text = asErrorText(event.error ?? event.message ?? event.result).trim(); - if (text) errorMessage = text; + if (type === "result") { + resultEvent = event; + accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats); + const resultText = + asString(event.result, "").trim() || + asString(event.text, "").trim() || + asString(event.response, "").trim(); + if (resultText && messages.length === 0) messages.push(resultText); + const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); + if (typeof reportedCost === "number") costUsd = reportedCost; + const status = asString(event.status, "").toLowerCase(); + const isError = + event.is_error === true || + asString(event.subtype, "").toLowerCase() === "error" || + status === "error" || + status === "failed"; + if (isError) { + const text = asErrorText(event.error ?? event.message ?? event.result).trim(); + if (text) errorMessage = text; + } + continue; } - continue; - } - if (type === "error") { - const text = asErrorText(event.error ?? event.message ?? event.detail).trim(); - if (text) errorMessage = text; - continue; - } - - if (type === "system") { - const subtype = asString(event.subtype, "").trim().toLowerCase(); - if (subtype === "error") { + if (type === "error") { const text = asErrorText(event.error ?? event.message ?? event.detail).trim(); if (text) errorMessage = text; + continue; + } + + if (type === "system") { + const subtype = asString(event.subtype, "").trim().toLowerCase(); + if (subtype === "error") { + const text = asErrorText(event.error ?? event.message ?? event.detail).trim(); + if (text) errorMessage = text; + } + continue; + } + + if (type === "text") { + const part = parseObject(event.part); + const text = asString(part.text, "").trim(); + if (text) messages.push(text); + continue; + } + + if (type === "step_finish" || event.usage || event.usageMetadata) { + accumulateUsage(usage, event.usage ?? event.usageMetadata); + const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); + if (typeof reportedCost === "number") costUsd = reportedCost; + continue; } - continue; } - if (type === "text") { - const part = parseObject(event.part); - const text = asString(part.text, "").trim(); - if (text) messages.push(text); - continue; - } - - if (type === "step_finish" || event.usage || event.usageMetadata) { - accumulateUsage(usage, event.usage ?? event.usageMetadata); - const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); - if (typeof reportedCost === "number") costUsd = reportedCost; - continue; - } - } - - return { - sessionId, - summary: messages.join("\n\n").trim(), - usage, - costUsd, - errorMessage, - resultEvent, - question, + return { + sessionId, + summary: messages.join("\n\n").trim(), + usage: { ...usage }, + costUsd, + errorMessage, + resultEvent, + question, + }; }; } diff --git a/packages/adapters/opencode-local/src/server/execute.ts b/packages/adapters/opencode-local/src/server/execute.ts index 6576bafe37a..72dd350f6b4 100644 --- a/packages/adapters/opencode-local/src/server/execute.ts +++ b/packages/adapters/opencode-local/src/server/execute.ts @@ -53,7 +53,7 @@ import { readPaperclipIssueWorkModeFromContext, resolveLegacyPaperclipDesiredSkillNames, } from "@paperclipai/adapter-utils/server-utils"; -import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl } from "./parse.js"; +import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl, createOpenCodeJsonlParser } from "./parse.js"; import { ensureOpenCodeModelConfiguredAndAvailable, isTruthyEnvFlag, @@ -646,8 +646,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseOpenCodeJsonl(stdout); + const parsed = consumeAccounting(stdout); const provider = parseModelProvider(model || null); return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false }; }); @@ -664,7 +665,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const currentSessionId = asString(event.sessionID, "").trim(); - if (currentSessionId) sessionId = currentSessionId; + const currentSessionId = asString(event.sessionID, "").trim(); + if (currentSessionId) sessionId = currentSessionId; - const type = asString(event.type, ""); + const type = asString(event.type, ""); - if (type === "text") { - const part = parseObject(event.part); - const text = asString(part.text, "").trim(); - if (text) messages.push(text); - continue; - } + if (type === "text") { + const part = parseObject(event.part); + const text = asString(part.text, "").trim(); + if (text) messages.push(text); + continue; + } - if (type === "step_finish") { - const part = parseObject(event.part); - const tokens = parseObject(part.tokens); - const cache = parseObject(tokens.cache); - usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); - usage.cachedInputTokens += asNumber(cache.read, 0); - usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0); - if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost; - else missingCost = true; - continue; - } + if (type === "step_finish") { + const part = parseObject(event.part); + const tokens = parseObject(part.tokens); + const cache = parseObject(tokens.cache); + usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); + usage.cachedInputTokens += asNumber(cache.read, 0); + usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0); + if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost; + else missingCost = true; + continue; + } - if (type === "tool_use") { - const part = parseObject(event.part); - const state = parseObject(part.state); - if (asString(state.status, "") === "error") { - const text = asString(state.error, "").trim(); - if (text) toolErrors.push(text); + if (type === "tool_use") { + const part = parseObject(event.part); + const state = parseObject(part.state); + if (asString(state.status, "") === "error") { + const text = asString(state.error, "").trim(); + if (text) toolErrors.push(text); + } + continue; + } + + if (type === "error") { + const text = errorText(event.error ?? event.message).trim(); + if (text) errors.push(text); + continue; } - continue; } - if (type === "error") { - const text = errorText(event.error ?? event.message).trim(); - if (text) errors.push(text); - continue; - } - } - - return { - sessionId, - summary: messages.join("\n\n").trim(), - usage, - costUsd: missingCost ? null : costUsd, - errorMessage: errors.length > 0 ? errors.join("\n") : null, - toolErrors, + return { + sessionId, + summary: messages.join("\n\n").trim(), + usage: { ...usage }, + costUsd: missingCost ? null : costUsd, + errorMessage: errors.length > 0 ? errors.join("\n") : null, + toolErrors, + }; }; } diff --git a/packages/adapters/pi-local/src/server/execute.ts b/packages/adapters/pi-local/src/server/execute.ts index 3a07b333e4a..0274d2810e5 100644 --- a/packages/adapters/pi-local/src/server/execute.ts +++ b/packages/adapters/pi-local/src/server/execute.ts @@ -55,7 +55,7 @@ import { runChildProcess, } from "@paperclipai/adapter-utils/server-utils"; import { shellQuote } from "@paperclipai/adapter-utils/ssh"; -import { isPiUnknownSessionError, parsePiJsonl } from "./parse.js"; +import { isPiUnknownSessionError, parsePiJsonl, createPiJsonlParser } from "./parse.js"; import { ensurePiModelConfiguredAndAvailable } from "./models.js"; import { preparePiRuntimeConfig } from "./runtime-config.js"; import { SANDBOX_INSTALL_COMMAND } from "../index.js"; @@ -737,8 +737,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parsePiJsonl(stdout); + const parsed = consumeAccounting(stdout); return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: parsed.sawAgentEnd }; }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { @@ -753,7 +754,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const eventType = asString(event.type, ""); + const eventType = asString(event.type, ""); - // Pi can exit successfully after a provider failure. The terminal assistant - // message carries that failure in both message_end and turn_end envelopes. - const terminalMessages = eventType === "agent_end" - ? (Array.isArray(event.messages) ? event.messages : []) - : eventType === "message_end" || eventType === "turn_end" - ? [event.message] - : []; - for (const rawMessage of terminalMessages) { - const message = asRecord(rawMessage); - if (message?.role !== "assistant" || message.stopReason !== "error") continue; - const error = asString(message.errorMessage, "").trim() || "Pi provider request failed."; - if (!result.errors.includes(error)) result.errors.push(error); - } - - // RPC protocol messages - skip these (internal implementation detail) - if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") { - continue; - } - - // Agent lifecycle - if (eventType === "agent_start") { - continue; - } - - if (eventType === "agent_end") { - result.sawAgentEnd = true; - const messages = event.messages as Array> | undefined; - if (messages && messages.length > 0) { - const lastMessage = messages[messages.length - 1]; - if (lastMessage?.role === "assistant") { - const content = lastMessage.content as string | Array<{ type: string; text?: string }>; - result.finalMessage = extractTextContent(content); - } + // Pi can exit successfully after a provider failure. The terminal assistant + // message carries that failure in both message_end and turn_end envelopes. + const terminalMessages = eventType === "agent_end" + ? (Array.isArray(event.messages) ? event.messages : []) + : eventType === "message_end" || eventType === "turn_end" + ? [event.message] + : []; + for (const rawMessage of terminalMessages) { + const message = asRecord(rawMessage); + if (message?.role !== "assistant" || message.stopReason !== "error") continue; + const error = asString(message.errorMessage, "").trim() || "Pi provider request failed."; + if (!result.errors.includes(error)) result.errors.push(error); } - continue; - } - if (eventType === "auto_retry_end") { - const succeeded = event.success === true; - if (!succeeded) { - const finalError = asString(event.finalError, "").trim(); - result.errors.push(finalError || "Pi exhausted automatic retries without producing a response."); + // RPC protocol messages - skip these (internal implementation detail) + if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") { + continue; } - continue; - } - // Turn lifecycle - if (eventType === "turn_start") { - continue; - } - - if (eventType === "turn_end") { - const message = asRecord(event.message); - if (message) { - const content = message.content as string | Array<{ type: string; text?: string }>; - const text = extractTextContent(content); - if (text) { - result.finalMessage = text; - result.messages.push(text); - } - - // Extract usage and cost from assistant message - const usage = asRecord(message.usage); - if (usage) { - result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0); - result.usage.outputTokens += asNumber(usage.output, 0); - result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0); - - // Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.) - const cost = asRecord(usage.cost); - addCost(cost?.total); - } + // Agent lifecycle + if (eventType === "agent_start") { + continue; } - - // Tool results are in toolResults array - const toolResults = event.toolResults as Array> | undefined; - if (toolResults) { - for (const tr of toolResults) { - const toolCallId = asString(tr.toolCallId, ""); - const content = tr.content; - const isError = tr.isError === true; - - // Find matching tool call by toolCallId - const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); - if (existingCall) { - existingCall.result = typeof content === "string" ? content : JSON.stringify(content); - existingCall.isError = isError; + + if (eventType === "agent_end") { + result.sawAgentEnd = true; + const messages = event.messages as Array> | undefined; + if (messages && messages.length > 0) { + const lastMessage = messages[messages.length - 1]; + if (lastMessage?.role === "assistant") { + const content = lastMessage.content as string | Array<{ type: string; text?: string }>; + result.finalMessage = extractTextContent(content); } } + continue; } - continue; - } - // Message updates (streaming) - if (eventType === "message_update") { - const assistantEvent = asRecord(event.assistantMessageEvent); - if (assistantEvent) { - const msgType = asString(assistantEvent.type, ""); - if (msgType === "text_delta") { - const delta = asString(assistantEvent.delta, ""); - if (delta) { - // Append to last message or create new - if (result.messages.length === 0) { - result.messages.push(delta); - } else { - result.messages[result.messages.length - 1] += delta; + if (eventType === "auto_retry_end") { + const succeeded = event.success === true; + if (!succeeded) { + const finalError = asString(event.finalError, "").trim(); + result.errors.push(finalError || "Pi exhausted automatic retries without producing a response."); + } + continue; + } + + // Turn lifecycle + if (eventType === "turn_start") { + continue; + } + + if (eventType === "turn_end") { + const message = asRecord(event.message); + if (message) { + const content = message.content as string | Array<{ type: string; text?: string }>; + const text = extractTextContent(content); + if (text) { + result.finalMessage = text; + result.messages.push(text); + } + + // Extract usage and cost from assistant message + const usage = asRecord(message.usage); + if (usage) { + result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0); + result.usage.outputTokens += asNumber(usage.output, 0); + result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0); + + // Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.) + const cost = asRecord(usage.cost); + addCost(cost?.total); + } + } + + // Tool results are in toolResults array + const toolResults = event.toolResults as Array> | undefined; + if (toolResults) { + for (const tr of toolResults) { + const toolCallId = asString(tr.toolCallId, ""); + const content = tr.content; + const isError = tr.isError === true; + + // Find matching tool call by toolCallId + const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); + if (existingCall) { + existingCall.result = typeof content === "string" ? content : JSON.stringify(content); + existingCall.isError = isError; } } } + continue; } - continue; - } - if (eventType === "error") { - const message = asString(event.message, "").trim(); - if (message) { - result.errors.push(message); + // Message updates (streaming) + if (eventType === "message_update") { + const assistantEvent = asRecord(event.assistantMessageEvent); + if (assistantEvent) { + const msgType = asString(assistantEvent.type, ""); + if (msgType === "text_delta") { + const delta = asString(assistantEvent.delta, ""); + if (delta) { + // Append to last message or create new + if (result.messages.length === 0) { + result.messages.push(delta); + } else { + result.messages[result.messages.length - 1] += delta; + } + } + } + } + continue; } - continue; - } - // Tool execution - if (eventType === "tool_execution_start") { - const toolCallId = asString(event.toolCallId, ""); - const toolName = asString(event.toolName, ""); - const args = event.args; - currentToolCall = { toolCallId, toolName, args }; - result.toolCalls.push({ - toolCallId, - toolName, - args, - result: null, - isError: false, - }); - continue; - } - - if (eventType === "tool_execution_end") { - const toolCallId = asString(event.toolCallId, ""); - const toolName = asString(event.toolName, ""); - const toolResult = event.result; - const isError = event.isError === true; - - // Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool) - const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); - if (existingCall) { - existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult); - existingCall.isError = isError; + if (eventType === "error") { + const message = asString(event.message, "").trim(); + if (message) { + result.errors.push(message); + } + continue; } - currentToolCall = null; - continue; - } - // Usage tracking if available in the event (fallback for standalone usage events) - if (eventType === "usage" || event.usage) { - const usage = asRecord(event.usage); - if (usage) { - // Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens) - result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0); - result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0); - result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0); - - // Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format) - const cost = asRecord(usage.cost); - addCost(cost?.total ?? usage.costUsd); + // Tool execution + if (eventType === "tool_execution_start") { + const toolCallId = asString(event.toolCallId, ""); + const toolName = asString(event.toolName, ""); + const args = event.args; + currentToolCall = { toolCallId, toolName, args }; + result.toolCalls.push({ + toolCallId, + toolName, + args, + result: null, + isError: false, + }); + continue; + } + + if (eventType === "tool_execution_end") { + const toolCallId = asString(event.toolCallId, ""); + const toolName = asString(event.toolName, ""); + const toolResult = event.result; + const isError = event.isError === true; + + // Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool) + const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); + if (existingCall) { + existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult); + existingCall.isError = isError; + } + currentToolCall = null; + continue; + } + + // Usage tracking if available in the event (fallback for standalone usage events) + if (eventType === "usage" || event.usage) { + const usage = asRecord(event.usage); + if (usage) { + // Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens) + result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0); + result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0); + result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0); + + // Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format) + const cost = asRecord(usage.cost); + addCost(cost?.total ?? usage.costUsd); + } } } - } - if (missingCost) result.usage.costUsd = null; - return result; + if (missingCost) result.usage.costUsd = null; + return { ...result, usage: { ...result.usage }, messages: [...result.messages], errors: [...result.errors], toolCalls: [...result.toolCalls] }; + }; } export function isPiUnknownSessionError(stdout: string, stderr: string): boolean { diff --git a/server/src/__tests__/adapter-cost-failure-paths.test.ts b/server/src/__tests__/adapter-cost-failure-paths.test.ts index 0ad2c8d0038..141dd5f16b7 100644 --- a/server/src/__tests__/adapter-cost-failure-paths.test.ts +++ b/server/src/__tests__/adapter-cost-failure-paths.test.ts @@ -50,8 +50,18 @@ describe("CLI adapter accounting on timeout", () => { context: {}, onLog: async () => {}, onUsage, }; const execute = await fixture.execute(); - await execute(context); - expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete: true }); + const initial = await execute(context); + const complete = !["claude", "pi"].includes(fixture.name); + expect(initial.usageComplete).toBe(complete); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete }); + if (["claude", "pi", "opencode"].includes(fixture.name)) { + processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => { + await options.onLog("stdout", stdout).catch(() => {}); + return { exitCode: 1, signal: null, timedOut: false, stdout, stderr: "process failed", pid: 123, startedAt: new Date().toISOString() }; + }); + expect((await execute(context)).usageComplete).toBe(false); + expect(onUsage.mock.calls.at(-1)![0].complete).toBe(false); + } if (fixture.name === "claude") { const mixed = [ { type: "system", subtype: "init", model: "actual-model" }, @@ -97,3 +107,36 @@ describe("CLI adapter accounting on timeout", () => { expect(result.costUsd).toBe(fixture.price); }); }); + + +describe("incremental protocol accounting", () => { + it("keeps Claude message updates idempotent and preserves immutable snapshots", async () => { + const { createClaudeStreamParser } = await import("../../../packages/adapters/claude-local/src/server/parse.js"); + const consume = createClaudeStreamParser(); + consume(JSON.stringify({ type: "system", subtype: "init", model: "actual-model" })); + const message = (id: string, output: number) => JSON.stringify({ type: "assistant", message: { id, usage: { input_tokens: 2, output_tokens: output } } }); + const first = consume(message("one", 1)); + expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 }); + expect(consume(message("one", 3)).usage).toMatchObject({ inputTokens: 2, outputTokens: 3 }); + expect(consume(message("two", 2)).usage).toMatchObject({ inputTokens: 4, outputTokens: 5 }); + expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 }); + const final = consume(JSON.stringify({ type: "result", total_cost_usd: 1, usage: { input_tokens: 9, output_tokens: 8 } })); + expect(final).toMatchObject({ model: "actual-model", costUsd: 1, usage: { inputTokens: 9, outputTokens: 8 } }); + }); + it.each(cases.filter(item => !["claude", "kimi"].includes(item.name)))("updates $name totals once per newly received record", async fixture => { + const factories = { + codex: async () => (await import("../../../packages/adapters/codex-local/src/server/parse.js")).createCodexJsonlParser(), + cursor: async () => (await import("../../../packages/adapters/cursor-local/src/server/parse.js")).createCursorJsonlParser(), + gemini: async () => (await import("../../../packages/adapters/gemini-local/src/server/parse.js")).createGeminiJsonlParser(), + pi: async () => (await import("../../../packages/adapters/pi-local/src/server/parse.js")).createPiJsonlParser(), + opencode: async () => (await import("../../../packages/adapters/opencode-local/src/server/parse.js")).createOpenCodeJsonlParser(), + }; + const consume = await factories[fixture.name as keyof typeof factories](); + const line = JSON.stringify(fixture.event); + const first = consume(line); + const second = consume(line); + expect(first.usage).toMatchObject(fixture.tokens!); + const multiplier = fixture.name === "codex" ? 1 : 2; + expect(second.usage).toMatchObject(Object.fromEntries(Object.entries(fixture.tokens!).map(([key, count]) => [key, count * multiplier]))); + }); +}); From dc8d57e2a4543f28ef4412c9d42d8e116dbb98dc Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:47:50 -0700 Subject: [PATCH 9/9] test: verify durable checkpoints through the final flush contract Co-Authored-By: Paperclip --- .../cost-accounting-checkpoints.test.ts | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/cost-accounting-checkpoints.test.ts b/server/src/__tests__/cost-accounting-checkpoints.test.ts index dccf8fd4b1c..01eb99e441f 100644 --- a/server/src/__tests__/cost-accounting-checkpoints.test.ts +++ b/server/src/__tests__/cost-accounting-checkpoints.test.ts @@ -42,14 +42,20 @@ describe("durable accounting stream projection", () => { }); it("propagates receipt failures and leaves ordinary logs and stderr alone when disabled", async () => { const onLog = vi.fn(async (..._args: any[]) => {}); - expect(createUsageCheckpointLog(onLog, undefined, () => null)).toBe(onLog); + const disabled = createUsageCheckpointLog(onLog, undefined, () => null); + await disabled("stdout", "ordinary output"); await disabled.flush(); + expect(onLog).toHaveBeenCalledWith("stdout", "ordinary output"); + onLog.mockClear(); const onUsage = vi.fn(async () => { throw new Error("spool full"); }); - const parse = vi.fn(() => ({ complete: false })); + const parse = vi.fn(() => ({ usage: { inputTokens: 1, outputTokens: 0 }, complete: false })); const log = createUsageCheckpointLog(onLog, onUsage, parse); await log("stderr", '{"usage":1}\n'); await log("stdout", 'invalid json\n{"text":"plain"}\n'); expect(onUsage).not.toHaveBeenCalled(); onLog.mockClear(); - await expect(log("stdout", '{"usage":{"input_tokens":1}}\n')).rejects.toThrow("spool full"); - expect(onLog).not.toHaveBeenCalled(); - await expect(log("stdout", "x".repeat(8*1024*1024+1))).rejects.toThrow("exceeds"); + await log("stdout", '{"usage":{"input_tokens":1}}\n'); + expect(onLog).toHaveBeenCalled(); + await expect(log.flush()).rejects.toThrow("spool full"); + const oversized = createUsageCheckpointLog(onLog, onUsage, parse); + await oversized("stdout", "x".repeat(8*1024*1024+1)); + await expect(oversized.flush()).rejects.toThrow("exceeds"); }); });