perf(pcat): rotating Floxy residential transport + resilience (decode latency)

Prod decode was dominated by the DataImpulse proxy: ~50% of ports dead at any
moment, 57% of captured tokens discarded by validation, the dead-port slot
never evicted (invalidateSession only fired on 401/403, never on transport
timeout), and a cold pool blocked the user inline on a multi-second capture.
pcat sub-timing was p50 8.3s / p90 56s, blowing the 25s decode budget.

Verified on prod: the TWS- token is NOT request-time IP-bound (0 auth 401/403
across thousands of calls; a token captured on one IP replays 200 through
arbitrary fresh residential IPs). So the proxy can rotate freely.

- Transport swap to Floxy residential (PCAT_PROXY_PROVIDER=floxy|dataimpulse|none).
  Call/validate leg = a fresh exit IP per request (max IP diversity → min per-IP
  ban); capture leg = a sticky session per Playwright attempt. DataImpulse +
  direct kept as env fallbacks. Token decoupled from the slot (proxyPort removed).
- Prune 3 zero-yield JWT sites (knkauto/autodo.kz/flynestauto — 0 captures, ~80%
  of attempts); add PCAT_JWT_SITES env override.
- Lower catalog-scrape timeout 30s→10s (env-tunable) so a slow IP aborts within
  budget and the retry rotates to a fresh IP.
- Cold-pool fast-fail on the decode race: warm in background, let EMEX/PL24
  answer instead of blocking inline; treated as a transient miss (no neg-cache).
- Evict a token after N consecutive transport failures on rotating IPs (the
  dead-slot bug); reset on any 2xx.
- PCAT circuit breaker mirroring PL24 (transient-only; cold-skip neutral; skip
  marks outcome.transient so a pcat-only VIN isn't negative-cached 30m).

Typecheck + Biome + unit tests (categories/vehicles) green. Default flips pcat
to Floxy on next deploy; instant rollback via PCAT_PROXY_PROVIDER.

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

apps/api/src/integrations/parts-catalogs/parts-catalogs-auth.service.ts

@@ -1,12 +1,16 @@
 /**
- * Parts-Catalogs Auth Service — v3 token warm pool via Playwright + DataImpulse proxy
+ * Parts-Catalogs Auth Service — v3 token warm pool via Playwright + rotating proxy
  *
  * Tokens are captured by navigating to partner sites that embed the v3 widget.
  * The widget calls /v3/api/proxy/* with `x-api-key: TWS-{UUID}` and four other
  * X-* headers (api-path, gui-version, user-id, origin, referer); we intercept
  * all of them so backend requests can replay the exact header set.
  *
- * Tokens are IP-bound — same proxy port must be reused for the API calls.
+ * Transport (PCAT_PROXY_PROVIDER): "floxy" residential (default) rotates a fresh
+ * exit IP per call (the token is NOT IP-bound — see parts-catalogs.types.ts — so
+ * max IP diversity = min per-IP ban), with a sticky IP only per Playwright
+ * capture; "dataimpulse" is the legacy port-bound datacenter proxy; "none" calls
+ * direct from the server IP.
  *
  * Warm pool behavior:
  *   24/7 → proactive: maintain >= 1 slot, auto-refresh ~9.5min before expiry.
@@ -22,10 +26,10 @@ import { RedisService } from "../../redis/redis.service";
 import { JwtSlot, PcatJwtToken, PcatSession } from "./parts-catalogs.types";
 
 // Shared single-slot cache so dev + prod (and any restarted container) can
-// inherit a freshly-captured warm slot instead of paying the Playwright
-// capture cost from cold. Token is still IP-bound to its proxyPort; if a
-// reader on a different IP gets 401/403, invalidateSession's existing
-// fallback re-captures locally. So worst case = today's behavior.
+// inherit a freshly-captured warm token instead of paying the Playwright
+// capture cost from cold. The token is portable (not IP-bound), so any
+// container can serve it through its own rotating call-leg IPs; if it's dead
+// at the source, invalidateSession drops it and the next read re-captures.
 const REDIS_SLOT_KEY = "pcat:jwt:slot";
 const REDIS_SAFETY_BUFFER_S = 60; // don't serve from Redis if <60s of life left
 
@@ -62,30 +66,46 @@ const RPM_PER_SLOT = 6; // 1 token per 6 req/min
  * Each site uses a different proxy port (IP) to avoid rate limiting.
  */
 // Order matters: round-robin starts from index 0, so the first site is the one
-// a cold pool waits on. Sites that consistently load in <10s through the proxy
-// stay at the top so initial JWT capture finishes in seconds.
-// (e-acca.com was previously included last but removed — the current DataImpulse
-// rotating proxy 74.81.81.81:10000-10999 cannot reach it; page.goto always
-// blocked until PAGE_TIMEOUT instead of failing fast like the other sites.)
-const JWT_SITES = [
+// a cold pool waits on. Sites that consistently load fast stay at the top so
+// initial JWT capture finishes in seconds.
+// Pruned 2026-06-10 to the 6 sites that actually yielded captures on prod —
+// knkauto.ru / autodo.kz / flynestauto.com produced 0 captures yet consumed
+// ~80% of capture attempts (the per-site cooldown kept forcing the retry loop
+// onto them). Override the list at runtime with PCAT_JWT_SITES (comma-separated)
+// without a deploy, e.g. to give dev a disjoint subset.
+const DEFAULT_JWT_SITES = [
   "https://www.alkatalog.com/cats/#/catalogs",
   "https://auto-komplekt.ru/goodvin#/catalogs",
   "https://www.autotrade.md/cats/#/catalogs",
   "https://www.e-trak.ru/cats/#/catalogs",
   "https://www.autopolyus.ru/cats/#/catalogs",
-  "https://knkauto.ru/goodvin#/catalogs",
-  "https://www.autodo.kz/#/catalogs",
   "https://avtoman124.ru/goodvin#/catalogs",
-  "https://flynestauto.com/auto-parts-oem-catalog",
 ];
 
-// DataImpulse proxy defaults (port-based IP rotation)
+// DataImpulse proxy defaults (port-based IP rotation) — legacy. Half its ports
+// were dead at any moment (measured 50% connect-timeout), which drove the bulk of
+// pcat decode latency + failures. Kept selectable via PCAT_PROXY_PROVIDER for
+// instant rollback, but Floxy is the default.
 const DI_HOST = "gw.dataimpulse.com";
 const DI_PORT_MIN = 10000;
 const DI_PORT_MAX = 10999;
 const DI_DEFAULT_USER = "1726bbe361918676d44e";
 const DI_DEFAULT_PASS = "78ebc3d881de6ec0";
 
+// Floxy residential defaults (session-based IP rotation; same account EMEX uses).
+// Call leg → plain creds = a fresh residential exit IP per request (max IP
+// diversity → minimal per-IP ban exposure, since each IP sees ~1 request).
+// Capture leg → a sticky session id pins one IP for the duration of a single
+// Playwright capture so the partner site sees a coherent session.
+const FLOXY_HOST = "residential.floxy.io";
+const FLOXY_PORT = 12321;
+const FLOXY_DEFAULT_USER = "d739255e819b";
+const FLOXY_DEFAULT_PASS = "9092873ba4e0";
+const FLOXY_CAPTURE_LIFETIME_S = 60; // sticky window for one capture attempt
+
+type ProxyProvider = "floxy" | "dataimpulse" | "none";
+type ProxyLeg = "call" | "capture";
+
 /** Simple counting semaphore (same pattern as EmexBrowserService) */
 class Semaphore {
   private current = 0;
@@ -119,18 +139,20 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
 
   private browser: Browser | null = null;
   private launching: Promise<void> | null = null;
-  // Concurrent captures capped at JWT_SITES.length — the real bottleneck is
-  // how many distinct partner sites we can drive in parallel (one capture per
-  // site, since each site = a different IP / cookie origin). Ports are
-  // ~unlimited (10000-10999) and Playwright contexts are isolated; the
-  // synchronous siteIndex round-robin hands each concurrent capture a
-  // different site so they don't collide.
-  private readonly semaphore = new Semaphore(JWT_SITES.length);
+  // Active JWT capture sites (PCAT_JWT_SITES override or DEFAULT_JWT_SITES),
+  // resolved in the constructor.
+  private readonly jwtSites: string[];
+  // Concurrent captures capped at jwtSites.length — the real bottleneck is how
+  // many distinct partner sites we can drive in parallel (one capture per site,
+  // since each site = a different cookie origin). Playwright contexts are
+  // isolated; the synchronous siteIndex round-robin hands each concurrent
+  // capture a different site so they don't collide. Initialized in constructor.
+  private readonly semaphore: Semaphore;
 
   // Pool state
   private pool: JwtSlot[] = [];
   private siteLastUsedAt = new Map<string, number>();
-  private siteIndex = 0; // round-robin across JWT_SITES
+  private siteIndex = 0; // round-robin across jwtSites
   private requestRoundRobin = 0; // round-robin across pool slots
 
   // Business hours scheduling
@@ -139,20 +161,108 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
   // RPM tracking
   private requestTimestamps: number[] = [];
 
-  // Config
-  private readonly useProxy: boolean;
-  private readonly proxyHost: string;
-  private readonly proxyUser: string;
-  private readonly proxyPass: string;
+  // Transport config — see ProxyProvider. The TWS- token is NOT request-time
+  // IP-bound (parts-catalogs.types.ts), so the call leg rotates exit IPs per
+  // request and only the capture leg holds a sticky IP per attempt.
+  private readonly proxyProvider: ProxyProvider;
+  // Floxy residential
+  private readonly floxyHost: string;
+  private readonly floxyPort: number;
+  private readonly floxyUser: string;
+  private readonly floxyPass: string;
+  private readonly floxyCaptureLifetime: number;
+  // DataImpulse legacy (port-rotated)
+  private readonly diHost: string;
+  private readonly diUser: string;
+  private readonly diPass: string;
 
   constructor(
     private configService: ConfigService,
     private readonly redis: RedisService,
   ) {
-    this.useProxy = this.configService.get<string>("PCAT_USE_PROXY", "true") === "true";
-    this.proxyHost = this.configService.get<string>("PCAT_PROXY_HOST", DI_HOST);
-    this.proxyUser = this.configService.get<string>("PCAT_PROXY_USER", DI_DEFAULT_USER);
-    this.proxyPass = this.configService.get<string>("PCAT_PROXY_PASS", DI_DEFAULT_PASS);
+    const cfg = this.configService;
+
+    // Active capture sites: PCAT_JWT_SITES override (comma-separated) or the
+    // pruned default. Empty/blank entries are dropped; falls back to default if
+    // the override resolves to nothing.
+    const sitesRaw = cfg.get<string>("PCAT_JWT_SITES", "");
+    const parsedSites = sitesRaw
+      .split(",")
+      .map((s) => s.trim())
+      .filter((s) => s.length > 0);
+    this.jwtSites = parsedSites.length > 0 ? parsedSites : DEFAULT_JWT_SITES;
+    this.semaphore = new Semaphore(this.jwtSites.length);
+
+    // Provider selection. Back-compat: PCAT_USE_PROXY=false still forces direct.
+    const rawProvider = cfg.get<string>("PCAT_PROXY_PROVIDER", "floxy").toLowerCase();
+    const useProxy = cfg.get<string>("PCAT_USE_PROXY", "true") === "true";
+    this.proxyProvider = !useProxy
+      ? "none"
+      : rawProvider === "dataimpulse" || rawProvider === "none"
+        ? (rawProvider as ProxyProvider)
+        : "floxy";
+
+    const toInt = (key: string, def: number): number => {
+      const n = Number(cfg.get(key, def));
+      return Number.isInteger(n) && n >= 0 ? n : def;
+    };
+    const toPort = (key: string, def: number): number => {
+      const n = Number(cfg.get(key, def));
+      return Number.isInteger(n) && n >= 1 && n <= 65535 ? n : def;
+    };
+
+    this.floxyHost = cfg.get<string>("PCAT_FLOXY_HOST", FLOXY_HOST);
+    this.floxyPort = toPort("PCAT_FLOXY_PORT", FLOXY_PORT);
+    this.floxyUser = cfg.get<string>("PCAT_FLOXY_USER", FLOXY_DEFAULT_USER);
+    this.floxyPass = cfg.get<string>("PCAT_FLOXY_PASS", FLOXY_DEFAULT_PASS);
+    this.floxyCaptureLifetime = toInt("PCAT_FLOXY_CAPTURE_LIFETIME", FLOXY_CAPTURE_LIFETIME_S);
+
+    this.diHost = cfg.get<string>("PCAT_PROXY_HOST", DI_HOST);
+    this.diUser = cfg.get<string>("PCAT_PROXY_USER", DI_DEFAULT_USER);
+    this.diPass = cfg.get<string>("PCAT_PROXY_PASS", DI_DEFAULT_PASS);
+
+    this.logger.log(
+      this.proxyProvider === "floxy"
+        ? `PCAT transport: floxy residential ${this.floxyHost}:${this.floxyPort} (call=rotating, capture=sticky ${this.floxyCaptureLifetime}s)`
+        : this.proxyProvider === "dataimpulse"
+          ? `PCAT transport: dataimpulse ${this.diHost}:${DI_PORT_MIN}-${DI_PORT_MAX}`
+          : "PCAT transport: direct (no proxy)",
+    );
+  }
+
+  /**
+   * Build a proxy for one leg. "call" rotates the exit IP per invocation (the
+   * token is not IP-bound, so each decode/scrape/validate request gets a fresh
+   * residential IP → max IP diversity, min ban). "capture" pins a sticky IP for
+   * the lifetime of a single Playwright capture so the partner site sees a
+   * coherent session. Returns nulls when provider is "none" (direct).
+   */
+  private buildProxy(leg: ProxyLeg): {
+    proxyUrl: string | null;
+    proxyConfig: { server: string; username: string; password: string } | null;
+  } {
+    if (this.proxyProvider === "none") return { proxyUrl: null, proxyConfig: null };
+
+    if (this.proxyProvider === "dataimpulse") {
+      const port = this.allocatePort();
+      const server = `http://${this.diHost}:${port}`;
+      return {
+        proxyUrl: `http://${this.diUser}:${this.diPass}@${this.diHost}:${port}`,
+        proxyConfig: { server, username: this.diUser, password: this.diPass },
+      };
+    }
+
+    // floxy
+    let password = this.floxyPass;
+    if (leg === "capture" && this.floxyCaptureLifetime > 0) {
+      const sid = Math.random().toString(36).slice(2, 10);
+      password = `${this.floxyPass}_session-${sid}_lifetime-${this.floxyCaptureLifetime}`;
+    }
+    const server = `http://${this.floxyHost}:${this.floxyPort}`;
+    return {
+      proxyUrl: `http://${this.floxyUser}:${password}@${this.floxyHost}:${this.floxyPort}`,
+      proxyConfig: { server, username: this.floxyUser, password },
+    };
   }
 
   async onModuleInit(): Promise<void> {
@@ -193,10 +303,14 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
   // ─── Public API ───────────────────────────────────────────
 
   /**
-   * Acquire a matched JWT + proxy session from the pool.
-   * Returns a valid slot (round-robin) or captures on-demand if pool is empty.
+   * Acquire a JWT session from the pool.
+   * Returns a valid slot (round-robin) or, if the pool is empty, captures
+   * on-demand. With `blockOnEmpty: false` it does NOT block on a cold pool —
+   * it kicks off a background warm-up and returns null so a latency-sensitive
+   * caller (the decode race) can let a faster source answer instead of eating
+   * the multi-second capture cost inline.
    */
-  async acquireSession(): Promise<PcatSession> {
+  async acquireSession(opts?: { blockOnEmpty?: boolean }): Promise<PcatSession | null> {
     this.trackRequest();
 
     // Try to find a valid slot in the pool
@@ -207,14 +321,26 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
       return this.slotToSession(slot);
     }
 
-    // No valid slot — capture on-demand
+    // No valid slot. Latency-sensitive callers fast-fail and warm in background.
+    if (opts?.blockOnEmpty === false) {
+      this.logger.log("JWT pool empty — fast-fail (warming in background)");
+      this.captureToPool().catch((err) => {
+        this.logger.warn(`Background warm capture failed: ${(err as Error).message}`);
+      });
+      return null;
+    }
+
+    // Capture on-demand (blocks the caller).
     this.logger.log("JWT pool empty — capturing on-demand...");
     const newSlot = await this.captureToPool();
     return this.slotToSession(newSlot);
   }
 
   /**
-   * Invalidate a session after 401/403, remove the slot, and capture a replacement.
+   * Invalidate a session (401/403, or N consecutive transport failures on
+   * rotating IPs = a token that's dead at the source), remove the slot, and
+   * capture a replacement only if the pool is now empty — a surviving sibling
+   * plus the scheduled refresh/scale-up covers the rest without a capture storm.
    */
   async invalidateSession(session: PcatSession): Promise<void> {
     const idx = this.pool.indexOf(session._slot);
@@ -222,17 +348,18 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
       this.clearSlotTimer(this.pool[idx]);
       this.pool.splice(idx, 1);
       this.logger.log(
-        `JWT pool: slot invalidated (port ${session._slot.proxyPort}), ${this.pool.length} remaining`,
+        `JWT pool: slot invalidated (${new URL(session._slot.siteUsed).hostname}), ${this.pool.length} remaining`,
       );
     }
 
-    // The Redis-shared slot may point at a now-401'd JWT (e.g. IP rotated
-    // away on the proxy). Drop it so other containers re-capture instead of
-    // inheriting the same dead slot.
+    // The Redis-shared slot may point at this same now-dead token. Drop it so
+    // a cold-booting container re-captures instead of hydrating the dead token.
     this.deleteFromRedis().catch(() => {});
 
-    // Capture replacement in background (don't block the caller's retry)
-    if (this.isBusinessHours() || this.pool.length === 0) {
+    // Only force an immediate replacement when the pool emptied; otherwise let
+    // the surviving slot serve and the refresh/scale-up timers backfill, so a
+    // burst of evictions can't stampede the capture pipeline.
+    if (this.pool.length === 0) {
       this.captureToPool().catch((err) => {
         this.logger.error(`Replacement capture failed: ${(err as Error).message}`);
       });
@@ -262,16 +389,10 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
   }
 
   private slotToSession(slot: JwtSlot): PcatSession {
-    const proxyUrl = this.useProxy
-      ? `http://${this.proxyUser}:${this.proxyPass}@${this.proxyHost}:${slot.proxyPort}`
-      : null;
-    const proxyConfig = this.useProxy
-      ? {
-          server: `http://${this.proxyHost}:${slot.proxyPort}`,
-          username: this.proxyUser,
-          password: this.proxyPass,
-        }
-      : null;
+    // Fresh rotating call-leg proxy per request — and per retry, since
+    // fetchWithAuth re-acquires the session each attempt → a new exit IP each
+    // time, so a slow/blocked residential IP is shed on the next try.
+    const { proxyUrl, proxyConfig } = this.buildProxy("call");
     return {
       apiKey: slot.jwt.raw,
       apiPath: slot.jwt.apiPath,
@@ -294,17 +415,16 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
 
       const maxRetries = 4;
       for (let attempt = 0; attempt < maxRetries; attempt++) {
-        const port = this.allocatePort();
         const siteUrl = this.getAvailableSite();
 
-        const jwt = await this.attemptCapture(siteUrl, port);
+        const jwt = await this.attemptCapture(siteUrl);
         if (jwt) {
-          // Validate the JWT+proxy combo against the real upstream before
-          // publishing the slot — see VALIDATION_VIN comment above.
-          const valid = await this.validateSlot(jwt, port);
+          // Validate the token against the real upstream (on a fresh rotating
+          // call-leg IP) before publishing the slot — see VALIDATION_VIN.
+          const valid = await this.validateSlot(jwt);
           if (!valid) {
             this.logger.warn(
-              `JWT capture attempt ${attempt + 1}/${maxRetries} validation failed (port ${port}, ${new URL(siteUrl).hostname}), trying next site...`,
+              `JWT capture attempt ${attempt + 1}/${maxRetries} validation failed (${new URL(siteUrl).hostname}), trying next site...`,
             );
             // Mark site as used anyway so we rotate away rather than retry it.
             this.siteLastUsedAt.set(siteUrl, Date.now());
@@ -313,10 +433,10 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
 
           const slot: JwtSlot = {
             jwt,
-            proxyPort: port,
             siteUsed: siteUrl,
             capturedAt: Date.now(),
             refreshTimer: null,
+            failCount: 0,
           };
 
           // Mark site as used
@@ -352,13 +472,14 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
   }
 
   /**
-   * Probe the just-captured token + proxy port with a single cheap upstream
-   * call (/car/info with a public demo VIN). Returns true iff the upstream
-   * accepts the combo with 2xx within VALIDATION_TIMEOUT_MS. A failure here
-   * (proxy dead, IP blocked, token 401/403) is far cheaper to absorb at
-   * capture time than to inherit when a real user clicks a category.
+   * Probe the just-captured token with a single cheap upstream call (/car/info
+   * with a public demo VIN) on a fresh rotating call-leg IP — the SAME transport
+   * a real request will use. Returns true iff upstream accepts it with 2xx within
+   * VALIDATION_TIMEOUT_MS. Validating on the call-leg transport (not the capture
+   * IP) is what makes the check meaningful: it rejects a token only if it can't
+   * be served the way it will actually be served.
    */
-  private async validateSlot(jwt: PcatJwtToken, port: number): Promise<boolean> {
+  private async validateSlot(jwt: PcatJwtToken): Promise<boolean> {
     const url = `${PCAT_API_BASE}/car/info?q=${VALIDATION_VIN}`;
     const fetchOptions: RequestInit & { dispatcher?: any } = {
       method: "GET",
@@ -375,10 +496,11 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
       },
       signal: AbortSignal.timeout(VALIDATION_TIMEOUT_MS),
     };
-    if (this.useProxy) {
+    const { proxyUrl } = this.buildProxy("call");
+    if (proxyUrl) {
       const { ProxyAgent } = await import("undici");
       fetchOptions.dispatcher = new ProxyAgent({
-        uri: `http://${this.proxyUser}:${this.proxyPass}@${this.proxyHost}:${port}`,
+        uri: proxyUrl,
         connect: { timeout: 6_000 },
       });
     }
@@ -539,15 +661,16 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
 
   private getAvailableSite(): string {
     const now = Date.now();
+    const sites = this.jwtSites;
 
     // Try round-robin, preferring sites not on cooldown
-    for (let i = 0; i < JWT_SITES.length; i++) {
-      const idx = (this.siteIndex + i) % JWT_SITES.length;
-      const site = JWT_SITES[idx];
+    for (let i = 0; i < sites.length; i++) {
+      const idx = (this.siteIndex + i) % sites.length;
+      const site = sites[idx];
       const lastUsed = this.siteLastUsedAt.get(site) || 0;
 
       if (now - lastUsed >= SITE_COOLDOWN) {
-        this.siteIndex = (idx + 1) % JWT_SITES.length;
+        this.siteIndex = (idx + 1) % sites.length;
         return site;
       }
     }
@@ -555,15 +678,15 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
     // All on cooldown — pick the one with oldest usage
     let oldestIdx = 0;
     let oldestTime = Number.POSITIVE_INFINITY;
-    for (let i = 0; i < JWT_SITES.length; i++) {
-      const lastUsed = this.siteLastUsedAt.get(JWT_SITES[i]) || 0;
+    for (let i = 0; i < sites.length; i++) {
+      const lastUsed = this.siteLastUsedAt.get(sites[i]) || 0;
       if (lastUsed < oldestTime) {
         oldestTime = lastUsed;
         oldestIdx = i;
       }
     }
-    this.siteIndex = (oldestIdx + 1) % JWT_SITES.length;
-    return JWT_SITES[oldestIdx];
+    this.siteIndex = (oldestIdx + 1) % sites.length;
+    return sites[oldestIdx];
   }
 
   private allocatePort(): number {
@@ -572,19 +695,17 @@ export class PartsCatalogsAuthService implements OnModuleInit, OnModuleDestroy {
 
   // ─── JWT capture via Playwright ───────────────────────────
 
-  private async attemptCapture(siteUrl: string, port: number): Promise<PcatJwtToken | null> {
+  private async attemptCapture(siteUrl: string): Promise<PcatJwtToken | null> {
     let context: BrowserContext | null = null;
     const startTime = Date.now();
 
     try {
-      // Build context options with proxy
+      // Sticky capture proxy: one residential exit IP pinned for the lifetime of
+      // this capture so the partner site + widget see a coherent session.
       const contextOptions: Record<string, unknown> = {};
-      if (this.useProxy) {
-        contextOptions.proxy = {
-          server: `http://${this.proxyHost}:${port}`,
-          username: this.proxyUser,
-          password: this.proxyPass,
-        };
+      const { proxyConfig } = this.buildProxy("capture");
+      if (proxyConfig) {
+        contextOptions.proxy = proxyConfig;
       }
 
       if (!this.browser) throw new Error("PCAT browser not initialized");

apps/api/src/integrations/parts-catalogs/parts-catalogs.service.ts

@@ -2,8 +2,9 @@
  * Parts-Catalogs API Service — HTTP client for parts-catalogs.com v3
  *
  * Calls the v3 widget proxy (gui.parts-catalogs.com/v3/api/proxy/*) with the
- * captured TWS- token + supporting X-* headers. Requests must go through the
- * same DataImpulse proxy port that captured the token (IP-bound).
+ * captured TWS- token + supporting X-* headers. The token is portable (NOT
+ * IP-bound), so each request rotates a fresh exit IP via the auth service's
+ * call-leg proxy (PartsCatalogsAuthService.buildProxy).
  */
 
 import { Injectable, Logger } from "@nestjs/common";
@@ -20,19 +21,38 @@ import {
 } from "./parts-catalogs.types";
 
 const API_BASE = "https://gui.parts-catalogs.com/v3/api/proxy";
-const REQUEST_TIMEOUT = Number(process.env.PCAT_REQUEST_TIMEOUT_MS) || 30_000;
+// Catalog scrape (groups2/parts2). Lowered 30s→10s (2026-06-10): the old 30s
+// default could blow the caller's 25s decode budget on a single slow attempt,
+// and through a rotating residential proxy a healthy call answers in 1-4s. A
+// slow/blocked exit IP now aborts at 10s and the retry re-acquires a fresh IP.
+const REQUEST_TIMEOUT = Number(process.env.PCAT_REQUEST_TIMEOUT_MS) || 10_000;
 const MAX_RETRIES = Number(process.env.PCAT_MAX_RETRIES) || 2;
-// A dead DataImpulse proxy port otherwise stalls for undici's 10s default connect
-// timeout before the retry rotates to a fresh port — three of those blow the
-// caller's 25s decode budget. Fail fast so retries reach a live port in time.
+// undici's ProxyAgent connect.timeout bounds the connect to the proxy entry.
 const PROXY_CONNECT_TIMEOUT = Number(process.env.PCAT_PROXY_CONNECT_TIMEOUT_MS) || 6_000;
-// VIN decode (/car/info) answers in <1s on a healthy proxy, so a long timeout only
-// prolongs dead-port connects. undici's ProxyAgent connect.timeout does NOT bound the
-// connection to the proxy itself (it stays at undici's 10s default), so we bound it at
-// the fetch level: a short per-call timeout + extra retries makes a stuck DataImpulse
-// port abort fast and rotate to a live one within the caller's 25s decode budget.
+// VIN decode (/car/info) answers in 1-4s through the rotating residential proxy.
+// A short per-call timeout + retries means a slow exit IP aborts fast and the
+// next attempt re-acquires the session → a fresh IP, all within the 25s budget.
 const DECODE_REQUEST_TIMEOUT = Number(process.env.PCAT_DECODE_TIMEOUT_MS) || 6_000;
 const DECODE_MAX_RETRIES = Number(process.env.PCAT_DECODE_MAX_RETRIES) || 4;
+// Evict a pooled token after this many consecutive transport failures (reset on
+// any 2xx). Through the rotating residential proxy each attempt hits a different
+// exit IP, so N failures in a row on N different IPs ⇒ the token is dead at the
+// source, not a flaky IP. Pre-rotation this was the "dead slot never evicted"
+// bug: invalidateSession only fired on 401/403, so a bad token was served until TTL.
+const TRANSPORT_FAIL_EVICT_THRESHOLD = Number(process.env.PCAT_TRANSPORT_FAIL_EVICT) || 2;
+
+/**
+ * Thrown when a fast-fail (blockOnEmpty:false) caller hits a cold pool. NOT a
+ * "VIN not found" — pcat was skipped while the pool warmed, so the caller must
+ * treat it as a transient miss (don't poison the negative cache) and let
+ * another source answer.
+ */
+class PcatColdPoolError extends Error {
+  constructor() {
+    super("pcat cold pool — skipped to avoid inline capture");
+    this.name = "PcatColdPoolError";
+  }
+}
 
 @Injectable()
 export class PartsCatalogsService {
@@ -62,7 +82,8 @@ export class PartsCatalogsService {
   async decodeVin(
     vin: string,
     signal?: AbortSignal,
-    outcome?: { transient: boolean },
+    outcome?: { transient: boolean; coldSkip?: boolean },
+    opts?: { blockOnEmpty?: boolean },
   ): Promise<PcatVinResult | null> {
     if (!(await this.posthog.isSourceLive("parts-catalogs"))) {
       this.logger.warn("parts-catalogs disabled by kill switch (kill-source-parts-catalogs)");
@@ -72,6 +93,7 @@ export class PartsCatalogsService {
       const data = await this.fetchWithAuth("/car/info", { q: vin }, signal, {
         timeoutMs: DECODE_REQUEST_TIMEOUT,
         maxRetries: DECODE_MAX_RETRIES,
+        blockOnEmpty: opts?.blockOnEmpty,
       });
 
       if (!data || typeof data !== "object") {
@@ -105,9 +127,23 @@ export class PartsCatalogsService {
       return { cars };
     } catch (err) {
       const e = err as Error & { cause?: unknown };
-      // Distinguish a transient transport blip (proxy timeout / dropped DataImpulse
-      // connection) from a genuine "not found". The caller uses `outcome.transient`
-      // to decide whether to poison the negative cache — a blip must not.
+      // Cold-pool fast-fail: pcat was skipped (pool warming), NOT a definitive
+      // miss. Mark transient so the orchestrator doesn't negative-cache a VIN
+      // pcat hasn't actually evaluated, and stay quiet (this is expected).
+      if (e instanceof PcatColdPoolError) {
+        if (outcome) {
+          outcome.transient = true;
+          // A deliberate cold-pool skip is not a transport fault — flag it so the
+          // circuit breaker doesn't count it (the pool is already warming).
+          outcome.coldSkip = true;
+        }
+        this.logger.debug(`pcat decode skipped for ${vin}: cold pool (warming)`);
+        return null;
+      }
+      // Distinguish a transient transport blip (proxy timeout / dropped
+      // connection) from a genuine "not found". The caller uses
+      // `outcome.transient` to decide whether to poison the negative cache — a
+      // blip must not.
       if (
         outcome &&
         (e.name === "TimeoutError" ||
@@ -212,7 +248,7 @@ export class PartsCatalogsService {
     endpoint: string,
     params?: Record<string, string>,
     externalSignal?: AbortSignal,
-    opts?: { timeoutMs?: number; maxRetries?: number },
+    opts?: { timeoutMs?: number; maxRetries?: number; blockOnEmpty?: boolean },
   ): Promise<any> {
     const maxRetries = opts?.maxRetries ?? MAX_RETRIES;
     const timeoutMs = opts?.timeoutMs ?? REQUEST_TIMEOUT;
@@ -223,7 +259,12 @@ export class PartsCatalogsService {
       if (externalSignal?.aborted) {
         throw new Error("Request aborted by caller");
       }
-      session = await this.authService.acquireSession();
+      session = await this.authService.acquireSession({ blockOnEmpty: opts?.blockOnEmpty });
+      if (!session) {
+        // Cold pool + fast-fail: warming in the background; don't retry, let the
+        // caller fall through to another source.
+        throw new PcatColdPoolError();
+      }
 
       const url = new URL(`${API_BASE}${endpoint}`);
       if (params) {
@@ -263,6 +304,8 @@ export class PartsCatalogsService {
         const response = await fetch(url.toString(), fetchOptions);
 
         if (response.ok) {
+          // Token proven good on this IP — clear any accrued failure streak.
+          session._slot.failCount = 0;
           return await response.json();
         }
 
@@ -291,16 +334,28 @@ export class PartsCatalogsService {
           /fetch failed|ECONNRESET|ECONNREFUSED|EAI_AGAIN|socket hang up|other side closed|terminated|UND_ERR/i.test(
             `${e.message} ${String(e.cause ?? "")}`,
           );
-        if (transient && attempt < maxRetries) {
-          this.logger.warn(
-            `Transient transport error on ${endpoint} (attempt ${attempt + 1}/${maxRetries + 1}): ${e.message}${
-              e.cause ? ` [cause: ${String(e.cause)}]` : ""
-            }`,
-          );
-          // Brief backoff so a momentarily-flaky proxy port can recover; the next
-          // loop iteration re-acquires a session (round-robin across the pool).
-          await new Promise((r) => setTimeout(r, 400 * (attempt + 1)));
-          continue;
+        if (transient) {
+          // A token that keeps failing transport across rotating IPs is dead at
+          // the source — evict it so it stops being round-robin-served (the old
+          // 401/403-only eviction never caught this). Only when a proxy is in use;
+          // direct-mode transport errors are upstream, not token/IP problems.
+          if (session.proxyUrl) {
+            session._slot.failCount = (session._slot.failCount ?? 0) + 1;
+            if (session._slot.failCount >= TRANSPORT_FAIL_EVICT_THRESHOLD) {
+              await this.authService.invalidateSession(session);
+            }
+          }
+          if (attempt < maxRetries) {
+            this.logger.warn(
+              `Transient transport error on ${endpoint} (attempt ${attempt + 1}/${maxRetries + 1}): ${e.message}${
+                e.cause ? ` [cause: ${String(e.cause)}]` : ""
+              }`,
+            );
+            // Brief backoff so a momentarily-flaky exit IP can recover; the next
+            // loop iteration re-acquires a session (a fresh rotating IP).
+            await new Promise((r) => setTimeout(r, 400 * (attempt + 1)));
+            continue;
+          }
         }
         throw err;
       }

apps/api/src/integrations/parts-catalogs/parts-catalogs.types.ts

@@ -1,7 +1,12 @@
 /**
  * Captured from a parts-catalogs.com v3 widget request.
  * Token + the supporting X-* headers the widget sends with every API call.
- * IP-bound (must be reused with the same proxy port that captured it).
+ *
+ * NOT request-time IP-bound. Verified on prod 2026-06-10: 0 auth 401/403 across
+ * thousands of proxied calls, and a token captured on one exit IP replays with a
+ * 200 through arbitrary other IPs (incl. fresh residential ones). Upstream issues
+ * a stable shared token per origin. So the call leg rotates exit IPs freely; only
+ * the capture (Playwright) leg holds a sticky IP for the duration of one capture.
  */
 export interface PcatJwtToken {
   raw: string; // x-api-key value, e.g. "TWS-016EA7BE-..."
@@ -15,10 +20,14 @@ export interface PcatJwtToken {
 
 export interface JwtSlot {
   jwt: PcatJwtToken;
-  proxyPort: number;
   siteUsed: string;
   capturedAt: number;
   refreshTimer: ReturnType<typeof setTimeout> | null;
+  // Consecutive transport failures across requests (reset on any 2xx). The slot
+  // is evicted once this crosses a threshold — a token that keeps failing on
+  // fresh rotating IPs is dead at the source, not a flaky exit IP. Optional so
+  // slots persisted to Redis by older builds still hydrate.
+  failCount?: number;
 }
 
 export interface PcatSession {

apps/api/src/vehicles/vehicles.service.ts

@@ -29,7 +29,7 @@ import {
 import { EmexService } from "../integrations/emex/emex.service";
 import { EmexCandidate } from "../integrations/emex/emex.service";
 import { PartsCatalogsService } from "../integrations/parts-catalogs/parts-catalogs.service";
-import { PcatCar } from "../integrations/parts-catalogs/parts-catalogs.types";
+import { PcatCar, PcatVinResult } from "../integrations/parts-catalogs/parts-catalogs.types";
 import { PL24Service } from "../integrations/pl24/pl24.service";
 import { VinApiService } from "../integrations/vin-api/vin-api.service";
 import { PrefetchSource } from "../jobs/prefetch.types";
@@ -488,6 +488,47 @@ export class VehiclesService {
     await this.redis.del(VehiclesService.PL24_CB_FAILURE_KEY);
   }
 
+  // ─── PCAT circuit breaker ──────────────────────────────
+  // Mirrors the PL24 breaker. parts-catalogs is the #1 winning decode source, so
+  // when its transport degrades (proxy/upstream bad window) it drags the whole
+  // race. After N consecutive TRANSIENT pcat failures we open the circuit briefly
+  // and skip pcat so EMEX/PL24 answer fast. CRITICAL: only transient faults count
+  // — a definitive "no cars" miss must NOT trip it (that's a real, fast answer),
+  // and a skip must mark outcome.transient so a pcat-only VIN isn't negative-cached
+  // for 30m while the circuit self-heals in 30s.
+  private static readonly PCAT_CB_FAILURE_THRESHOLD = 3;
+  private static readonly PCAT_CB_COOLDOWN_MS = 30_000;
+  private static readonly PCAT_CB_COUNTER_TTL_S = 300;
+  private static readonly PCAT_CB_FAILURE_KEY = "pcat:cb:consec_failures";
+  private static readonly PCAT_CB_COOLDOWN_KEY = "pcat:cb:cooldown_until";
+
+  private async isPcatCircuitOpen(): Promise<boolean> {
+    const until = await this.redis.get(VehiclesService.PCAT_CB_COOLDOWN_KEY);
+    return until !== null && Number(until) > Date.now();
+  }
+
+  private async recordPcatFailure(): Promise<void> {
+    const c = await this.redis.incr(VehiclesService.PCAT_CB_FAILURE_KEY);
+    await this.redis.expire(
+      VehiclesService.PCAT_CB_FAILURE_KEY,
+      VehiclesService.PCAT_CB_COUNTER_TTL_S,
+    );
+    if (c >= VehiclesService.PCAT_CB_FAILURE_THRESHOLD) {
+      await this.redis.set(
+        VehiclesService.PCAT_CB_COOLDOWN_KEY,
+        String(Date.now() + VehiclesService.PCAT_CB_COOLDOWN_MS),
+        Math.ceil(VehiclesService.PCAT_CB_COOLDOWN_MS / 1000) + 5,
+      );
+      this.logger.warn(
+        `PCAT circuit opened for ${VehiclesService.PCAT_CB_COOLDOWN_MS}ms (${c} consecutive failures)`,
+      );
+    }
+  }
+
+  private async recordPcatSuccess(): Promise<void> {
+    await this.redis.del(VehiclesService.PCAT_CB_FAILURE_KEY);
+  }
+
   /**
    * Shared VIN decode chain — Corgi → PartsCatalogs → PL24 → EMEX.
    *
@@ -694,13 +735,43 @@ export class VehiclesService {
 
     const pcatStart = Date.now();
     let pcatResolved = false;
-    const pcatPromise = this.partsCatalogsService
-      .decodeVin(vin, signal, outcome)
+    // Circuit breaker: when pcat is in a bad transport window, skip it so the
+    // race isn't dragged; EMEX/PL24 answer instead. A pcat-scoped outcome lets
+    // the breaker count ONLY pcat's transient faults (not another source's).
+    const pcatCircuitOpen = await this.isPcatCircuitOpen();
+    if (ctx && pcatCircuitOpen) ctx.timings.pcat_circuit_open = true;
+    const pcatOutcome: { transient: boolean; coldSkip?: boolean } = { transient: false };
+    // Fast-fail on a cold JWT pool: EMEX (already racing) or PL24 can answer
+    // while the pool warms in the background, instead of the user eating the
+    // multi-second inline capture. The pcat result self-heals on the next decode.
+    const basePcat: Promise<PcatVinResult | null> = pcatCircuitOpen
+      ? (() => {
+          // Skipped (open) — NOT a definitive miss. Mark transient so a
+          // pcat-only VIN isn't negative-cached while the circuit self-heals.
+          this.logger.warn(`PCAT skipped for ${vin}: circuit breaker is open`);
+          pcatOutcome.transient = true;
+          return Promise.resolve(null);
+        })()
+      : this.partsCatalogsService.decodeVin(vin, signal, pcatOutcome, { blockOnEmpty: false });
+    const pcatPromise = basePcat
+      .then(async (r) => {
+        // Breaker bookkeeping (skip when already open — nothing was attempted):
+        // a transient TRANSPORT fault counts toward opening; any definitive
+        // answer (cars OR a clean "no cars" miss) clears the streak. A deliberate
+        // cold-pool skip is neutral — it isn't a fault and the pool is warming.
+        if (!pcatCircuitOpen && !pcatOutcome.coldSkip) {
+          if (pcatOutcome.transient) await this.recordPcatFailure();
+          else await this.recordPcatSuccess();
+        }
+        return r;
+      })
       .catch((err: Error) => {
         this.logger.warn(`PartsCatalogs decode failed for ${vin}: ${err.message}`);
         return null;
       })
       .finally(() => {
+        // Propagate pcat's transient signal to the shared outcome (neg-cache gate).
+        if (pcatOutcome.transient && outcome) outcome.transient = true;
         pcatResolved = true;
         if (ctx && ctx.timings.pcat === undefined) {
           ctx.timings.pcat = Date.now() - pcatStart;