fix(pollinations/duckduckgo): wire session pool for fingerprint rotation (#3049)

* fix(pollinations): wire session pool + add idle pruning

PollinationsExecutor.getPool() returned null because poolConfig was
never set. Now sets DEFAULT_POOL_CONFIG in constructor so anonymous
requests get fingerprint rotation and 429 cooldown management.

Also:
- Use acquireBlocking() instead of acquire() to wait for available session
- Add startAutoPrune() for periodic idle session cleanup (5min idle timeout)
- Improve execute() error handling with proper finally block

* fix(duckduckgo-web): wire session pool for fingerprint rotation

DuckDuckGoWebExecutor had poolConfig set but never called getPool()
in execute(). Added session acquisition via acquireBlocking() and
merges fingerprint headers into fetch calls for rate limit evasion.

* fix: address PR #3049 review comments

- duckduckgo-web: fix session leak (add finally release), fix race
  condition (report status before release), remove duplicate sessionHeaders
- pollinations: fix race condition (move release to finally, report before)

* fix: address all PR #3049 review comments

- duckduckgo-web: add pool reporting on 429/500 early returns (was missing)
- duckduckgo-web: add sessionHeaders to retry fetch on 401/403
- sessionPool: add .unref() to setInterval to prevent keeping process alive

---------

Co-authored-by: oyi77 <oyi77@users.noreply.github.com>

open-sse/executors/duckduckgo-web.ts

@@ -73,6 +73,16 @@ export class DuckDuckGoWebExecutor extends BaseExecutor {
       );
     }
 
+    // Acquire session from pool for fingerprint rotation
+    const pool = this.getPool();
+    let session;
+    try {
+      session = pool ? await pool.acquireBlocking(10_000) : null;
+    } catch {
+      session = null;
+    }
+    const sessionHeaders = session ? session.buildHeaders() : {};
+
     try {
       const controller = new AbortController();
       const timeout = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS);
@@ -101,6 +111,7 @@ export class DuckDuckGoWebExecutor extends BaseExecutor {
         method: "POST",
         headers: {
           ...FAKE_HEADERS,
+          ...sessionHeaders,
           "Content-Type": "application/json",
           "x-vqd-hash-1": vqdToken,
         },
@@ -115,6 +126,7 @@ export class DuckDuckGoWebExecutor extends BaseExecutor {
       clearTimeout(timeout);
 
       if (chatResponse.status === 429) {
+        if (pool && session) pool.reportCooldown(session);
         return new Response(
           JSON.stringify({ error: { message: "DuckDuckGo rate limited" } }),
           { status: 429, headers: { "Content-Type": "application/json" } }
@@ -148,14 +160,32 @@ export class DuckDuckGoWebExecutor extends BaseExecutor {
       }
 
       if (chatResponse.status >= 500) {
+        if (pool && session) pool.reportDead(session);
         return new Response(
           JSON.stringify({ error: { message: "Upstream error" } }),
           { status: 502, headers: { "Content-Type": "application/json" } }
         );
       }
 
-      return this.processResponse(chatResponse, stream !== false);
+      const result = this.processResponse(chatResponse, stream !== false);
+
+      // Report pool status based on response
+      if (pool && session) {
+        if (chatResponse.status === 429) {
+          pool.reportCooldown(session);
+        } else if (chatResponse.status >= 500) {
+          pool.reportDead(session);
+        } else {
+          pool.reportSuccess(session);
+        }
+      }
+
+      return result;
     } catch (error) {
+      if (pool && session) {
+        pool.reportCooldown(session);
+      }
+
       if (error instanceof DOMException && error.name === "AbortError") {
         return new Response(
           JSON.stringify({ error: { message: "Request cancelled" } }),
@@ -167,6 +197,8 @@ export class DuckDuckGoWebExecutor extends BaseExecutor {
         JSON.stringify({ error: { message: error instanceof Error ? error.message : "Unknown error" } }),
         { status: 500, headers: { "Content-Type": "application/json" } }
       );
+    } finally {
+      session?.release();
     }
   }
 

open-sse/executors/pollinations.ts

@@ -1,11 +1,12 @@
 import { BaseExecutor } from "./base.ts";
 import { PROVIDERS } from "../config/constants.ts";
-import { SessionPool, PoolRegistry } from "../services/sessionPool/index.ts";
+import { DEFAULT_POOL_CONFIG } from "../services/sessionPool/types.ts";
 import type { ExecuteInput } from "./base.ts";
 
 export class PollinationsExecutor extends BaseExecutor {
   constructor() {
     super("pollinations", PROVIDERS["pollinations"] || { format: "openai" });
+    this.poolConfig = DEFAULT_POOL_CONFIG;
   }
 
   buildUrl(_model: string, _stream: boolean, urlIndex = 0, _credentials = null): string {
@@ -50,7 +51,15 @@ export class PollinationsExecutor extends BaseExecutor {
     }
 
     const pool = this.getPool();
-    const session = pool ? pool.acquire() : null;
+
+    // Use acquireBlocking for anonymous requests to wait for available session
+    let session;
+    try {
+      session = pool ? await pool.acquireBlocking(10_000) : null;
+    } catch {
+      // Pool exhausted — fall through to direct request without fingerprint
+      session = null;
+    }
 
     if (session) {
       const fpHeaders = session.buildHeaders();
@@ -60,19 +69,10 @@ export class PollinationsExecutor extends BaseExecutor {
       };
     }
 
-    let result;
     try {
-      result = await super.execute(input);
-    } catch (err) {
-      if (session && pool) {
-        pool.reportCooldown(session);
-        session.release();
-      }
-      throw err;
-    }
+      const result = await super.execute(input);
 
-    if (session && pool) {
-      try {
+      if (session && pool) {
         const status = result.response.status;
         if (status === 429) {
           pool.reportCooldown(session);
@@ -81,12 +81,17 @@ export class PollinationsExecutor extends BaseExecutor {
         } else {
           pool.reportSuccess(session);
         }
-      } finally {
-        session.release();
       }
-    }
 
-    return result;
+      return result;
+    } catch (err) {
+      if (session && pool) {
+        pool.reportCooldown(session);
+      }
+      throw err;
+    } finally {
+      session?.release();
+    }
   }
 }
 

open-sse/services/sessionPool/sessionPool.ts

@@ -300,14 +300,27 @@ export class SessionPool {
     }
   }
 
-  /** Remove dead sessions (call periodically for reclamation */
-  pruneDeadSessions(): void {
+  /** Remove dead sessions and idle sessions older than maxIdleMs */
+  pruneDeadSessions(maxIdleMs = 300_000): void {
+    const now = Date.now();
     const before = this.sessions.length;
-    this.sessions = this.sessions.filter((s) => s.status !== "dead");
+    this.sessions = this.sessions.filter((s) => {
+      if (s.status === "dead") return false;
+      // Prune idle sessions older than maxIdleMs (default 5min)
+      if (s.inflight === 0 && s.lastUsedAt > 0 && now - s.lastUsedAt > maxIdleMs) return false;
+      return true;
+    });
 
-    // If we pruned sessions, report
+    // If we pruned sessions, ensure minimum
     if (this.sessions.length < before && this.sessions.length < this.config.minSessions) {
       this.ensureMinSessions();
     }
   }
+
+  /** Start periodic pruning (every 60s) */
+  startAutoPrune(intervalMs = 60_000): ReturnType<typeof setInterval> {
+    const timer = setInterval(() => this.pruneDeadSessions(), intervalMs);
+    timer.unref();
+    return timer;
+  }
 }