feat(api): send team concurrency with FirePDF async submits (firecrawl#4051)

* feat(api): send team concurrency with FirePDF async submits

FirePDF's per-team admission (its ENG-5049) runs in observation mode
and needs the account context to measure anything. Submit now carries
team_concurrency from the cached ACUC lookup, best-effort: lookup
failure or absence never blocks a scrape — FirePDF simply skips team
observation for that submit. team_tier is deliberately NOT sent: no
plan/tier string exists in this package today; FirePDF defaults the
tier for observation, and the authoritative tier source is being
decided alongside its entitlement policy.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* refactor: snapshot team concurrency via internalOptions, not a re-fetch

Per review: the ACUC is already in hand at the controller, so thread
it down instead of re-reading in the worker. teamConcurrency is
snapshotted from req.acuc at the same two sites that snapshot
teamFlags (v2 scrape + parse) and rides the job payload; the FirePDF
async engine reads meta.internalOptions.teamConcurrency and does no
lookup. Also pins the entitlement to acceptance time.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: abimaelmartell

apps/api/src/controllers/v2/parse.ts

@@ -512,6 +512,7 @@ export async function parseController(
                       bypassBilling: isDirectToBullMQ || !shouldBill,
                       zeroDataRetention,
                       teamFlags: req.acuc?.flags ?? null,
+                      teamConcurrency: req.acuc?.concurrency ?? null,
                       uploadedFile: file,
                       forceEngine,
                       isParse: true,

apps/api/src/controllers/v2/scrape.ts

@@ -328,6 +328,7 @@ export async function scrapeController(
                       bypassBilling: isDirectToBullMQ || !shouldBill,
                       zeroDataRetention,
                       teamFlags: req.acuc?.flags ?? null,
+                      teamConcurrency: req.acuc?.concurrency ?? null,
                       agentIndexOnly: (req as any).agentIndexOnly ?? false,
                       threatProtection: threatProtection.policy ?? undefined,
                     },

apps/api/src/scraper/scrapeURL/engines/pdf/__tests__/firePDFAsync.test.ts

@@ -7,6 +7,7 @@ vi.mock("../../../../../lib/gcs-pdf-cache", () => ({
   savePdfResultToCache: vi.fn(async () => null),
 }));
 
+
 import {
   FirePdfAsyncFailure,
   scrapePDFWithFirePDFAsync,
@@ -71,6 +72,7 @@ function makeMeta(overrides: Record<string, unknown> = {}) {
     internalOptions: {
       zeroDataRetention: false,
       teamId: "team-x",
+      teamConcurrency: 12,
       crawlId: undefined,
     },
     options: {
@@ -91,11 +93,18 @@ function makeFetchFromSequence(
     url: string;
     method: string;
     headers: Record<string, string> | undefined;
+    body: unknown;
   }> = [];
   const cursor = { idx: 0 };
   const fetchImpl: any = async (url: string, init: any) => {
     const method = (init?.method ?? "GET").toUpperCase();
-    calls.push({ url, method, headers: init?.headers });
+    let body: unknown;
+    try {
+      body = init?.body ? JSON.parse(init.body) : undefined;
+    } catch {
+      body = init?.body;
+    }
+    calls.push({ url, method, headers: init?.headers, body });
     const matcher = matchers[cursor.idx++];
     if (!matcher) {
       throw new Error(
@@ -136,6 +145,7 @@ describe("scrapePDFWithFirePDFAsync", () => {
       internalOptions: {
         zeroDataRetention: true,
         teamId: "team-x",
+      teamConcurrency: 12,
         crawlId: undefined,
       },
     });
@@ -241,6 +251,49 @@ describe("scrapePDFWithFirePDFAsync", () => {
     expect(result.pagesProcessed).toBe(12);
     expect(fallback).not.toHaveBeenCalled();
     expect(calls).toHaveLength(4);
+    // Account context rides the submit body (FirePDF ENG-5049).
+    expect((calls[0].body as { team_concurrency?: number }).team_concurrency).toBe(12);
+  });
+
+  it("submits without team context when the snapshot is absent", async () => {
+    const { fetchImpl, calls } = makeFetchFromSequence([
+      {
+        matchUrl: /\/jobs$/,
+        matchMethod: "POST",
+        response: {
+          status: 200,
+          body: { scrape_id: "scrape-id-test", status: "done", lane: "fast", retry_after_ms: 0 },
+        },
+      },
+      {
+        matchUrl: /\/jobs\/scrape-id-test\/result$/,
+        matchMethod: "GET",
+        response: {
+          status: 200,
+          body: {
+            schema_version: 1,
+            markdown: "# No context",
+            pages_processed: 1,
+            failed_pages: null,
+            partial_pages: null,
+          },
+        },
+      },
+    ]);
+    const fallback = vi.fn();
+
+    const meta = makeMeta();
+    meta.internalOptions.teamConcurrency = null;
+    const result = await scrapePDFWithFirePDFAsync(meta, "BASE64", undefined, undefined, undefined, {
+      fetchImpl,
+      fallbackImpl: fallback,
+      sleepImpl: noopSleep,
+    });
+
+    // Missing snapshot must never block the scrape — field simply absent.
+    expect(result.markdown).toBe("# No context");
+    expect((calls[0].body as { team_concurrency?: number }).team_concurrency).toBeUndefined();
+    expect(fallback).not.toHaveBeenCalled();
   });
 
   it("cancels accepted work when polling is abandoned", async () => {
@@ -282,7 +335,7 @@ describe("scrapePDFWithFirePDFAsync", () => {
 
     expect(error).toBeInstanceOf(FirePdfAsyncFailure);
     expect(error.reason).toBe("network_error");
-    expect(calls).toEqual([
+    expect(calls.map(({ url, method, headers }) => ({ url, method, ...(headers !== undefined && { headers }) }))).toEqual([
       {
         url: "http://fire-pdf.test/jobs",
         method: "POST",
@@ -401,7 +454,7 @@ describe("scrapePDFWithFirePDFAsync", () => {
     expect(err).toBeInstanceOf(FirePdfAsyncFailure);
     expect(err.reason).toBe("network_error");
     expect(fallback).not.toHaveBeenCalled();
-    expect(calls).toEqual([
+    expect(calls.map(({ url, method }) => ({ url, method }))).toEqual([
       { method: "POST", url: "http://fire-pdf.test/jobs" },
       {
         method: "DELETE",

apps/api/src/scraper/scrapeURL/engines/pdf/fire-pdf/async.ts

@@ -82,6 +82,16 @@ export async function scrapePDFWithFirePDFAsync(
   const deadlineAt = new Date(submitTime + deadlineFromNow).toISOString();
   const pollingDeadline = submitTime + deadlineFromNow + POLL_TIMEOUT_BUFFER_MS;
 
+  // Account context for FirePDF's per-team admission observation,
+  // snapshotted from the request ACUC into internalOptions at
+  // acceptance (same pattern as teamFlags) — no re-fetch here. Absence
+  // means FirePDF skips team observation for this submit.
+  const rawConcurrency = meta.internalOptions.teamConcurrency;
+  const teamConcurrency =
+    typeof rawConcurrency === "number" && rawConcurrency > 0
+      ? rawConcurrency
+      : undefined;
+
   // ── Step 1: POST /jobs ────────────────────────────────────────────────
   let submissionAccepted = false;
   let terminalReached = false;
@@ -96,6 +106,7 @@ export async function scrapePDFWithFirePDFAsync(
       pagesProcessed,
       mode,
       deadlineAt,
+      teamConcurrency,
       fetchImpl,
     });
     submissionAccepted = true;

apps/api/src/scraper/scrapeURL/engines/pdf/fire-pdf/submit.ts

@@ -20,6 +20,9 @@ type SubmitArgs = {
   pagesProcessed: number | undefined;
   mode: PDFMode | undefined;
   deadlineAt: string;
+  /** Team's sold concurrency from the ACUC (ENG-5049 account context).
+   * Optional: entitlement lookup must never block or fail a scrape. */
+  teamConcurrency: number | undefined;
   fetchImpl: typeof undiciFetch;
 };
 
@@ -56,6 +59,7 @@ export async function submitJob(args: SubmitArgs): Promise<SubmitOutcome> {
     pagesProcessed,
     mode,
     deadlineAt,
+    teamConcurrency,
     fetchImpl,
   } = args;
   const scrapeId = meta.id;
@@ -72,6 +76,12 @@ export async function submitJob(args: SubmitArgs): Promise<SubmitOutcome> {
     ...(meta.internalOptions.crawlId && {
       crawl_id: meta.internalOptions.crawlId,
     }),
+    // FirePDF per-team admission observation (its ENG-5049): sold
+    // concurrency from the account context. Absence means FirePDF
+    // skips team observation for this submit — never a rejection.
+    ...(teamConcurrency !== undefined && {
+      team_concurrency: teamConcurrency,
+    }),
     options: {
       ...(pagesProcessed !== undefined && { pages_estimate: pagesProcessed }),
       ...(maxPages !== undefined && { max_pages: maxPages }),

apps/api/src/scraper/scrapeURL/index.ts

@@ -493,6 +493,10 @@ export type InternalOptions = {
   bypassBilling?: boolean;
   zeroDataRetention?: boolean;
   teamFlags?: TeamFlags;
+  /** Team's sold concurrency, snapshotted from the request ACUC at
+   * acceptance (same pattern as teamFlags). Rides the job payload so
+   * downstream engines (FirePDF async account context) never re-fetch. */
+  teamConcurrency?: number | null;
 
   /**
    * Effective threat protection policy for this scrape, resolved at the