fix(video): retain hosted fal receipts before delivery

Author: waleedlatif1

apps/sim/lib/internal/video/client.test.ts

@@ -8,6 +8,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 vi.mock('@/lib/core/execution-limits', () => executionLimitsMock)
 
 import { generateVideo } from '@/lib/internal/video/client'
+import { executeVideoOperation } from '@/lib/internal/video/operations'
+import { observeServiceCosts } from '@/lib/mothership/billing/service-observer'
 
 executionLimitsMockFns.mockGetMaxExecutionTimeout.mockReturnValue(5000)
 
@@ -117,6 +119,60 @@ describe('Video provider client', () => {
     )
   })
 
+  it('preserves the completed hosted receipt for reconciliation when Stop interrupts delivery', async () => {
+    const controller = new AbortController()
+    const requestId = 'completed-video'
+    const endpoint = 'fal-ai/veo3.1'
+    vi.stubGlobal('fetch', async (url: string | URL, init?: RequestInit) => {
+      init?.signal?.throwIfAborted()
+      const path = new URL(url).pathname
+      if (path === `/${endpoint}`) return jsonResponse({ request_id: requestId })
+      if (path.endsWith('/status')) {
+        controller.abort(new Error('Stopped'))
+        return jsonResponse({ status: 'COMPLETED' })
+      }
+      if (path.endsWith('/billing-events')) {
+        return jsonResponse({
+          billing_events: [
+            { request_id: requestId, endpoint_id: endpoint, cost_estimate_nano_usd: 400000000 },
+          ],
+        })
+      }
+      throw new Error('Unexpected provider request')
+    })
+    const pending: string[] = []
+    const result = observeServiceCosts(
+      async () => {
+        throw new Error('Unexpected successful charge')
+      },
+      () =>
+        executeVideoOperation(
+          {
+            provider: 'falai',
+            model: 'veo-3.1',
+            prompt: 'A sunrise',
+            apiKey: 'key',
+            useHostedCostTracking: true,
+          },
+          {
+            requestId: 'trace-id',
+            userId: 'user',
+            headers: new Headers(),
+            signal: controller.signal,
+          }
+        ),
+      async (receipt) => {
+        pending.push(receipt)
+      }
+    )
+    const rejection = expect(result).rejects.toThrow('Stopped')
+    await vi.advanceTimersByTimeAsync(5000)
+    await rejection
+    expect(pending).toHaveLength(1)
+    expect(pending[0]).toContain(requestId)
+    expect(pending[0]).toContain('0.4')
+  })
+
   it('cancels during the provider wait without polling or resubmitting', async () => {
     const controller = new AbortController()
     const mockFetch = vi.fn().mockResolvedValueOnce(jsonResponse({ id: 'task-1' }))

apps/sim/lib/internal/video/client.ts

@@ -51,6 +51,7 @@ export interface VideoGenerationResult {
 }
 
 export interface VideoGenerationContext {
+  onFalCost?: (receipt: FalAICostMetadata) => void
   requestId: string
   signal?: AbortSignal
 }
@@ -820,6 +821,12 @@ async function generateWithFalAI(
 
       logger.info(`[${requestId}] Fal.ai generation completed after ${attempts * 5}s`)
 
+      const falaiCost = useHostedCostTracking
+        ? await getFalAICostMetadata({ apiKey, endpointId: endpoint, requestId: requestIdFal })
+        : undefined
+      if (falaiCost) context.onFalCost?.(falaiCost)
+      signal?.throwIfAborted()
+
       const resultResponse = await fetch(
         falQueueUrl(getStringProperty(statusData, 'response_url') || responseUrl),
         {
@@ -870,14 +877,7 @@ async function generateWithFalAI(
           getNumberProperty(resultData, 'duration') ??
           duration ??
           (legacyMetadata ? 5 : undefined),
-        falaiCost: useHostedCostTracking
-          ? await getFalAICostMetadata({
-              apiKey,
-              endpointId: endpoint,
-              requestId: requestIdFal,
-              signal,
-            })
-          : undefined,
+        falaiCost,
       }
     }
 

apps/sim/lib/internal/video/operations.ts

@@ -1,4 +1,5 @@
 import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
 import { getUserFacingErrorMessage } from '@/lib/core/errors/database-query-error'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { validateOpaqueModelInputProvenance } from '@/lib/execution/model-input-provenance'
@@ -8,6 +9,8 @@ import {
   type VideoGenerationInput,
 } from '@/lib/internal/video/client'
 import { VideoOperationError } from '@/lib/internal/video/errors'
+import { recordServiceMeteringFailure } from '@/lib/mothership/billing/service-observer'
+import type { FalAICostMetadata } from '@/lib/tools/falai-pricing'
 import { StorageService } from '@/lib/uploads'
 import { uploadExecutionFile } from '@/lib/uploads/contexts/execution'
 import {
@@ -142,22 +145,45 @@ export async function executeVideoOperation(
   if (validationError) throw new VideoOperationError(validationError, 400)
   await authorizeVisualReference(input, context)
 
-  const generated = await generateVideo(input, {
-    requestId: context.requestId,
-    signal: context.signal,
-  })
-  context.signal?.throwIfAborted()
-  const stored = await storeVideo(input, generated.buffer, context)
+  let completedFalCost: FalAICostMetadata | undefined
+  try {
+    const generated = await generateVideo(input, {
+      requestId: context.requestId,
+      signal: context.signal,
+      onFalCost: (cost) => {
+        completedFalCost = cost
+      },
+    })
+    context.signal?.throwIfAborted()
+    const stored = await storeVideo(input, generated.buffer, context)
 
-  return {
-    ...stored,
-    duration: generated.duration || input.duration,
-    width: generated.width,
-    height: generated.height,
-    provider: input.provider,
-    model: input.model || 'default',
-    jobId: generated.jobId,
-    __falaiCostDollars: generated.falaiCost?.costDollars,
-    __falaiBilling: generated.falaiCost,
+    return {
+      ...stored,
+      duration: generated.duration || input.duration,
+      width: generated.width,
+      height: generated.height,
+      provider: input.provider,
+      model: input.model || 'default',
+      jobId: generated.jobId,
+      __falaiCostDollars: generated.falaiCost?.costDollars,
+      __falaiBilling: generated.falaiCost,
+    }
+  } catch (error) {
+    if (completedFalCost) {
+      logger.warn('Hosted fal video completed but delivery failed', {
+        requestId: context.requestId,
+        falaiBilling: completedFalCost,
+      })
+      const { requestId, endpointId, costDollars } = completedFalCost
+      await recordServiceMeteringFailure(
+        `Completed fal video ${requestId} (${endpointId}): USD ${costDollars}; delivery failed`
+      ).catch((meteringError) => {
+        logger.error('Failed to record completed fal video for reconciliation', {
+          requestId: context.requestId,
+          error: getErrorMessage(meteringError),
+        })
+      })
+    }
+    throw error
   }
 }