diff --git a/.github/workflows/fix-drift.yml b/.github/workflows/fix-drift.yml index 4e054f4b..5ea33041 100644 --- a/.github/workflows/fix-drift.yml +++ b/.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 diff --git a/.github/workflows/test-drift.yml b/.github/workflows/test-drift.yml index 73f8073b..4c97902b 100644 --- a/.github/workflows/test-drift.yml +++ b/.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 diff --git a/src/__tests__/drift/fal-queue.drift.ts b/src/__tests__/drift/fal-queue.drift.ts index 34e92340..b12ebc53 100644 --- a/src/__tests__/drift/fal-queue.drift.ts +++ b/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([]); + }); +}); diff --git a/src/__tests__/drift/providers.ts b/src/__tests__/drift/providers.ts index 434749e6..1c2251c5 100644 --- a/src/__tests__/drift/providers.ts +++ b/src/__tests__/drift/providers.ts @@ -727,3 +727,111 @@ export async function listOpenRouterVideoModels(apiKey: string): Promise m.id); }); } + +// --------------------------------------------------------------------------- +// fal.ai queue lifecycle +// --------------------------------------------------------------------------- + +/** One lifecycle step's HTTP status + parsed JSON envelope. */ +interface FalQueueStep { + status: number; + body: Record | 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 { + 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; + + // 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, + }; + } 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 | null = null; + try { + cancelBody = cancelRaw ? (JSON.parse(cancelRaw) as Record) : 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 }, + }; + }); +}