feat: fal.ai live queue-lifecycle drift probe (cost-safe, FAL_KEY-gated)

.github/workflows/fix-drift.yml

@@ -102,6 +102,7 @@ jobs:
           # Un-gates the Ollama live drift leg against the daemon provisioned above.
           OLLAMA_HOST: 127.0.0.1:11434
           OLLAMA_MODEL: qwen2:0.5b
+          FAL_KEY: ${{ secrets.FAL_KEY }}
           PRE_FIX_REPORT: ${{ runner.temp }}/drift-report.json
         run: |
           set +e
@@ -202,6 +203,7 @@ jobs:
           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
           GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
           OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }}
+          FAL_KEY: ${{ secrets.FAL_KEY }}
           PINNED_REPORT: ${{ runner.temp }}/drift-report.pinned.json
         run: npx tsx scripts/fix-drift.ts --report "${PINNED_REPORT}"
 
@@ -238,6 +240,7 @@ jobs:
           # (the daemon provisioned earlier in this job is still running).
           OLLAMA_HOST: 127.0.0.1:11434
           OLLAMA_MODEL: qwen2:0.5b
+          FAL_KEY: ${{ secrets.FAL_KEY }}
         run: pnpm test:drift
 
       # Step 4c: Re-collect drift AUTHORITATIVELY against the LIVE SDK, writing
@@ -266,6 +269,7 @@ jobs:
           # exercises it (the daemon provisioned earlier in this job is running).
           OLLAMA_HOST: 127.0.0.1:11434
           OLLAMA_MODEL: qwen2:0.5b
+          FAL_KEY: ${{ secrets.FAL_KEY }}
           POST_FIX_REPORT: ${{ runner.temp }}/drift-report.post-fix.json
         run: |
           set +e

.github/workflows/test-drift.yml

@@ -81,6 +81,7 @@ jobs:
           ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
           GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
           OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }}
+          FAL_KEY: ${{ secrets.FAL_KEY }}
         run: |
           # Written at the checkout root (cwd) so the relative provider import
           # below resolves against the repo. ESM relative specifiers key off the
@@ -188,6 +189,7 @@ jobs:
           # Un-gates the Ollama live drift leg against the daemon provisioned above.
           OLLAMA_HOST: 127.0.0.1:11434
           OLLAMA_MODEL: qwen2:0.5b
+          FAL_KEY: ${{ secrets.FAL_KEY }}
         run: |
           set +e
           npx tsx scripts/drift-retry.ts
@@ -369,6 +371,7 @@ jobs:
       ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
       GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
       OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }}
+      FAL_KEY: ${{ secrets.FAL_KEY }}
     steps:
       # HEAD checkout: the PR ref (default checkout behavior on pull_request).
       # persist-credentials:false — this leg never pushes; keeps the token off

src/__tests__/drift/fal-queue.drift.ts

@@ -12,7 +12,10 @@
 
 import { describe, it, expect, beforeAll, afterAll } from "vitest";
 import { LLMock } from "../../llmock.js";
-import { extractShape, compareShapes, formatDriftReport } from "./schema.js";
+import { extractShape, compareShapes, triangulate, formatDriftReport } from "./schema.js";
+import { falQueueLifecycleCanary } from "./providers.js";
+
+const FAL_KEY = process.env.FAL_KEY;
 
 // ---------------------------------------------------------------------------
 // Expected shapes (fal.ai queue contract)
@@ -227,3 +230,95 @@ describe("fal.ai queue lifecycle shapes", () => {
     ).toEqual([]);
   });
 });
+
+// ---------------------------------------------------------------------------
+// LIVE queue-lifecycle canary (COST-SAFE, gated on FAL_KEY).
+//
+// Skips in CI until FAL_KEY is mirrored to repo secrets. Drives the REAL fal
+// queue API through submit -> status -> IMMEDIATE cancel and triangulates each
+// envelope (exemplar x real x aimock mock). It NEVER fetches the completed
+// result payload — that is the only paid retrieval, so the completed-result
+// envelope stays STATIC-only (the mock-vs-exemplar tests above).
+//
+// COST: fal bills compute only when a queued job RUNS. Submitting is free and
+// the job is cancelled while still IN_QUEUE, so the expected cost is $0.
+// Cheapest reliably-available model is used to bound the worst case; residual
+// exposure is at most one sub-cent generation if the model races to completion
+// before the cancel lands.
+// ---------------------------------------------------------------------------
+
+/** Cheapest reliably-available fal image model; cancelled before it runs. */
+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 () => {
+    // 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" } }),
+      }),
+    ]);
+
+    // --- Submit: the queue contract's load-bearing fields, then triangulate ---
+    expect(live.submit.status, JSON.stringify(live.submit.body)).toBe(200);
+    expect(typeof live.submit.body?.request_id).toBe("string");
+    expect(typeof live.submit.body?.status_url).toBe("string");
+    expect(typeof live.submit.body?.response_url).toBe("string");
+    expect(typeof live.submit.body?.cancel_url).toBe("string");
+
+    const mockSubmitBody = await mockSubmitRes.json();
+    const submitDiffs = triangulate(
+      falQueueSubmitShape(),
+      extractShape(live.submit.body),
+      extractShape(mockSubmitBody),
+    );
+    expect(
+      submitDiffs.filter((d) => d.severity === "critical"),
+      formatDriftReport("fal.ai queue submit (live)", submitDiffs, "fal-queue"),
+    ).toEqual([]);
+
+    // --- Status: triangulate real vs aimock vs exemplar (shape, not value) ---
+    const mockStatusRes = await fetch(
+      `${mock.url}/fal/${FAL_CANARY_MODEL}/requests/${mockSubmitBody.request_id}/status`,
+      { headers: { "x-fal-target-host": "queue.fal.run" } },
+    );
+    expect(live.statusPoll.status, JSON.stringify(live.statusPoll.body)).toBe(200);
+    expect(typeof live.statusPoll.body?.status).toBe("string");
+
+    const statusDiffs = triangulate(
+      falQueueStatusShape(),
+      extractShape(live.statusPoll.body),
+      extractShape(await mockStatusRes.json()),
+    );
+    expect(
+      statusDiffs.filter((d) => d.severity === "critical"),
+      formatDriftReport("fal.ai queue status (live)", statusDiffs, "fal-queue"),
+    ).toEqual([]);
+
+    // --- Cancel: real fal returns a { status } envelope (200
+    // CANCELLATION_REQUESTED while queued, or 400 ALREADY_COMPLETED on a race).
+    // The load-bearing field is `status: string` either way. ---
+    expect([200, 400]).toContain(live.cancel.status);
+    expect(typeof live.cancel.body?.status).toBe("string");
+
+    const cancelDiffs = triangulate(
+      falQueueCancelShape(),
+      extractShape(live.cancel.body),
+      falQueueCancelShape(),
+    );
+    expect(
+      cancelDiffs.filter((d) => d.severity === "critical"),
+      formatDriftReport("fal.ai queue cancel (live)", cancelDiffs, "fal-queue"),
+    ).toEqual([]);
+  });
+});

src/__tests__/drift/providers.ts

@@ -727,3 +727,111 @@ export async function listOpenRouterVideoModels(apiKey: string): Promise<string[
     return json.data.map((m) => m.id);
   });
 }
+
+// ---------------------------------------------------------------------------
+// fal.ai queue lifecycle
+// ---------------------------------------------------------------------------
+
+/** One lifecycle step's HTTP status + parsed JSON envelope. */
+interface FalQueueStep {
+  status: number;
+  body: Record<string, unknown> | null;
+}
+
+/** The three cost-safe queue envelopes the canary observes. */
+export interface FalQueueCanaryResult {
+  submit: FalQueueStep;
+  statusPoll: FalQueueStep;
+  cancel: FalQueueStep;
+}
+
+/**
+ * Cost-safe live queue-lifecycle canary for the fal.ai queue surface
+ * (`src/fal.ts` handleFal / walkFalQueue). Drives the REAL fal queue API and
+ * returns the SUBMIT, STATUS, and CANCEL envelope bodies for drift comparison.
+ *
+ * COST SAFETY: fal bills COMPUTE only when a queued job actually RUNS. This
+ * canary submits a job (free) and cancels it IMMEDIATELY — while it is still
+ * `IN_QUEUE` no compute is charged, so the expected cost is $0. The cancel is
+ * issued even if the status poll throws (see the finally-style flow below) so a
+ * job is never left to run. The `response_url` (completed result) endpoint is
+ * NEVER fetched — that is the only paid retrieval, and its envelope stays
+ * STATIC-only. Residual: if the (cheapest) model races to completion before the
+ * cancel lands, at most one sub-cent generation is billed.
+ */
+export async function falQueueLifecycleCanary(
+  apiKey: string,
+  modelId: string,
+  input: object,
+): Promise<FalQueueCanaryResult> {
+  return withInfraErrorTag("fal Queue Lifecycle", async () => {
+    const authHeaders = { Authorization: `Key ${apiKey}` };
+
+    // 1. Submit — enqueues the job. Free; compute is billed only when it RUNS.
+    const submitUrl = `https://queue.fal.run/${modelId}`;
+    const submitRes = await fetchWithRetry(submitUrl, {
+      method: "POST",
+      headers: { ...authHeaders, "Content-Type": "application/json" },
+      body: JSON.stringify(input),
+    });
+    const submitRaw = await submitRes.text();
+    const submitBody = parseJsonResponse(
+      submitRaw,
+      submitRes.status,
+      "fal queue submit",
+      submitUrl,
+    ) as Record<string, unknown>;
+
+    // fal returns absolute lifecycle URLs; fall back to the documented layout.
+    const requestId = String(submitBody.request_id ?? "");
+    const statusUrl =
+      typeof submitBody.status_url === "string"
+        ? submitBody.status_url
+        : `${submitUrl}/requests/${requestId}/status`;
+    const cancelUrl =
+      typeof submitBody.cancel_url === "string"
+        ? submitBody.cancel_url
+        : `${submitUrl}/requests/${requestId}/cancel`;
+
+    // 2. Status — metadata only (never the result payload). Capture any error so
+    // the cancel below still fires: leaving a job to RUN is the only cost risk.
+    let statusPoll: FalQueueStep | null = null;
+    let statusError: unknown = null;
+    try {
+      const statusRes = await fetchWithRetry(statusUrl, { method: "GET", headers: authHeaders });
+      const statusRaw = await statusRes.text();
+      statusPoll = {
+        status: statusRes.status,
+        body: parseJsonResponse(
+          statusRaw,
+          statusRes.status,
+          "fal queue status",
+          statusUrl,
+        ) as Record<string, unknown>,
+      };
+    } catch (err) {
+      statusError = err;
+    }
+
+    // 3. Cancel IMMEDIATELY — while IN_QUEUE, no compute is charged. Real fal
+    // returns 200 `{status:"CANCELLATION_REQUESTED"}` (or 400
+    // `{status:"ALREADY_COMPLETED"}` on a race); both are valid envelopes.
+    const cancelRes = await fetchWithRetry(cancelUrl, { method: "PUT", headers: authHeaders });
+    const cancelRaw = await cancelRes.text();
+    let cancelBody: Record<string, unknown> | null = null;
+    try {
+      cancelBody = cancelRaw ? (JSON.parse(cancelRaw) as Record<string, unknown>) : null;
+    } catch {
+      cancelBody = null;
+    }
+
+    // Surface a status-poll failure only after cancellation is guaranteed.
+    if (statusError) throw statusError;
+
+    return {
+      submit: { status: submitRes.status, body: submitBody },
+      statusPoll: statusPoll!,
+      cancel: { status: cancelRes.status, body: cancelBody },
+    };
+  });
+}