Kilo-Org · jeanduplessis · Jul 16, 2026 · Jul 16, 2026
diff --git a/apps/web/src/app/api/cron/cost-insights-hourly/route.test.ts b/apps/web/src/app/api/cron/cost-insights-hourly/route.test.ts
@@ -30,6 +30,11 @@ function summary(
       repaired: 0,
       failed: [],
     },
+    dailyUsageRollupRepairs: {
+      claimed: 0,
+      repaired: 0,
+      failed: [],
+    },
     notifications: {
       claimed: 1,
       sent: 1,
@@ -141,6 +146,31 @@ describe('GET /api/cron/cost-insights-hourly', () => {
     );
   });
 
+  test('returns failure status and telemetry when a daily usage rollup repair fails', async () => {
+    const result = summary();
+    result.dailyUsageRollupRepairs.failed.push({
+      usageId: '00000000-0000-4000-8000-000000000001',
+      kiloUserId: 'user-3',
+      organizationId: null,
+      usageDate: '2026-07-13',
+      error: 'postgres:55P03',
+    });
+    mockRunCostInsightHourlySweep.mockResolvedValue(result);
+
+    const response = await GET(
+      new NextRequest('http://localhost:3000/api/cron/cost-insights-hourly', {
+        headers: { authorization: 'Bearer cron-secret' },
+      })
+    );
+
+    expect(response.status).toBe(500);
+    await expect(response.json()).resolves.toMatchObject({ success: false, partialFailure: true });
+    expect(mockSentryLog).toHaveBeenCalledWith(
+      'Cost Insights hourly sweep completed with partial failures',
+      expect.objectContaining({ failedDailyUsageRollupRepairCount: 1 })
+    );
+  });
+
   test('exports a bounded function duration for resumable sweeps', () => {
     expect(maxDuration).toBe(300);
   });

diff --git a/apps/web/src/app/api/cron/cost-insights-hourly/route.ts b/apps/web/src/app/api/cron/cost-insights-hourly/route.ts
@@ -26,6 +26,7 @@ export async function GET(request: Request) {
   }
 
   const summary = await runCostInsightHourlySweep(db);
+  const dailyUsageRollupRepairs = summary.dailyUsageRollupRepairs;
   sentryLogger('cron', 'info')('Cost Insights hourly sweep completed', {
     evaluatedOwnerCount: summary.evaluatedOwners,
     failedOwnerCount: summary.failedOwners.length,
@@ -35,6 +36,9 @@ export async function GET(request: Request) {
     rollupRepairClaimedCount: summary.rollupRepairs.claimed,
     rollupRepairCompletedCount: summary.rollupRepairs.repaired,
     rollupRepairFailedCount: summary.rollupRepairs.failed.length,
+    dailyUsageRollupRepairClaimedCount: dailyUsageRollupRepairs.claimed,
+    dailyUsageRollupRepairCompletedCount: dailyUsageRollupRepairs.repaired,
+    dailyUsageRollupRepairFailedCount: dailyUsageRollupRepairs.failed.length,
     evaluationDurationMs: summary.evaluationDurationMs,
     rawCanonicalFallbackCount: summary.rawCanonicalFallbackCount,
     rollupDegradedIntervalCount: summary.rollupDegradedIntervalCount,
@@ -46,12 +50,14 @@ export async function GET(request: Request) {
   const hasFailures =
     summary.failedOwners.length > 0 ||
     summary.rollupRepairs.failed.length > 0 ||
+    dailyUsageRollupRepairs.failed.length > 0 ||
     summary.notifications.failed > 0 ||
     summary.notifications.terminalized > 0;
   if (hasFailures) {
     sentryLogger('cron', 'error')('Cost Insights hourly sweep completed with partial failures', {
       failedOwnerCount: summary.failedOwners.length,
       failedRollupRepairCount: summary.rollupRepairs.failed.length,
+      failedDailyUsageRollupRepairCount: dailyUsageRollupRepairs.failed.length,
       failedNotificationCount: summary.notifications.failed,
       terminalizedNotificationCount: summary.notifications.terminalized,
     });

diff --git a/apps/web/src/lib/ai-gateway/processUsage.test.ts b/apps/web/src/lib/ai-gateway/processUsage.test.ts
@@ -11,6 +11,7 @@ import {
   processOpenRouterUsage,
   stripNulBytesInPlace,
   toInsertableDbUsageRecord,
+  usageTransactionIdleTimeoutQuery,
 } from './processUsage';
 import type { OpenRouterGeneration } from '@/lib/ai-gateway/providers/openrouter/types';
 import { verifyApproval } from '../../tests/helpers/approval.helper';
@@ -30,15 +31,17 @@ import {
   cost_insight_rollup_repairs,
   microdollar_usage,
   microdollar_usage_daily,
+  microdollar_usage_daily_repairs,
   microdollar_usage_metadata,
   organization_user_usage,
   organizations,
 } from '@kilocode/db/schema';
-import { and, eq, getTableColumns, isNull, sql } from 'drizzle-orm';
+import { eq, getTableColumns } from 'drizzle-orm';
 import { findUserById } from '../user';
 import { Readable } from 'node:stream';
 import { getFraudDetectionHeaders, toMicrodollars } from '../utils';
 import { createTestOrganization } from '@/tests/helpers/organization.helper';
+import { PgDialect } from 'drizzle-orm/pg-core';
 
 // Note: Legacy banned_ja4/whitelist_ja4 tests removed - abuse classification
 // is now handled by the external abuse detection service (src/lib/abuse-service.ts)
@@ -98,6 +101,16 @@ describe('processOpenRouterUsage', () => {
   });
 });
 
+describe('authoritative usage transaction configuration', () => {
+  test('sets a transaction-local 60-second idle timeout', async () => {
+    const compiled = new PgDialect().sqlToQuery(usageTransactionIdleTimeoutQuery());
+
+    expect(compiled.sql).toContain("'idle_in_transaction_session_timeout'");
+    expect(compiled.sql).toContain('true');
+    expect(compiled.params).toEqual(['60000ms']);
+  });
+});
+
 const sampleDir = join(process.cwd(), 'src/tests/sample');
 describe('parseMicrodollarUsageFromStream approval tests', () => {
   const normalAnthropic = 'normal-anthropic.log.resp.sse';
@@ -955,7 +968,7 @@ describe('logMicrodollarUsage', () => {
     expect(updatedUser?.microdollars_used).toBe(4000); // unchanged
   });
 
-  test('insertUsageRecord populates microdollar_usage_daily for personal usage', async () => {
+  test('insertUsageRecord enqueues durable daily-rollup work without mutating the daily table', async () => {
     const user = await insertTestUser({
       id: 'test-daily-personal-user',
       microdollars_used: 0,
@@ -967,86 +980,101 @@ describe('logMicrodollarUsage', () => {
       cost: 1500,
     });
 
+    const [repair] = await db
+      .select()
+      .from(microdollar_usage_daily_repairs)
+      .where(eq(microdollar_usage_daily_repairs.kilo_user_id, user.id));
+    expect(repair).toMatchObject({
+      kilo_user_id: user.id,
+      organization_id: null,
+      usage_date: '2025-08-15',
+      attempt_count: 0,
+    });
+    expect((await findUserById(user.id))?.microdollars_used).toBe(1500);
     const dailyRows = await db
       .select()
       .from(microdollar_usage_daily)
-      .where(
-        and(
-          eq(microdollar_usage_daily.kilo_user_id, user.id),
-          isNull(microdollar_usage_daily.organization_id)
-        )
-      );
-
-    expect(dailyRows).toHaveLength(1);
-    expect(dailyRows[0].total_cost_microdollars).toBe(1500);
-    expect(dailyRows[0].organization_id).toBeNull();
+      .where(eq(microdollar_usage_daily.kilo_user_id, user.id));
+    expect(dailyRows).toHaveLength(0);
   });
 
-  test('insertUsageRecord increments microdollar_usage_daily on subsequent inserts on the same day', async () => {
+  test('does not enqueue or apply a daily repair when the authoritative usage insert fails', async () => {
+    const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(() => {});
     const user = await insertTestUser({
-      id: 'test-daily-increment-user',
-      microdollars_used: 0,
-      google_user_email: 'daily-increment@example.com',
+      id: 'test-daily-authoritative-failure-user',
+      microdollars_used: 1000,
+      google_user_email: 'daily-authoritative-failure@example.com',
     });
+    const { core, metadata } = await defineMicrodollarUsage();
+    const invalidUsageId = 'not-a-usage-uuid';
 
-    await insertUsageWithOverrides({ kilo_user_id: user.id, cost: 1000 });
-    await insertUsageWithOverrides({ kilo_user_id: user.id, cost: 2500 });
-    await insertUsageWithOverrides({ kilo_user_id: user.id, cost: 700 });
-
-    const [row] = await db
-      .select({
-        total: sql<number>`coalesce(sum(${microdollar_usage_daily.total_cost_microdollars}), 0)::int`,
-      })
-      .from(microdollar_usage_daily)
-      .where(
-        and(
-          eq(microdollar_usage_daily.kilo_user_id, user.id),
-          isNull(microdollar_usage_daily.organization_id)
+    try {
+      await expect(
+        insertUsageRecord(
+          { ...core, id: invalidUsageId, kilo_user_id: user.id, cost: 600 },
+          metadata
         )
-      );
+      ).resolves.toBeNull();
 
-    expect(row.total).toBe(4200);
+      const sourceRows = await db
+        .select()
+        .from(microdollar_usage)
+        .where(eq(microdollar_usage.kilo_user_id, user.id));
+      expect(sourceRows).toHaveLength(0);
+      expect((await findUserById(user.id))?.microdollars_used).toBe(1000);
+
+      const repairRows = await db
+        .select()
+        .from(microdollar_usage_daily_repairs)
+        .where(eq(microdollar_usage_daily_repairs.kilo_user_id, user.id));
+      expect(repairRows).toHaveLength(0);
+      const dailyRows = await db
+        .select()
+        .from(microdollar_usage_daily)
+        .where(eq(microdollar_usage_daily.kilo_user_id, user.id));
+      expect(dailyRows).toHaveLength(0);
+    } finally {
+      consoleErrorSpy.mockRestore();
+    }
   });
 
-  test('insertUsageRecord writes org-scoped rollup separately from personal rollup', async () => {
+  test('rolls negative usage into its UTC day without changing the personal balance', async () => {
     const user = await insertTestUser({
-      id: 'test-daily-org-scope-user',
-      microdollars_used: 0,
-      google_user_email: 'daily-org-scope@example.com',
-    });
-    const organization = await createTestOrganization('Daily usage organization', user.id, 10_000);
-    const orgId = organization.id;
-
-    await insertUsageWithOverrides({ kilo_user_id: user.id, cost: 500 });
-    await insertUsageWithOverrides({
-      kilo_user_id: user.id,
-      organization_id: orgId,
-      cost: 9000,
+      id: 'test-daily-negative-utc-user',
+      microdollars_used: 1000,
+      google_user_email: 'daily-negative-utc@example.com',
     });
+    const { core: beforeMidnightCore, metadata: beforeMidnightMetadata } =
+      await defineMicrodollarUsage();
+    const { core: afterMidnightCore, metadata: afterMidnightMetadata } =
+      await defineMicrodollarUsage();
+
+    await insertUsageRecord(
+      {
+        ...beforeMidnightCore,
+        kilo_user_id: user.id,
+        cost: -250,
+        created_at: '2025-08-15T23:59:59.999Z',
+      },
+      beforeMidnightMetadata
+    );
+    await insertUsageRecord(
+      {
+        ...afterMidnightCore,
+        kilo_user_id: user.id,
+        cost: 400,
+        created_at: '2025-08-16T00:00:00.000Z',
+      },
+      afterMidnightMetadata
+    );
 
-    const personalRows = await db
-      .select()
-      .from(microdollar_usage_daily)
-      .where(
-        and(
-          eq(microdollar_usage_daily.kilo_user_id, user.id),
-          isNull(microdollar_usage_daily.organization_id)
-        )
-      );
-    expect(personalRows).toHaveLength(1);
-    expect(personalRows[0].total_cost_microdollars).toBe(500);
-
-    const orgRows = await db
-      .select()
-      .from(microdollar_usage_daily)
-      .where(
-        and(
-          eq(microdollar_usage_daily.kilo_user_id, user.id),
-          eq(microdollar_usage_daily.organization_id, orgId)
-        )
-      );
-    expect(orgRows).toHaveLength(1);
-    expect(orgRows[0].total_cost_microdollars).toBe(9000);
+    expect((await findUserById(user.id))?.microdollars_used).toBe(1400);
+    const repairRows = await db
+      .select({ usage_date: microdollar_usage_daily_repairs.usage_date })
+      .from(microdollar_usage_daily_repairs)
+      .where(eq(microdollar_usage_daily_repairs.kilo_user_id, user.id))
+      .orderBy(microdollar_usage_daily_repairs.usage_date);
+    expect(repairRows).toEqual([{ usage_date: '2025-08-15' }, { usage_date: '2025-08-16' }]);
   });
 
   test('insertUsageRecord skips microdollar_usage_daily for zero-cost rows', async () => {

diff --git a/apps/web/src/lib/ai-gateway/processUsage.ts b/apps/web/src/lib/ai-gateway/processUsage.ts
@@ -93,6 +93,7 @@ import {
   getAiGatewayCostInsightProductKey,
 } from '@/lib/cost-insights/canonical-sources';
 import { scheduleCostInsightEvaluationAfterSpend } from '@/lib/cost-insights/evaluation';
+import { enqueueDailyUsageRollupRepair } from './usage-daily-rollup-repairs';
 
 const posthogClient = PostHogClient();
 
@@ -443,11 +444,29 @@ type UsageTransactionResult = {
   inserted: UsageStatementResult;
 };
 
+export const USAGE_TRANSACTION_IDLE_TIMEOUT_MS = 60_000;
+
+export function usageTransactionIdleTimeoutQuery(): SQL {
+  return sql`
+    SELECT pg_catalog.set_config(
+      'idle_in_transaction_session_timeout',
+      ${`${USAGE_TRANSACTION_IDLE_TIMEOUT_MS}ms`},
+      true
+    )
+  `;
+}
+
+export async function setUsageTransactionIdleTimeout(tx: UsageStatementExecutor): Promise<void> {
+  // This only bounds idle time between statements while the transaction is open.
+  await tx.execute(usageTransactionIdleTimeoutQuery());
+}
+
 async function insertUsageTransaction(
   coreUsageFields: MicrodollarUsage,
   metadataFields: UsageMetaData
 ): Promise<UsageTransactionResult> {
   return db.transaction(async tx => {
+    await setUsageTransactionIdleTimeout(tx);
     if (coreUsageFields.cost > 0) {
       const owner = coreUsageFields.organization_id
         ? { type: 'organization' as const, id: coreUsageFields.organization_id }
@@ -463,6 +482,14 @@ async function insertUsageTransaction(
       coreUsageFields,
       metadataFields
     );
+    if (coreUsageFields.cost !== 0) {
+      await enqueueDailyUsageRollupRepair(tx, {
+        usageId: coreUsageFields.id,
+        kiloUserId: coreUsageFields.kilo_user_id,
+        organizationId: coreUsageFields.organization_id,
+        createdAt: coreUsageFields.created_at,
+      });
+    }
     if (coreUsageFields.cost > 0) {
       const owner = coreUsageFields.organization_id
         ? { type: 'organization' as const, id: coreUsageFields.organization_id }
@@ -715,15 +742,8 @@ async function insertUsageAndMetadataWithBalanceUpdate(
   coreUsageFields: MicrodollarUsage,
   metadataFields: UsageMetaData
 ): Promise<UsageStatementResult> {
-  // Pick the matching partial unique index for the daily-rollup upsert. The
-  // microdollar_usage_daily table has two partial unique indexes; the upsert
-  // must target the one corresponding to this row's scope.
-  const dailyConflictTarget =
-    coreUsageFields.organization_id === null
-      ? sql`(kilo_user_id, usage_date) WHERE organization_id IS NULL`
-      : sql`(kilo_user_id, organization_id, usage_date) WHERE organization_id IS NOT NULL`;
-
-  // Use a single SQL statement with CTEs to insert usage, upsert all lookup values, metadata, and update user balance in one roundtrip
+  // Use a single SQL statement with CTEs to insert usage, upsert all lookup values, metadata, and update user balance in one roundtrip.
+  // The contended daily rollup is deliberately maintained after this transaction commits.
   // This ensures atomicity: microdollar_usage insert and kilocode_users.microdollars_used update happen together
   const result = await executor.execute<{
     usage_id: string;
@@ -851,22 +871,6 @@ async function insertUsageAndMetadataWithBalanceUpdate(
               (SELECT mode_id FROM mode_cte),
               (SELECT auto_model_id FROM auto_model_cte)
           )
-          , microdollar_usage_daily_upsert AS (
-            INSERT INTO microdollar_usage_daily (
-              kilo_user_id, organization_id, usage_date, total_cost_microdollars
-            )
-            SELECT
-              ${coreUsageFields.kilo_user_id},
-              ${coreUsageFields.organization_id}::uuid,
-              date_trunc('day', ${coreUsageFields.created_at}::timestamptz)::date,
-              ${coreUsageFields.cost}::bigint
-            WHERE ${coreUsageFields.cost} <> 0
-            ON CONFLICT ${dailyConflictTarget}
-            DO UPDATE SET
-              total_cost_microdollars =
-                microdollar_usage_daily.total_cost_microdollars + EXCLUDED.total_cost_microdollars,
-              updated_at = NOW()
-          )
           , balance_update AS (
             UPDATE kilocode_users
             SET microdollars_used = microdollars_used + ${coreUsageFields.cost}