Merge reviewed token aggregation and credential contention fixes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/COST-ACCOUNTING.md

@@ -42,7 +42,7 @@ Reservations are estimates, not provider-enforced spending caps. Delayed receipt
 
 Cost and finance endpoints default to the current UTC month through now. Explicit `from`/`to` bounds remain inclusive; `period=all` requests all time. Reversed or malformed bounds are rejected. Date-filterable CLI cost/finance reports expose `--from`, `--to`, and `--all-time`. Operational accounting, budget, live quota and rolling-window commands reject these flags because their endpoints do not filter by arbitrary dates. The board sends explicit ranges, and month/year-to-date starts use UTC. Rolling spend excludes future events. The dashboard compares spend with the monthly cap only for Month to Date; historical ranges show the monthly limit without a utilization percentage. Provider report imports default to the previous complete UTC day, with an exclusive end date.
 
-Cost summaries expose `eventCount`, `estimatedEventCount`, `unpricedEventCount`, `pendingRunCount`, and `pricingComplete`. Estimated charges count toward budgets, while the UI distinguishes estimates from provider-reported dollars. Agent totals and expanded model rows show **Estimated** when every charge in that group is estimated, or **Partially estimated** for a mix. Groups with no estimates have no estimate badge. These labels follow the selected date range and use ledger-event counts, not run counts; they do not infer pricing status from the provider or model name. `pricingComplete` means no missing prices or pending receipts; it does not mean invoice reconciliation has occurred. The Costs page warns when totals are incomplete. Provider cards include cached tokens and calculate subscription shares using each token once. Provider spend is compared with the real company budget, not a fabricated proportional allocation. Run displays prefer cache-adjusted prices and preserve explicit zeroes.
+Cost summaries expose `eventCount`, `estimatedEventCount`, `unpricedEventCount`, `pendingRunCount`, and `pricingComplete`. Estimated charges count toward budgets, while the UI distinguishes estimates from provider-reported dollars. Agent totals and expanded model rows show **Estimated** when every charge in that group is estimated, or **Partially estimated** for a mix. Groups with no estimates have no estimate badge. These labels follow the selected date range and use ledger-event counts, not run counts; they do not infer pricing status from the provider or model name. `pricingComplete` means no missing prices or pending receipts; it does not mean invoice reconciliation has occurred. The Costs page warns when totals are incomplete. Provider cards include cached tokens and calculate subscription shares using each token once. Reports normalize historical OpenAI events without a receipt fingerprint from inclusive input to exclusive input before grouping, so mixed historical/new totals count cache reads once across agents, models, providers, billers, projects, issues, and rolling windows. This read-time conversion preserves original ledger rows and charges; legacy Anthropic input already excludes cache reads. Provider spend is compared with the real company budget, not a fabricated proportional allocation. Run displays prefer cache-adjusted prices and preserve explicit zeroes.
 
 Gemini token normalization follows the [CLI stream statistics](https://github.com/google-gemini/gemini-cli/blob/main/packages/core/src/output/stream-json-formatter.ts) and [API usage metadata](https://ai.google.dev/api/generate-content#UsageMetadata): cached prompt tokens are separated from input, and thinking tokens contribute to output.
 

doc/connections/AI-CONNECTIONS.md

@@ -498,9 +498,10 @@ existing account name. Connecting installs access for that agent and resumes the
 work automatically. Explicit incompatible bindings and shared-account permission
 denials still fail; hiring never expands a restricted shared account's audience.
 
-Concurrent runs of one subscription do not wait for each other. No credential
-lease exists to hold them, so a fresh task execution cannot enter a contention
-wait. A run that already entered this wait keeps its scheduled retries. It does
+Concurrent runs of one subscription do not hold a credential lease. A fresh
+task can briefly wait when a credential rotation holds the company file lock or
+a grant/secret database row lock. Lock timeouts become `ai_connection_busy`,
+keeping the task on its automatic pre-provider scheduled retry path. It does
 not request new credentials, and it does not consume the provider-failure retry
 allowance. Each retry revalidates the account, and existing run-dispatch rules
 still suppress cancelled, reassigned, or otherwise ineligible work. An assignee

server/src/__tests__/agent-hire-ai-connections.test.ts

@@ -1,3 +1,5 @@
+import * as codexAdapter from "@paperclipai/adapter-codex-local/server";
+import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
 import { randomUUID } from "node:crypto";
 import { mkdtemp, rm } from "node:fs/promises";
 import os from "node:os";
@@ -323,6 +325,36 @@ describe("agent-created hires use managed AI connections", () => {
 });
 
 describe("hired agents sharing a subscription", () => {
+  it("keeps a credential-lock timeout on automatic retry without blocking the task or starting a provider", async () => {
+    const f = await fixture("openai", "subscription");
+    const agent = hired(await request(f.app).post(`/api/companies/${f.companyId}/agent-hires`).send({ name: "Waiting teammate", role: "engineer", adapterType: f.adapterType, reportsTo: f.agentId, adapterConfig: { cwd: home, engine: "cli" }, runtimeConfig: { heartbeat: { enabled: false } } }));
+    const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Wait for credential rotation", status: "todo", assigneeAgentId: agent.id, responsibleUserId: f.userId, createdByUserId: f.userId }).returning();
+    const execute = vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false, resultJson: {} }));
+    registerServerAdapter({ ...getServerAdapter(f.adapterType), execute });
+    const lock = vi.spyOn(codexAdapter, "withAccountHomeSecretMutationLock")
+      .mockRejectedValueOnce(Object.assign(new Error("quota refresh holds company lock"), { code: WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE }));
+    const heartbeat = heartbeatService(db);
+    try {
+      const run = await heartbeat.invoke(agent.id, "assignment", { issueId: issue.id, wakeReason: "issue_assigned", responsibleUserId: f.userId }, "system");
+      expect(run).not.toBeNull();
+      await expect.poll(async () => (await heartbeat.getRun(run!.id))?.status, { timeout: 20_000 }).toBe("cancelled");
+      await heartbeat.drainActiveRunExecutions();
+      expect(await heartbeat.getRun(run!.id)).toMatchObject({ errorCode: "ai_connection_busy", resultJson: { executionRecovery: { providerWorkStarted: false } } });
+      const retries = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id));
+      expect(retries).toHaveLength(1);
+      expect(retries[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "ai_connection_busy" });
+      const [savedIssue] = await db.select().from(issues).where(eq(issues.id, issue.id));
+      expect(savedIssue.status).not.toBe("blocked");
+      expect(savedIssue.executionRunId).toBe(retries[0].id);
+      expect(execute).not.toHaveBeenCalled();
+      expect(await aiConnectionService(db).list(f.companyId, f.userId)).toEqual([expect.objectContaining({ status: "connected" })]);
+    } finally {
+      lock.mockRestore();
+      await heartbeat.drainActiveRunExecutions();
+      unregisterServerAdapter(f.adapterType);
+    }
+  });
+
   it.each(["openai", "anthropic"] as const)("runs the %s child alongside a live parent and inherits its connection", async (provider) => {
     const f = await fixture(provider, "subscription");
     const agent = hired(await request(f.app).post(`/api/companies/${f.companyId}/agent-hires`).send({ name: "Concurrent teammate", role: "engineer", adapterType: f.adapterType, reportsTo: f.agentId, adapterConfig: { cwd: home, engine: "cli" }, runtimeConfig: { heartbeat: { enabled: false } } }));

server/src/__tests__/ai-connections.test.ts

@@ -16,7 +16,7 @@ import { aiConnectionService } from "../services/ai-connections.js";
 import * as codexAdapter from "@paperclipai/adapter-codex-local/server";
 import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
 import * as executionTarget from "@paperclipai/adapter-utils/execution-target";
-import { prepareManagedAiRuntime, assertManagedAiProjectAuth } from "../services/ai-connection-runtime.js";
+import { prepareManagedAiRuntime, assertManagedAiProjectAuth, isAiConnectionBusy } from "../services/ai-connection-runtime.js";
 import { toolAccessService } from "../services/tool-access.js";
 import { secretService } from "../services/secrets.js";
 import { aiConnectionBindingSchema, connectionPurposeTransportSchema, isAiConnectionCompatible } from "@paperclipai/shared";
@@ -143,6 +143,35 @@ describe("managed AI connections", () => {
     }
   });
 
+  it.each(["company_file", "grant_row", "secret_row"] as const)("defers runtime preparation on a contended %s lock and succeeds once it clears", async (kind) => {
+    const owner = `runtime-lock-${kind}`;
+    await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
+    const auth = JSON.stringify({ tokens: { access_token: "valid-access", refresh_token: "valid-refresh", account_id: owner } });
+    await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, auth);
+    const [row] = await service.quotaAccounts(companyId, owner);
+    const ref = row.grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!;
+    const prepare = () => prepareManagedAiRuntime(db, { companyId, agentId, responsibleUserId: owner, adapterType: "codex_local",
+      binding: { provider: "openai", method: "subscription", mode: "responsible_user" }, config: {} });
+    const held = deferredSignal(), release = deferredSignal();
+    const lock = kind === "company_file" ? vi.spyOn(codexAdapter, "withAccountHomeSecretMutationLock")
+      .mockRejectedValueOnce(Object.assign(new Error("company lock held by another account"), { code: WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE })) : undefined;
+    const holding = kind === "company_file" ? Promise.resolve() : db.transaction(async tx => {
+      if (kind === "grant_row") await tx.select().from(connectionGrants).where(eq(connectionGrants.id, row.grant.id)).for("update");
+      else await tx.select().from(companySecrets).where(eq(companySecrets.id, ref.secretId)).for("update");
+      held.resolve(); await release.promise;
+    });
+    try {
+      if (kind !== "company_file") await held.promise;
+      const error = await prepare().then(() => { throw new Error("Runtime must defer"); }, error => error);
+      expect(isAiConnectionBusy(error)).toBe(true);
+      expect(error).toMatchObject({ status: 422, details: { code: "ai_connection_busy" } });
+      expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, row.grant.id)))[0].status).toBe("active");
+    } finally { release.resolve(); await holding; lock?.mockRestore(); }
+    const run = await prepare();
+    try { expect(await readFile(path.join(String(run.config.env.CODEX_HOME), "auth.json"), "utf8")).toBe(auth); }
+    finally { await run.cleanup(); }
+  });
+
   it.each(["runtime", "failed_run"] as const)("recovers pending quota credentials before %s can reuse or invalidate the consumed token", async (consumer) => {
     const owner = `quota-consumer-${consumer}`;
     await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });

server/src/__tests__/cost-accounting-reliability.test.ts

@@ -41,6 +41,49 @@ databaseDescribe("cost accounting reliability (PostgreSQL)", () => {
     return { company, agent, project, issue, goal, run, receipt };
   }
 
+  it("counts cache reads once across mixed historical and receipt-backed reports without changing charges", async () => {
+    const f = await fixture();
+    const costs = costService(db);
+    const scope = { agentId: f.agent.id, issueId: f.issue.id, projectId: f.project.id, heartbeatRunId: f.run.id,
+      provider: "openai", biller: "openai", model: "gpt-test", occurredAt: new Date() };
+    // These two layouts both describe 100 input (80 cached) and 10 output.
+    for (const billingType of ["metered_api", "subscription_included"]) {
+      await db.insert(costEvents).values({ companyId: f.company.id, ...scope, billingType,
+        inputTokens: 100, cachedInputTokens: 80, outputTokens: 10, costCents: 7 });
+      await costs.createEvent(f.company.id, { ...scope, billingType,
+        inputTokens: 20, cachedInputTokens: 80, outputTokens: 10, costCents: 11 });
+    }
+    const [provider, models] = await Promise.all([costs.byProvider(f.company.id), costs.byAgentModel(f.company.id)]);
+    for (const rows of [provider, models]) {
+      expect(rows).toHaveLength(2);
+      for (const row of rows) {
+        expect(row).toMatchObject({ inputTokens: 40, cachedInputTokens: 160, outputTokens: 20, costCents: 18 });
+        expect(row.inputTokens + row.cachedInputTokens + row.outputTokens).toBe(220);
+      }
+    }
+    const [agent] = await costs.byAgent(f.company.id);
+    const [biller] = await costs.byBiller(f.company.id);
+    const [project] = await costs.byProject(f.company.id);
+    const issue = await costs.issueTreeSummary(f.company.id, f.issue.id);
+    const windows = await costs.windowSpend(f.company.id);
+    expect(windows).toHaveLength(3);
+    for (const row of [agent, biller, project, issue, ...windows]) {
+      expect(row).toMatchObject({ inputTokens: 80, cachedInputTokens: 320, outputTokens: 40, costCents: 36 });
+      expect(row.inputTokens + row.cachedInputTokens + row.outputTokens).toBe(440);
+    }
+    for (const row of [agent, biller, provider.find(row => row.billingType === "subscription_included")!]) {
+      expect(row).toMatchObject({ subscriptionInputTokens: 40, subscriptionCachedInputTokens: 160, subscriptionOutputTokens: 20 });
+    }
+    // Reading reports never rewrites historical evidence or retroactively bills.
+    const stored = await db.select().from(costEvents).where(eq(costEvents.companyId, f.company.id));
+    expect(stored.filter(row => row.receiptHash === null).map(row => row.inputTokens)).toEqual([100, 100]);
+    expect((await costs.summary(f.company.id)).spendCents).toBe(36);
+    // Legacy Anthropic input was already exclusive; retain its cache semantics.
+    await db.insert(costEvents).values({ companyId: f.company.id, ...scope, provider: "anthropic", billingType: "metered_api",
+      inputTokens: 100, cachedInputTokens: 80, outputTokens: 10, costCents: 0 });
+    expect((await costs.byProvider(f.company.id)).find(row => row.provider === "anthropic")).toMatchObject({ inputTokens: 100, cachedInputTokens: 80, outputTokens: 10 });
+  });
+
   it.each(["issueId", "projectId", "goalId", "heartbeatRunId"] as const)("rejects a foreign-company %s without writing anything", async (field) => {
     const a = await fixture(); const b = await fixture();
     const foreignIds = { issueId: b.issue.id, projectId: b.project.id, goalId: b.goal.id, heartbeatRunId: b.run.id };

server/src/__tests__/costs-service.test.ts

@@ -551,7 +551,9 @@ describeEmbeddedPostgres("cost and finance aggregate overflow handling", () => {
     const [byAgentModelRow] = await costs.byAgentModel(companyId, range);
 
     expect(byAgentRow?.costCents).toBe(4_000_000_000);
-    expect(byAgentRow?.inputTokens).toBe(4_000_000_000);
+    // Historical OpenAI input includes these ten cached tokens.
+    expect(byAgentRow?.inputTokens).toBe(3_999_999_990);
+    expect(byAgentRow!.inputTokens + byAgentRow!.cachedInputTokens).toBe(4_000_000_000);
     expect(byProjectRow?.costCents).toBe(4_000_000_000);
     expect(byAgentModelRow?.costCents).toBe(4_000_000_000);
   });
@@ -701,7 +703,7 @@ describeEmbeddedPostgres("cost and finance aggregate overflow handling", () => {
       includeDescendants: true,
       costCents: 600,
       costCentsExact: "600.0000000",
-      inputTokens: 60,
+      inputTokens: 54,
       cachedInputTokens: 6,
       outputTokens: 12,
       runCount: 0,

server/src/services/ai-connections.ts

@@ -1,3 +1,4 @@
+import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
 import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server";
 import { syncConnectionCredentialBindings } from "./connection-credential-bindings.js";
 import { createHash, randomUUID } from "node:crypto";
@@ -861,6 +862,18 @@ export function aiConnectionService(db: Db) {
       if (recovered) await quotaCredentialRecovery(row.connection.companyId, row.grant.id).clear()
         .catch(error => logger.warn({ err: error, grantId: row.grant.id }, "Saved runtime credential recovery cleanup pending"));
       return value;
+    }).catch(error => {
+      // Drizzle wraps Postgres errors in `cause`. Contention is a temporary
+      // pre-provider wait, not evidence that this account needs reconnecting.
+      let current: unknown = error;
+      for (let depth = 0; depth < 4 && current && typeof current === "object"; depth += 1) {
+        const value = current as { code?: unknown; cause?: unknown };
+        if (value.code === WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE || value.code === "55P03") {
+          throw unprocessable("AI credentials are being updated. This execution will retry automatically.", { code: "ai_connection_busy" });
+        }
+        current = value.cause;
+      }
+      throw error;
     });
   }
 

server/src/services/costs.ts

@@ -16,7 +16,15 @@ export type { CostDateRange } from "./cost-date-range.js";
 const METERED_BILLING_TYPE = "metered_api";
 const SUBSCRIPTION_BILLING_TYPES = ["subscription_included", "subscription_overage"] as const;
 
-function sumAsNumber(column: typeof costEvents.costCents | typeof costEvents.inputTokens | typeof costEvents.cachedInputTokens | typeof costEvents.outputTokens) {
+// Pre-receipt OpenAI events stored cache reads inside input. Normalize each
+// historical row before aggregation so mixed old/new groups share the current
+// exclusive-input contract. Preserve the original ledger and monetary amounts.
+const ordinaryInputTokens = sql<number>`case
+  when ${costEvents.receiptHash} is null and ${costEvents.provider} = 'openai'
+    then greatest(0, ${costEvents.inputTokens} - ${costEvents.cachedInputTokens})
+  else ${costEvents.inputTokens} end`;
+
+function sumAsNumber(column: typeof costEvents.costCents | typeof costEvents.inputTokens | typeof costEvents.cachedInputTokens | typeof costEvents.outputTokens | ReturnType<typeof sql>) {
   return sql<number>`coalesce(sum(${column}), 0)::double precision`;
 }
 
@@ -305,7 +313,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
             issueCount: sql<number>`count(distinct ${issues.id})::int`,
             costCents: sumAsNumber(costEvents.costCents),
           costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
-            inputTokens: sumAsNumber(costEvents.inputTokens),
+            inputTokens: sumAsNumber(ordinaryInputTokens),
             cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
             outputTokens: sumAsNumber(costEvents.outputTokens),
           })
@@ -362,7 +370,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
           costCents: sumAsNumber(costEvents.costCents),
           costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
-          inputTokens: sumAsNumber(costEvents.inputTokens),
+          inputTokens: sumAsNumber(ordinaryInputTokens),
           cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
           outputTokens: sumAsNumber(costEvents.outputTokens),
           apiRunCount:
@@ -372,7 +380,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           subscriptionCachedInputTokens:
             sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`,
           subscriptionInputTokens:
-            sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.inputTokens} else 0 end), 0)::double precision`,
+            sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`,
           subscriptionOutputTokens:
             sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`,
         })
@@ -401,7 +409,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           model: costEvents.model,
           costCents: sumAsNumber(costEvents.costCents),
           costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
-          inputTokens: sumAsNumber(costEvents.inputTokens),
+          inputTokens: sumAsNumber(ordinaryInputTokens),
           cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
           outputTokens: sumAsNumber(costEvents.outputTokens),
           apiRunCount:
@@ -411,7 +419,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           subscriptionCachedInputTokens:
             sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`,
           subscriptionInputTokens:
-            sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.inputTokens} else 0 end), 0)::double precision`,
+            sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`,
           subscriptionOutputTokens:
             sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`,
         })
@@ -432,7 +440,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           biller: costEvents.biller,
           costCents: sumAsNumber(costEvents.costCents),
           costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
-          inputTokens: sumAsNumber(costEvents.inputTokens),
+          inputTokens: sumAsNumber(ordinaryInputTokens),
           cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
           outputTokens: sumAsNumber(costEvents.outputTokens),
           apiRunCount:
@@ -442,7 +450,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           subscriptionCachedInputTokens:
             sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.cachedInputTokens} else 0 end), 0)::double precision`,
           subscriptionInputTokens:
-            sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.inputTokens} else 0 end), 0)::double precision`,
+            sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${ordinaryInputTokens} else 0 end), 0)::double precision`,
           subscriptionOutputTokens:
             sql<number>`coalesce(sum(case when ${costEvents.billingType} in (${sql.join(SUBSCRIPTION_BILLING_TYPES.map((value) => sql`${value}`), sql`, `)}) then ${costEvents.outputTokens} else 0 end), 0)::double precision`,
           providerCount: sql<number>`count(distinct ${costEvents.provider})::int`,
@@ -476,7 +484,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
               biller: sql<string>`case when count(distinct ${costEvents.biller}) = 1 then min(${costEvents.biller}) else 'mixed' end`,
               costCents: sumAsNumber(costEvents.costCents),
           costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
-              inputTokens: sumAsNumber(costEvents.inputTokens),
+              inputTokens: sumAsNumber(ordinaryInputTokens),
               cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
               outputTokens: sumAsNumber(costEvents.outputTokens),
             })
@@ -531,7 +539,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           estimatedEventCount: sql<number>`count(*) filter (where ${costEvents.costStatus} = 'estimated')::int`,
           costCents: sumAsNumber(costEvents.costCents),
           costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
-          inputTokens: sumAsNumber(costEvents.inputTokens),
+          inputTokens: sumAsNumber(ordinaryInputTokens),
           cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
           outputTokens: sumAsNumber(costEvents.outputTokens),
         })
@@ -595,7 +603,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
           projectName: projects.name,
           costCents: costCentsExpr,
           costCentsExact: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
-          inputTokens: sumAsNumber(costEvents.inputTokens),
+          inputTokens: sumAsNumber(ordinaryInputTokens),
           cachedInputTokens: sumAsNumber(costEvents.cachedInputTokens),
           outputTokens: sumAsNumber(costEvents.outputTokens),
         })

server/src/services/heartbeat.ts

@@ -16375,9 +16375,8 @@ export function heartbeatService(
       .then((rows) => rows[0] ?? null);
   }
 
-  // No code path raises the `ai_connection_busy` error any more. This
-  // function stays because a stored run row can still carry that error code
-  // from an earlier release.
+  // Credential rotation can briefly contend with a fresh runtime read. Keep
+  // the task on its automatic pre-provider retry path while the lock clears.
   async function finalizeAiConnectionBusyDeferral(
     run: typeof heartbeatRuns.$inferSelect,
     error: HttpError,