diff --git a/src/__tests__/drift/fal-canary-skip.test.ts b/src/__tests__/drift/fal-canary-skip.test.ts new file mode 100644 index 00000000..70f62d94 --- /dev/null +++ b/src/__tests__/drift/fal-canary-skip.test.ts @@ -0,0 +1,120 @@ +/** + * Regression + guard tests for the live fal queue canary's infra resilience. + * + * BUG (main baseline regression, #327): with FAL_KEY set but the fal account + * balance EXHAUSTED, fal returns `403 {"detail":"User is locked. Reason: + * Exhausted balance..."}` at submit. The canary threw a hard `InfraError`, which + * the drift collector could not parse into a finding → exit-5 quarantine → the + * "Base drift report" step (run against main) FAILED on every PR and the + * scheduled run. + * + * FIX: infra-class statuses (401/402/403/429/5xx) become an honest + * `FalCanarySkip` the live leg catches → `ctx.skip()`, NEVER a hard InfraError. + * + * RED (pre-fix): canary rejects with `InfraError` (…403…Exhausted balance). + * GREEN (post-fix): canary rejects with `FalCanarySkip` carrying a clear reason. + * + * GUARD (must stay RED on real drift): a 2xx submit whose envelope shape + * diverges from the mock is NOT an infra status, so it is NOT swallowed by the + * skip — the canary returns normally and the shape comparison still reports + * critical drift. + */ +import { describe, it, expect, afterEach } from "vitest"; +import { falQueueLifecycleCanary, FalCanarySkip, InfraError } from "./providers.js"; +import { extractShape, triangulate } from "./schema.js"; + +const origFetch = globalThis.fetch; +afterEach(() => { + globalThis.fetch = origFetch; +}); + +/** A valid fal queue submit envelope (the contract the mock must match). */ +const GOOD_SUBMIT_ENVELOPE = { + request_id: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + status_url: "https://queue.fal.run/fal-ai/flux/schnell/requests/aaaaaaaa/status", + response_url: "https://queue.fal.run/fal-ai/flux/schnell/requests/aaaaaaaa", + cancel_url: "https://queue.fal.run/fal-ai/flux/schnell/requests/aaaaaaaa/cancel", + queue_position: 0, +}; + +describe("fal canary infra resilience", () => { + it("REGRESSION: 403 user-locked / exhausted-balance at submit → FalCanarySkip, not InfraError", async () => { + globalThis.fetch = (async () => + new Response( + '{"detail":"User is locked. Reason: Exhausted balance. Add funds to continue."}', + { + status: 403, + }, + )) as typeof fetch; + + const err = await falQueueLifecycleCanary("locked-key", "fal-ai/flux/schnell", { + prompt: "canary", + }).then( + () => { + throw new Error("expected canary to reject with a skip"); + }, + (e: unknown) => e, + ); + + // Honest skip — NOT a hard InfraError that the collector would quarantine. + expect(err).toBeInstanceOf(FalCanarySkip); + expect(err).not.toBeInstanceOf(InfraError); + const skip = err as FalCanarySkip; + expect(skip.status).toBe(403); + expect(skip.message).toMatch(/user-locked/); + expect(skip.message).toMatch(/skipping live canary/); + }); + + it("skips on other infra statuses (401 stale-key, 402 payment-required) too", async () => { + // Non-retryable statuses so the assertion is instant (429/5xx go through the + // same isFalInfraStatus gate but incur real retry backoff). + for (const status of [401, 402]) { + globalThis.fetch = (async () => + new Response("upstream unavailable", { status })) as typeof fetch; + const err = await falQueueLifecycleCanary("k", "fal-ai/flux/schnell", { prompt: "c" }).then( + () => { + throw new Error("expected skip"); + }, + (e: unknown) => e, + ); + expect(err, `status ${status}`).toBeInstanceOf(FalCanarySkip); + expect((err as FalCanarySkip).status, `status ${status}`).toBe(status); + } + }); + + it("GUARD: a 2xx submit whose envelope diverges from the mock is NOT skipped — real drift still surfaces", async () => { + // fal returns 200 but the REAL envelope's queue_position is a string where + // the mock has a number — a genuine, critical shape drift. This mirrors the + // live leg's own triangulate(exemplar, real, mock) drift check exactly. + const driftedRealSubmit = { + request_id: "id-1", + status_url: "https://queue.fal.run/fal-ai/flux/schnell/requests/id-1/status", + response_url: "https://queue.fal.run/fal-ai/flux/schnell/requests/id-1", + cancel_url: "https://queue.fal.run/fal-ai/flux/schnell/requests/id-1/cancel", + queue_position: "0", // TYPE DRIFT: string vs mock's number + }; + globalThis.fetch = (async (_url: string | URL, init?: RequestInit) => { + const method = (init?.method ?? "GET").toUpperCase(); + if (method === "POST") + return new Response(JSON.stringify(driftedRealSubmit), { status: 200 }); + if (method === "PUT") + return new Response(JSON.stringify({ status: "CANCELLATION_REQUESTED" }), { status: 200 }); + return new Response(JSON.stringify({ status: "IN_QUEUE", request_id: "id-1" }), { + status: 200, + }); + }) as typeof fetch; + + // Must NOT throw a skip: 200 is not an infra status. + const result = await falQueueLifecycleCanary("k", "fal-ai/flux/schnell", { prompt: "c" }); + expect(result.submit.status).toBe(200); + + // The exact triangulation the live leg runs still flags the drift as critical. + const diffs = triangulate( + extractShape(GOOD_SUBMIT_ENVELOPE), // exemplar + extractShape(result.submit.body), // real (drifted) + extractShape(GOOD_SUBMIT_ENVELOPE), // mock + ); + const critical = diffs.filter((d) => d.severity === "critical"); + expect(critical.length).toBeGreaterThan(0); + }); +}); diff --git a/src/__tests__/drift/fal-queue.drift.ts b/src/__tests__/drift/fal-queue.drift.ts index b12ebc53..a62cba7a 100644 --- a/src/__tests__/drift/fal-queue.drift.ts +++ b/src/__tests__/drift/fal-queue.drift.ts @@ -13,7 +13,7 @@ import { describe, it, expect, beforeAll, afterAll } from "vitest"; import { LLMock } from "../../llmock.js"; import { extractShape, compareShapes, triangulate, formatDriftReport } from "./schema.js"; -import { falQueueLifecycleCanary } from "./providers.js"; +import { falQueueLifecycleCanary, FalCanarySkip, type FalQueueCanaryResult } from "./providers.js"; const FAL_KEY = process.env.FAL_KEY; @@ -251,23 +251,38 @@ describe("fal.ai queue lifecycle shapes", () => { const FAL_CANARY_MODEL = "fal-ai/flux/schnell"; describe.skipIf(!FAL_KEY)("fal.ai queue lifecycle (live, cost-safe)", () => { - it("real submit + status + cancel envelopes match aimock's queue contract", async () => { + it("real submit + status + cancel envelopes match aimock's queue contract", async (ctx) => { // Drive the real fal queue (submit + immediate cancel) and the aimock // server in parallel, then triangulate exemplar x real x mock per step. - const [live, mockSubmitRes] = await Promise.all([ - falQueueLifecycleCanary(FAL_KEY!, FAL_CANARY_MODEL, { - prompt: "aimock drift canary — cancelled immediately", - num_images: 1, - }), - fetch(`${mock.url}/fal/${FAL_CANARY_MODEL}`, { - method: "POST", - headers: { - "Content-Type": "application/json", - "x-fal-target-host": "queue.fal.run", - }, - body: JSON.stringify({ input: { prompt: "a cat" } }), - }), - ]); + let live: FalQueueCanaryResult; + let mockSubmitRes: Response; + try { + [live, mockSubmitRes] = await Promise.all([ + falQueueLifecycleCanary(FAL_KEY!, FAL_CANARY_MODEL, { + prompt: "aimock drift canary — cancelled immediately", + num_images: 1, + }), + fetch(`${mock.url}/fal/${FAL_CANARY_MODEL}`, { + method: "POST", + headers: { + "Content-Type": "application/json", + "x-fal-target-host": "queue.fal.run", + }, + body: JSON.stringify({ input: { prompt: "a cat" } }), + }), + ]); + } catch (err) { + // fal itself unavailable (locked account / exhausted balance / rate / + // 5xx) is NOT drift — skip the live canary with a clear reason instead of + // failing the drift suite and poisoning the baseline. A genuine envelope + // drift is a 2xx-with-wrong-shape and never lands here. + if (err instanceof FalCanarySkip) { + console.warn(`[fal drift] ${err.message}`); + ctx.skip(err.message); + return; + } + throw err; + } // --- Submit: the queue contract's load-bearing fields, then triangulate --- expect(live.submit.status, JSON.stringify(live.submit.body)).toBe(200); diff --git a/src/__tests__/drift/providers.ts b/src/__tests__/drift/providers.ts index 1c2251c5..a4dc4817 100644 --- a/src/__tests__/drift/providers.ts +++ b/src/__tests__/drift/providers.ts @@ -184,6 +184,10 @@ function toSSEEventShapes(events: { type: string; data: unknown }[]): SSEEventSh function withInfraErrorTag(provider: string, fn: () => Promise): Promise { return fn().catch((err: unknown) => { + // A canary skip is an HONEST "provider unavailable" signal, not a drift + // finding — never re-tag it as an InfraError (that would poison the drift + // baseline via exit-5 quarantine). Let it propagate for the leg to catch. + if (err instanceof FalCanarySkip) throw err; const msg = err instanceof Error ? err.message : String(err); const status = err instanceof InfraError ? err.status : 0; throw new InfraError(`INFRA_ERROR: ${provider} — ${msg}`, status); @@ -732,6 +736,53 @@ export async function listOpenRouterVideoModels(apiKey: string): Promise= 500; +} + +/** + * Throw a {@link FalCanarySkip} if `status` is an infra-class status. Called at + * each live lifecycle step (submit, status) BEFORE the hard `parseJsonResponse` + * → `InfraError` path, so an unavailable fal account skips instead of poisoning + * the baseline. + */ +function skipIfFalUnavailable(step: string, status: number, raw: string): void { + if (!isFalInfraStatus(status)) return; + const locked = /user is locked|exhausted balance/i.test(raw) ? " user-locked" : ""; + throw new FalCanarySkip( + `fal infra/auth unavailable (${status}${locked}) at ${step} — skipping live canary`, + status, + ); +} + /** One lifecycle step's HTTP status + parsed JSON envelope. */ interface FalQueueStep { status: number; @@ -775,6 +826,10 @@ export async function falQueueLifecycleCanary( body: JSON.stringify(input), }); const submitRaw = await submitRes.text(); + // fal unavailable (locked/exhausted-balance/rate/5xx) → honest skip, NOT a + // hard InfraError. Fires before the job is enqueued, so there is no cost and + // nothing to cancel. + skipIfFalUnavailable("submit", submitRes.status, submitRaw); const submitBody = parseJsonResponse( submitRaw, submitRes.status, @@ -800,6 +855,9 @@ export async function falQueueLifecycleCanary( try { const statusRes = await fetchWithRetry(statusUrl, { method: "GET", headers: authHeaders }); const statusRaw = await statusRes.text(); + // fal unavailable mid-lifecycle → skip (rethrown after cancel fires, so a + // job is never left to run). A FalCanarySkip is preserved end-to-end. + skipIfFalUnavailable("status", statusRes.status, statusRaw); statusPoll = { status: statusRes.status, body: parseJsonResponse(