fix: fal drift probe skips on 403/locked/infra instead of poisoning baseline

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);
+  });
+});

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);

src/__tests__/drift/providers.ts

@@ -184,6 +184,10 @@ function toSSEEventShapes(events: { type: string; data: unknown }[]): SSEEventSh
 
 function withInfraErrorTag<T>(provider: string, fn: () => Promise<T>): Promise<T> {
   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<string[
 // fal.ai queue lifecycle
 // ---------------------------------------------------------------------------
 
+/**
+ * Raised by the live fal canary when fal ITSELF is unavailable for an
+ * infra/auth/balance/rate reason (401 | 402 | 403 incl. "user locked" /
+ * "exhausted balance" | 429 | 5xx). This is NOT drift — the live leg catches it
+ * and SKIPS (exactly like the no-key skip), so a locked fal account never
+ * poisons the drift baseline.
+ *
+ * Deliberately NOT an `InfraError`: `withInfraErrorTag` re-throws it untouched,
+ * so the drift collector never sees an unparseable INFRA_ERROR to quarantine
+ * (exit 5). A genuine SHAPE drift (a 2xx whose envelope diverges from the mock)
+ * is never an infra status, so it still fails and reports drift as before.
+ */
+export class FalCanarySkip extends Error {
+  readonly status: number;
+
+  constructor(reason: string, status: number) {
+    super(reason);
+    this.name = "FalCanarySkip";
+    this.status = status;
+  }
+}
+
+/**
+ * True when a fal HTTP status means fal is UNAVAILABLE (auth/balance/rate/5xx),
+ * not that the response envelope drifted. Only these convert to a skip — every
+ * other status (incl. the expected 400 `ALREADY_COMPLETED` cancel race, or a
+ * 2xx with a wrong shape) flows through normally so real drift still fails.
+ */
+function isFalInfraStatus(status: number): boolean {
+  return status === 401 || status === 402 || status === 403 || status === 429 || status >= 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(