diff --git a/apps/sim/lib/media/falai-audio.ts b/apps/sim/lib/media/falai-audio.ts index 5b611a73204..e9705d9e89e 100644 --- a/apps/sim/lib/media/falai-audio.ts +++ b/apps/sim/lib/media/falai-audio.ts @@ -141,7 +141,7 @@ function buildInput( export async function generateFalAudio(params: GenerateFalAudioParams): Promise { const type: AudioType = params.type || 'speech' - const apiKey = getFalApiKey() + const { apiKey, envVarName } = getFalApiKey() // Voice cloning: a reference sample routes to a zero-shot clone model (F5-TTS), // which conditions on the sample (ref_audio_url) and speaks the prompt in that voice. @@ -152,7 +152,13 @@ export async function generateFalAudio(params: GenerateFalAudioParams): Promise< ref_audio_url: params.voiceSampleDataUri, model_type: 'F5-TTS', } - const { requestId, data } = await runFalQueue(model, input, apiKey, params.signal) + const { requestId, data } = await runFalQueue( + model, + input, + apiKey, + { tool: 'generate_audio', envVarName }, + params.signal + ) const cost = await recordFalCost(apiKey, model, requestId, params.onCost) const url = extractFalMediaUrl(data, ['audio', 'audio_url', 'audio_file', 'output']) if (!url) throw new Error('No audio URL in Fal.ai clone response') @@ -172,7 +178,13 @@ export async function generateFalAudio(params: GenerateFalAudioParams): Promise< const input = buildInput(type, params, model) // For fal audio models the model ID is the queue endpoint. - const { requestId, data } = await runFalQueue(model, input, apiKey, params.signal) + const { requestId, data } = await runFalQueue( + model, + input, + apiKey, + { tool: 'generate_audio', envVarName }, + params.signal + ) const cost = await recordFalCost(apiKey, model, requestId, params.onCost) const url = extractFalMediaUrl(data, ['audio', 'audio_url', 'audio_file', 'output']) if (!url) throw new Error('No audio URL in Fal.ai response') diff --git a/apps/sim/lib/media/falai-hosted-key-metrics.test.ts b/apps/sim/lib/media/falai-hosted-key-metrics.test.ts new file mode 100644 index 00000000000..11b51a2e55f --- /dev/null +++ b/apps/sim/lib/media/falai-hosted-key-metrics.test.ts @@ -0,0 +1,90 @@ +import { jsonResponse } from '@sim/testing/helpers/http' +import { resetEnvFlagsMock, setEnvFlags } from '@sim/testing/mocks/env-flags.mock' +import { inputValidationMock } from '@sim/testing/mocks/input-validation.mock' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { sentMetricData } = vi.hoisted(() => ({ + sentMetricData: [] as Array<{ + MetricName?: string + Dimensions?: Array<{ Name?: string; Value?: string }> + }>, +})) + +vi.mock('@/lib/core/security/input-validation.server', () => inputValidationMock) +vi.mock('@aws-sdk/client-cloudwatch', () => ({ + StandardUnit: { Count: 'Count', None: 'None', Milliseconds: 'Milliseconds' }, + PutMetricDataCommand: class { + constructor(readonly input: { MetricData: typeof sentMetricData }) {} + }, + CloudWatchClient: class { + async send(command: { input: { MetricData: typeof sentMetricData } }) { + sentMetricData.push(...command.input.MetricData) + } + }, +})) + +import { runFalQueue } from '@/lib/media/falai' +import { flushHostedKeyMetrics } from '@/lib/monitoring/metrics' + +function stubFalSubmit(status: number, body: unknown) { + vi.stubGlobal( + 'fetch', + vi.fn(async () => jsonResponse(body, status)) + ) +} + +async function failedHostedKeyDimensions(): Promise>> { + await flushHostedKeyMetrics() + return sentMetricData + .filter((datum) => datum.MetricName === 'Failed') + .map((datum) => Object.fromEntries((datum.Dimensions ?? []).map((d) => [d.Name, d.Value]))) +} + +describe('a fal submit the hosted account refuses', () => { + beforeEach(() => { + vi.useFakeTimers() + setEnvFlags({ isHosted: true }) + sentMetricData.length = 0 + }) + + afterEach(() => { + vi.useRealTimers() + resetEnvFlagsMock() + }) + + it.each([ + [403, { detail: 'User is locked. Reason: Exhausted balance.' }], + [401, { detail: 'Invalid API key' }], + ])('records a hosted-key auth failure against the refused key for a %i', async (status, body) => { + stubFalSubmit(status, body) + + await expect( + runFalQueue('fal-ai/model', {}, 'key', { + tool: 'generate_audio', + envVarName: 'FALAI_API_KEY_2', + }) + ).rejects.toThrow(`Fal.ai API error: ${status}`) + + expect(await failedHostedKeyDimensions()).toEqual([ + expect.objectContaining({ + Provider: 'falai', + Tool: 'generate_audio', + Key: 'FALAI_API_KEY_2', + Reason: 'auth', + }), + ]) + }) + + it('records nothing when fal rejects the request input', async () => { + stubFalSubmit(422, { detail: [{ msg: 'field required' }] }) + + await expect( + runFalQueue('fal-ai/model', {}, 'key', { + tool: 'generate_video', + envVarName: 'FALAI_API_KEY_1', + }) + ).rejects.toThrow('Fal.ai API error: 422') + + expect(await failedHostedKeyDimensions()).toEqual([]) + }) +}) diff --git a/apps/sim/lib/media/falai-video.ts b/apps/sim/lib/media/falai-video.ts index cdea37ea4cc..ddeb7e56b6d 100644 --- a/apps/sim/lib/media/falai-video.ts +++ b/apps/sim/lib/media/falai-video.ts @@ -148,7 +148,7 @@ export async function generateFalVideo(params: GenerateFalVideoParams): Promise< ) } - const apiKey = getFalApiKey() + const { apiKey, envVarName } = getFalApiKey() let endpoint = config.endpoint const input: Record = { prompt: params.prompt } @@ -177,7 +177,13 @@ export async function generateFalVideo(params: GenerateFalVideoParams): Promise< input.prompt_optimizer = params.promptOptimizer } - const { requestId, data } = await runFalQueue(endpoint, input, apiKey, params.signal) + const { requestId, data } = await runFalQueue( + endpoint, + input, + apiKey, + { tool: 'generate_video', envVarName }, + params.signal + ) const cost = await recordFalCost(apiKey, endpoint, requestId, params.onCost) const url = extractFalMediaUrl(data, ['video', 'output']) if (!url) throw new Error('No video URL in Fal.ai response') diff --git a/apps/sim/lib/media/falai.test.ts b/apps/sim/lib/media/falai.test.ts index a9dd83f2da7..410ec48b575 100644 --- a/apps/sim/lib/media/falai.test.ts +++ b/apps/sim/lib/media/falai.test.ts @@ -13,6 +13,7 @@ import { downloadFalMedia, type FalCostRecorder, runFalQueue } from '@/lib/media import { generateFalAudio } from '@/lib/media/falai-audio' import { generateFalVideo } from '@/lib/media/falai-video' +const FAL_KEY_USE = { tool: 'generate_video', envVarName: 'FALAI_API_KEY' } as const const CANCEL_URL = 'https://queue.fal.run/fal-ai/model/requests/req-1/cancel' const STATUS_URL = 'https://queue.fal.run/fal-ai/model/requests/req-1/status' @@ -67,7 +68,13 @@ describe('runFalQueue on a request that never finishes', () => { it('stops polling and cancels the request once its caller gives up', async () => { const fal = stubStuckFalQueue() const caller = new AbortController() - const generation = runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', caller.signal) + const generation = runFalQueue( + 'fal-ai/model', + { prompt: 'p' }, + 'key', + FAL_KEY_USE, + caller.signal + ) const outcome = generation.then( () => 'resolved', () => 'rejected' @@ -87,7 +94,13 @@ describe('runFalQueue on a request that never finishes', () => { it('ends a poll that never answers when its caller gives up', async () => { const fal = stubStuckFalQueue({ statusHangs: true }) const caller = new AbortController() - const generation = runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', caller.signal) + const generation = runFalQueue( + 'fal-ai/model', + { prompt: 'p' }, + 'key', + FAL_KEY_USE, + caller.signal + ) const outcome = generation.then( () => 'resolved', () => 'rejected' @@ -104,7 +117,7 @@ describe('runFalQueue on a request that never finishes', () => { it('fails and cancels after its own ceiling even when nothing aborts it', async () => { const fal = stubStuckFalQueue() - const generation = runFalQueue('fal-ai/model', { prompt: 'p' }, 'key') + const generation = runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', FAL_KEY_USE) const outcome = generation.then( () => 'resolved', (error: unknown) => String(error) @@ -119,9 +132,13 @@ describe('runFalQueue on a request that never finishes', () => { it('warns when fal refuses to cancel a request', async () => { const fal = stubStuckFalQueue({ cancelStatus: 500 }) const caller = new AbortController() - const generation = runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', caller.signal).catch( - () => undefined - ) + const generation = runFalQueue( + 'fal-ai/model', + { prompt: 'p' }, + 'key', + FAL_KEY_USE, + caller.signal + ).catch(() => undefined) await vi.advanceTimersByTimeAsync(3_500) caller.abort() await generation @@ -137,7 +154,7 @@ describe('runFalQueue on a request that never finishes', () => { const fal = stubStuckFalQueue() await expect( - runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', AbortSignal.abort()) + runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', FAL_KEY_USE, AbortSignal.abort()) ).rejects.toThrow() expect(fal.calls).toEqual([]) }) @@ -160,9 +177,13 @@ describe('runFalQueue on a request that never finishes', () => { }) ) const caller = new AbortController() - const generation = runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', caller.signal).catch( - () => 'rejected' - ) + const generation = runFalQueue( + 'fal-ai/model', + { prompt: 'p' }, + 'key', + FAL_KEY_USE, + caller.signal + ).catch(() => 'rejected') await vi.advanceTimersByTimeAsync(0) caller.abort(new Error('watchdog')) @@ -175,9 +196,13 @@ describe('runFalQueue on a request that never finishes', () => { it('asks fal never to start the request past that ceiling', async () => { const fal = stubStuckFalQueue() const caller = new AbortController() - const generation = runFalQueue('fal-ai/model', { prompt: 'p' }, 'key', caller.signal).catch( - () => undefined - ) + const generation = runFalQueue( + 'fal-ai/model', + { prompt: 'p' }, + 'key', + FAL_KEY_USE, + caller.signal + ).catch(() => undefined) await vi.advanceTimersByTimeAsync(0) caller.abort() await generation diff --git a/apps/sim/lib/media/falai.ts b/apps/sim/lib/media/falai.ts index 40db1bd5f6e..e92dadac42b 100644 --- a/apps/sim/lib/media/falai.ts +++ b/apps/sim/lib/media/falai.ts @@ -13,6 +13,7 @@ import { readResponseTextWithLimit, readResponseToBufferWithLimit, } from '@/lib/core/utils/stream-limits' +import { hostedKeyMetrics } from '@/lib/monitoring/metrics' import { type FalAICostMetadata, getFalAICostMetadata } from '@/lib/tools/falai-pricing' const logger = createLogger('FalMediaClient') @@ -33,20 +34,33 @@ const FAL_CANCEL_TIMEOUT_MS = 5000 */ const FAL_SUBMIT_TIMEOUT_MS = 60_000 +/** Who spent a hosted Fal.ai key: the Mothership tool and the env var the key came from. */ +interface FalKeyUse { + tool: 'generate_audio' | 'generate_video' + envVarName: string +} + +/** + * Statuses fal answers when the hosted account itself refuses the call (a revoked key, or an + * account locked for an exhausted balance), so every generation fails until someone acts. + */ +const FAL_ACCOUNT_REFUSAL_STATUSES = new Set([401, 402, 403]) + /** * Resolves a hosted Fal.ai API key from the numbered env pool * (FALAI_API_KEY_COUNT + FALAI_API_KEY_1..N), round-robined by minute, * mirroring getRotatingApiKey. Falls back to a single FALAI_API_KEY for dev. */ -export function getFalApiKey(): string { +export function getFalApiKey(): { apiKey: string; envVarName: string } { const count = Number.parseInt(process.env.FALAI_API_KEY_COUNT || '0', 10) - const keys: string[] = [] + const keys: Array<{ apiKey: string; envVarName: string }> = [] for (let i = 1; i <= count; i++) { - const key = process.env[`FALAI_API_KEY_${i}`] - if (key) keys.push(key) + const envVarName = `FALAI_API_KEY_${i}` + const apiKey = process.env[envVarName] + if (apiKey) keys.push({ apiKey, envVarName }) } if (keys.length === 0 && process.env.FALAI_API_KEY) { - keys.push(process.env.FALAI_API_KEY) + keys.push({ apiKey: process.env.FALAI_API_KEY, envVarName: 'FALAI_API_KEY' }) } if (keys.length === 0) { throw new Error( @@ -101,6 +115,7 @@ export async function runFalQueue( endpoint: string, input: Record, apiKey: string, + keyUse: FalKeyUse, signal?: AbortSignal ): Promise { const ceiling = new AbortController() @@ -110,7 +125,7 @@ export async function runFalQueue( ) const queueSignal = signal ? AbortSignal.any([signal, ceiling.signal]) : ceiling.signal try { - return await submitAndPollFalQueue(endpoint, input, apiKey, queueSignal) + return await submitAndPollFalQueue(endpoint, input, apiKey, keyUse, queueSignal) } finally { clearTimeout(ceilingTimer) } @@ -120,6 +135,7 @@ async function submitAndPollFalQueue( endpoint: string, input: Record, apiKey: string, + keyUse: FalKeyUse, queueSignal: AbortSignal ): Promise { // Only a submit already under way outlives its caller; a stopped call never starts one. @@ -135,6 +151,14 @@ async function submitAndPollFalQueue( signal: AbortSignal.timeout(FAL_SUBMIT_TIMEOUT_MS), }) if (!createResponse.ok) { + if (FAL_ACCOUNT_REFUSAL_STATUSES.has(createResponse.status)) { + hostedKeyMetrics.recordFailed({ + provider: 'falai', + tool: keyUse.tool, + key: keyUse.envVarName, + reason: 'auth', + }) + } const err = await readResponseTextWithLimit(createResponse, { maxBytes: DEFAULT_MAX_ERROR_BODY_BYTES, label: 'Fal.ai create error response', diff --git a/apps/sim/lib/monitoring/metrics.ts b/apps/sim/lib/monitoring/metrics.ts index 9639078773f..9562fe609db 100644 --- a/apps/sim/lib/monitoring/metrics.ts +++ b/apps/sim/lib/monitoring/metrics.ts @@ -46,11 +46,6 @@ type QueueReason = 'actor_requests' | 'dimension' | 'queue_position' /** `metering` marks a provider call that succeeded but could not be priced. */ type FailureReason = 'rate_limited' | 'auth' | 'other' | 'metering' -// Hosted keys exist only on Sim Cloud, whose app and workers can always reach -// CloudWatch (task role or workload identity). Elsewhere recorders are no-ops, -// so they stay always-safe to call. -const ENABLED = isHosted - const ENVIRONMENT = process.env.OTEL_DEPLOYMENT_ENVIRONMENT || process.env.DEPLOYMENT_ENVIRONMENT || @@ -98,13 +93,16 @@ function buildDimensions(labels: Record) { return dimensions } +// Hosted keys exist only on Sim Cloud, whose app and workers can always reach +// CloudWatch (task role or workload identity). Elsewhere recorders are no-ops, +// so they stay always-safe to call. function enqueue( MetricName: string, Value: number, Unit: StandardUnit, labels: Record ): void { - if (!ENABLED) return + if (!isHosted) return buffer.push({ MetricName, Value, @@ -129,7 +127,7 @@ export async function flushHostedKeyMetrics(): Promise { logger.warn('Dropped hosted-key metric points (buffer cap reached)', { dropped }) dropped = 0 } - if (!ENABLED || buffer.length === 0) return + if (!isHosted || buffer.length === 0) return const pending = buffer buffer = [] for (let i = 0; i < pending.length; i += MAX_BATCH) { diff --git a/scripts/check-unused-exports.baseline.json b/scripts/check-unused-exports.baseline.json index e4448b682b2..e9849ac92c2 100644 --- a/scripts/check-unused-exports.baseline.json +++ b/scripts/check-unused-exports.baseline.json @@ -4702,7 +4702,6 @@ "apps/sim/lib/messaging/email/mailer.ts#sendBatchEmails", "apps/sim/lib/messaging/email/types.ts#EmailAttachment", "apps/sim/lib/messaging/email/utils.ts#extractEmailFromAddress", - "apps/sim/lib/monitoring/metrics.ts#flushHostedKeyMetrics", "apps/sim/lib/mothership/agent-cli/types.ts#AgentCliClient", "apps/sim/lib/mothership/application/execute-knowledge-use-case.ts#requireCopilotKnowledgeWorkspaceId", "apps/sim/lib/mothership/application/execute-knowledge-use-case.ts#resolveCopilotKnowledgePrincipal",