fix: restore openai-compatible proxy support after 0.7.0 regressions

apps/webapp/app/bullmq/start-workers.ts

@@ -138,26 +138,30 @@ export async function initWorkers(): Promise<void> {
   // Log worker startup
   logger.log("\n🚀 Starting BullMQ workers...");
   logger.log("─".repeat(80));
-  logger.log(`✓ Ingest worker: ${ingestWorker.name} (concurrency: 1)`);
   logger.log(
-    `✓ Document ingest worker: ${preprocessWorker.name} (concurrency: 3)`,
+    `✓ Ingest worker: ${ingestWorker.name} (concurrency: ${ingestWorker.opts.concurrency ?? 1})`,
   );
   logger.log(
-    `✓ Conversation title worker: ${conversationTitleWorker.name} (concurrency: 10)`,
+    `✓ Document ingest worker: ${preprocessWorker.name} (concurrency: ${preprocessWorker.opts.concurrency ?? 1})`,
   );
   logger.log(
-    `✓ Session compaction worker: ${sessionCompactionWorker.name} (concurrency: 3)`,
+    `✓ Conversation title worker: ${conversationTitleWorker.name} (concurrency: ${conversationTitleWorker.opts.concurrency ?? 1})`,
   );
   logger.log(
-    `✓ Label assignment worker: ${labelAssignmentWorker.name} (concurrency: 5)`,
+    `✓ Session compaction worker: ${sessionCompactionWorker.name} (concurrency: ${sessionCompactionWorker.opts.concurrency ?? 1})`,
   );
   logger.log(
-    `✓ Title generation worker: ${titleGenerationWorker.name} (concurrency: 10)`,
+    `✓ Label assignment worker: ${labelAssignmentWorker.name} (concurrency: ${labelAssignmentWorker.opts.concurrency ?? 1})`,
   );
   logger.log(
-    `✓ Integration run worker: ${integrationRunWorker.name} (concurrency: 3)`,
+    `✓ Title generation worker: ${titleGenerationWorker.name} (concurrency: ${titleGenerationWorker.opts.concurrency ?? 1})`,
+  );
+  logger.log(
+    `✓ Integration run worker: ${integrationRunWorker.name} (concurrency: ${integrationRunWorker.opts.concurrency ?? 1})`,
+  );
+  logger.log(
+    `✓ Scratchpad scan worker: ${scratchpadScanWorker.name} (concurrency: ${scratchpadScanWorker.opts.concurrency ?? 1})`,
   );
-  logger.log(`✓ Scratchpad scan worker: ${scratchpadScanWorker.name} (concurrency: 5)`);
   logger.log(`✓ Reminder scheduler: reminder-queue + followup-queue`);
   logger.log(`✓ Scheduled task scheduler: scheduled-task-queue`);
   logger.log("─".repeat(80));

apps/webapp/app/bullmq/workers/index.ts

@@ -37,6 +37,7 @@ import {
   enqueueGraphResolution,
 } from "~/lib/queue-adapter.server";
 import { logger } from "~/services/logger.service";
+import { getBurstSafeBullmqConcurrency } from "~/services/llm-provider.server";
 import {
   type PersonaGenerationPayload,
   processPersonaGeneration,
@@ -106,7 +107,8 @@ export const preprocessWorker = new Worker(
       // Callback to enqueue individual chunk ingestion jobs
       enqueueIngestEpisode,
       // Callback to enqueue session compaction for conversations
-      enqueueSessionCompaction,
+      (compactionPayload, delayMs) =>
+        enqueueSessionCompaction(compactionPayload, delayMs),
     );
     if (!result?.success) {
       throw new Error(result?.error || "Episode preprocessing failed");
@@ -147,7 +149,10 @@ export const ingestWorker = new Worker(
   },
   {
     connection: getRedisConnection(),
-    concurrency: env.BULLMQ_CONCURRENCY_INGEST, // Global limit for ingestion jobs
+    concurrency: getBurstSafeBullmqConcurrency(
+      "BULLMQ_CONCURRENCY_INGEST",
+      env.BULLMQ_CONCURRENCY_INGEST,
+    ), // Keep proxy/self-hosted chat from being dogpiled by background ingestion unless explicitly overridden
   },
 );
 
@@ -162,7 +167,10 @@ export const conversationTitleWorker = new Worker(
   },
   {
     connection: getRedisConnection(),
-    concurrency: env.BULLMQ_CONCURRENCY_CONVERSATION_TITLE, // Process title creations in parallel
+    concurrency: getBurstSafeBullmqConcurrency(
+      "BULLMQ_CONCURRENCY_CONVERSATION_TITLE",
+      env.BULLMQ_CONCURRENCY_CONVERSATION_TITLE,
+    ), // Strict providers benefit from serial title generation unless explicitly overridden
   },
 );
 
@@ -177,7 +185,10 @@ export const sessionCompactionWorker = new Worker(
   },
   {
     connection: getRedisConnection(),
-    concurrency: env.BULLMQ_CONCURRENCY_SESSION_COMPACTION, // Process compactions in parallel
+    concurrency: getBurstSafeBullmqConcurrency(
+      "BULLMQ_CONCURRENCY_SESSION_COMPACTION",
+      env.BULLMQ_CONCURRENCY_SESSION_COMPACTION,
+    ), // Compaction is background polish; keep it from competing with chat unless explicitly overridden
   },
 );
 
@@ -193,7 +204,10 @@ export const labelAssignmentWorker = new Worker(
   },
   {
     connection: getRedisConnection(),
-    concurrency: env.BULLMQ_CONCURRENCY_LABEL_ASSIGNMENT, // Process label assignments in parallel
+    concurrency: getBurstSafeBullmqConcurrency(
+      "BULLMQ_CONCURRENCY_LABEL_ASSIGNMENT",
+      env.BULLMQ_CONCURRENCY_LABEL_ASSIGNMENT,
+    ), // Labels are non-critical background work for proxy mode unless explicitly overridden
   },
 );
 
@@ -209,7 +223,10 @@ export const titleGenerationWorker = new Worker(
   },
   {
     connection: getRedisConnection(),
-    concurrency: env.BULLMQ_CONCURRENCY_TITLE_GENERATION, // Process title generations in parallel
+    concurrency: getBurstSafeBullmqConcurrency(
+      "BULLMQ_CONCURRENCY_TITLE_GENERATION",
+      env.BULLMQ_CONCURRENCY_TITLE_GENERATION,
+    ), // Keep follow-up title generation from stampeding strict providers unless explicitly overridden
   },
 );
 
@@ -225,7 +242,10 @@ export const personaGenerationWorker = new Worker(
   },
   {
     connection: getRedisConnection(),
-    concurrency: env.BULLMQ_CONCURRENCY_PERSONA_GENERATION, // Persona is CPU-intensive
+    concurrency: getBurstSafeBullmqConcurrency(
+      "BULLMQ_CONCURRENCY_PERSONA_GENERATION",
+      env.BULLMQ_CONCURRENCY_PERSONA_GENERATION,
+    ), // Serialize expensive background LLM work for burst-sensitive setups unless explicitly overridden
   },
 );
 
@@ -241,7 +261,10 @@ export const graphResolutionWorker = new Worker(
   },
   {
     connection: getRedisConnection(),
-    concurrency: env.BULLMQ_CONCURRENCY_GRAPH_RESOLUTION, // Graph resolution concurrency
+    concurrency: getBurstSafeBullmqConcurrency(
+      "BULLMQ_CONCURRENCY_GRAPH_RESOLUTION",
+      env.BULLMQ_CONCURRENCY_GRAPH_RESOLUTION,
+    ), // Proxy mode needs stricter background pacing unless explicitly overridden
   },
 );
 

apps/webapp/app/components/chat-panel/global-chat-panel.client.tsx

@@ -25,6 +25,7 @@ interface GlobalChatPanelProps {
   onClose: () => void;
   models: LLMModel[];
   integrationAccountMap: Record<string, string>;
+  enableBurstSafeReplyRecovery?: boolean;
 }
 
 // Minimal chat input — creates a conversation and hands off to ConversationView
@@ -186,12 +187,14 @@ export function GlobalChatPanel({
   onClose,
   models,
   integrationAccountMap,
+  enableBurstSafeReplyRecovery = false,
 }: GlobalChatPanelProps) {
   const { pinnedConversationId } = useChatPanel()!;
   const navigate = useNavigate();
 
   const [activeConversation, setActiveConversation] = useState<{
     conversationId: string;
+    status?: string;
     history: Array<{
       id: string;
       userType: string;
@@ -231,6 +234,7 @@ export function GlobalChatPanel({
   ) => {
     setActiveConversation({
       conversationId,
+      status: "pending",
       history: [
         {
           id: historyId,
@@ -256,6 +260,7 @@ export function GlobalChatPanel({
       const conv = historyFetcher.data.conversation;
       setActiveConversation({
         conversationId: pendingHistoryId.current,
+        status: conv.status,
         history: conv.ConversationHistory ?? [],
       });
       pendingHistoryId.current = null;
@@ -314,7 +319,9 @@ export function GlobalChatPanel({
             conversationId={activeConversation.conversationId}
             history={activeConversation.history}
             autoRegenerate
+            enableBurstSafeFirstReplyRecovery={enableBurstSafeReplyRecovery}
             integrationAccountMap={integrationAccountMap}
+            conversationStatus={activeConversation.status}
             models={models}
           />
         </div>