fix(cost-insights): bound rollup transaction lifecycle

apps/web/src/app/api/cron/cost-insights-hourly/route.test.ts

@@ -6,11 +6,14 @@ const mockSentryLog = jest.fn();
 jest.mock('@/lib/utils.server', () => ({ sentryLogger: jest.fn(() => mockSentryLog) }));
 
 import { runCostInsightHourlySweep } from '@/lib/cost-insights/jobs';
+import type { CostInsightHourlySweepSummary } from '@/lib/cost-insights/jobs';
 import { GET, maxDuration } from './route';
 
 const mockRunCostInsightHourlySweep = jest.mocked(runCostInsightHourlySweep);
 
-function summary(failedOwners: Array<{ owner: { type: 'user'; id: string }; error: string }> = []) {
+function summary(
+  failedOwners: Array<{ owner: { type: 'user'; id: string }; error: string }> = []
+): CostInsightHourlySweepSummary {
   return {
     evaluatedOwners: 2,
     failedOwners,
@@ -22,6 +25,11 @@ function summary(failedOwners: Array<{ owner: { type: 'user'; id: string }; erro
       rawCanonicalFallbackCount: 0,
       rollupDegradedIntervalCount: 0,
     },
+    rollupRepairs: {
+      claimed: 0,
+      repaired: 0,
+      failed: [],
+    },
     notifications: {
       claimed: 1,
       sent: 1,
@@ -110,6 +118,29 @@ describe('GET /api/cron/cost-insights-hourly', () => {
     );
   });
 
+  test('returns failure status when a rollup repair fails', async () => {
+    const result = summary();
+    result.rollupRepairs.failed.push({
+      owner: { type: 'user', id: 'user-3' },
+      hourStart: '2026-07-13T10:00:00.000Z',
+      error: 'repair failed',
+    });
+    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({ failedRollupRepairCount: 1 })
+    );
+  });
+
   test('exports a bounded function duration for resumable sweeps', () => {
     expect(maxDuration).toBe(300);
   });

apps/web/src/app/api/cron/cost-insights-hourly/route.ts

@@ -32,6 +32,9 @@ export async function GET(request: Request) {
     dirtyQueueDepthBefore: summary.dirtyQueueDepthBefore,
     dirtyQueueDepthAfter: summary.dirtyQueueDepthAfter,
     dirtyClaimedCount: summary.dirtyEvaluations.claimed,
+    rollupRepairClaimedCount: summary.rollupRepairs.claimed,
+    rollupRepairCompletedCount: summary.rollupRepairs.repaired,
+    rollupRepairFailedCount: summary.rollupRepairs.failed.length,
     evaluationDurationMs: summary.evaluationDurationMs,
     rawCanonicalFallbackCount: summary.rawCanonicalFallbackCount,
     rollupDegradedIntervalCount: summary.rollupDegradedIntervalCount,
@@ -42,11 +45,13 @@ export async function GET(request: Request) {
   });
   const hasFailures =
     summary.failedOwners.length > 0 ||
+    summary.rollupRepairs.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,
       failedNotificationCount: summary.notifications.failed,
       terminalizedNotificationCount: summary.notifications.terminalized,
     });

apps/web/src/lib/ai-gateway/processUsage.test.ts

@@ -26,6 +26,8 @@ import { db } from '@/lib/drizzle';
 import {
   cost_insight_owner_hour_driver_buckets,
   cost_insight_owner_hour_totals,
+  cost_insight_rollup_degraded_intervals,
+  cost_insight_rollup_repairs,
   microdollar_usage,
   microdollar_usage_daily,
   microdollar_usage_metadata,
@@ -587,7 +589,7 @@ describe('logMicrodollarUsage', () => {
     expect(totals).toHaveLength(0);
   });
 
-  test('keeps AI source rows when post-commit cost insight capture fails', async () => {
+  test('keeps AI source rows when legacy usage owner does not exist', async () => {
     const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(() => {});
     const { core, metadata } = await defineMicrodollarUsage();
     const missingUserId = `missing-ai-user-${crypto.randomUUID()}`;
@@ -613,29 +615,38 @@ describe('logMicrodollarUsage', () => {
         .from(cost_insight_owner_hour_totals)
         .where(eq(cost_insight_owner_hour_totals.owned_by_user_id, missingUserId));
       expect(totals).toHaveLength(0);
+
+      const degradedIntervals = await db
+        .select()
+        .from(cost_insight_rollup_degraded_intervals)
+        .where(eq(cost_insight_rollup_degraded_intervals.source, 'ai_gateway'));
+      expect(degradedIntervals).toHaveLength(0);
     } finally {
       consoleErrorSpy.mockRestore();
     }
   });
 
-  test('keeps usage and balance write when post-commit cost insight capture fails', async () => {
+  test('keeps usage and balance write and enqueues owner-hour repair after capture fails', async () => {
     const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(() => {});
     const user = await insertTestUser({
       id: ` test-cost-insight-failure-user-${crypto.randomUUID()} `,
       microdollars_used: 0,
       google_user_email: 'cost-insight-failure@example.com',
     });
+    const organization = await createTestOrganization(
+      'Cost insight capture failure organization',
+      user.id,
+      10_000
+    );
     const { core, metadata } = await defineMicrodollarUsage();
 
     try {
       const result = await insertUsageRecord(
-        { ...core, kilo_user_id: user.id, organization_id: null, cost: 1000 },
+        { ...core, kilo_user_id: user.id, organization_id: organization.id, cost: 1000 },
         metadata
       );
 
-      expect(result).toMatchObject({ usageId: core.id, newMicrodollarsUsed: 1000 });
-      const updatedUser = await findUserById(user.id);
-      expect(updatedUser?.microdollars_used).toBe(1000);
+      expect(result).toMatchObject({ usageId: core.id, newMicrodollarsUsed: null });
 
       const sourceRows = await db
         .select()
@@ -649,9 +660,29 @@ describe('logMicrodollarUsage', () => {
       const totals = await db
         .select()
         .from(cost_insight_owner_hour_totals)
-        .where(eq(cost_insight_owner_hour_totals.owned_by_user_id, user.id));
+        .where(eq(cost_insight_owner_hour_totals.owned_by_organization_id, organization.id));
       expect(totals).toHaveLength(0);
+
+      const degradedIntervals = await db
+        .select()
+        .from(cost_insight_rollup_degraded_intervals)
+        .where(eq(cost_insight_rollup_degraded_intervals.source, 'ai_gateway'));
+      expect(degradedIntervals).toHaveLength(0);
+
+      const repairs = await db
+        .select()
+        .from(cost_insight_rollup_repairs)
+        .where(eq(cost_insight_rollup_repairs.owned_by_organization_id, organization.id));
+      expect(repairs).toHaveLength(1);
+      expect(repairs[0].usage_id).toBe(core.id);
+      expect(new Date(repairs[0].hour_start).toISOString()).toBe(
+        new Date(Math.floor(Date.parse(core.created_at) / 3_600_000) * 3_600_000).toISOString()
+      );
+      expect(Date.parse(repairs[0].next_attempt_at)).toBeGreaterThan(Date.parse(core.created_at));
     } finally {
+      await db
+        .delete(cost_insight_rollup_repairs)
+        .where(eq(cost_insight_rollup_repairs.owned_by_organization_id, organization.id));
       consoleErrorSpy.mockRestore();
     }
   });