fix(worker-bus): P0 timeout enforcement + P1 memory-tier inference pool sizing

packages/worker-bus/src/workerBus.ts

@@ -262,6 +262,9 @@ export class WorkerBus {
     const worker = await pool.acquire(token.signal);
     const startedAt = performance.now();
     const queueTimeMs = Math.round(startedAt - task.createdAt);
+    // QNBS-v3: [P0 timeout enforcement — runOnPool previously awaited RESULT/abort forever;
+    //          a wedged worker (no crash, no message) left the task's promise pending indefinitely.]
+    let timedOut = false;
 
     try {
       const port = worker.port;
@@ -271,12 +274,55 @@ export class WorkerBus {
 
       const result = await new Promise<TaskResult<TResult>>((resolve, _reject) => {
         let settled = false;
-        const handler = (event: MessageEvent) => {
+        let timeoutTimer: ReturnType<typeof setTimeout>;
+        // QNBS-v3: `handler`/`onTimeout` are forward-declared via `let` (hoisted, no TDZ read)
+        //          because `settle`/`armTimeout` close over them before their real definitions
+        //          run — the three closures are mutually recursive, so a strict top-to-bottom
+        //          `const` chain isn't possible without one forward reference.
+        let handler: (event: MessageEvent) => void;
+        let onTimeout: () => void;
+
+        const settle = () => {
+          settled = true;
+          clearTimeout(timeoutTimer);
+          port.removeEventListener('message', handler);
+        };
+
+        // QNBS-v3: watchdog, not a hard ceiling — re-armed on every PROGRESS message so
+        //          long-running jobs (ProForge stages, LoRA training) that report progress
+        //          aren't falsely killed while genuinely wedged workers still get caught.
+        const armTimeout = () => {
+          clearTimeout(timeoutTimer);
+          timeoutTimer = setTimeout(onTimeout, task.timeoutMs);
+        };
+
+        onTimeout = () => {
+          if (settled) return;
+          settle();
+          timedOut = true;
+          resolve({
+            taskId: task.taskId,
+            success: false,
+            error: {
+              code: 'TIMEOUT',
+              message: `Task exceeded its ${task.timeoutMs}ms deadline with no response from the worker`,
+              recoverable: true,
+              retryCount: 0,
+            },
+            latencyMs: Math.round(performance.now() - startedAt),
+            queueTimeMs,
+            workerId: worker.workerId,
+            layer: 'web',
+          });
+        };
+
+        handler = (event: MessageEvent) => {
           if (settled) return;
           const msg = validateWorkerMessage(event.data);
           if (!msg) return;
 
           if (msg.kind === 'PROGRESS') {
+            armTimeout();
             this.progress.emit(task.taskId, {
               taskId: task.taskId,
               taskType: task.taskType,
@@ -286,8 +332,7 @@ export class WorkerBus {
               timestamp: Date.now(),
             });
           } else if (msg.kind === 'RESULT') {
-            settled = true;
-            port.removeEventListener('message', handler);
+            settle();
             const latencyMs = Math.round(performance.now() - startedAt);
             resolve({
               taskId: task.taskId,
@@ -309,11 +354,11 @@ export class WorkerBus {
           }
         };
         port.addEventListener('message', handler);
+        armTimeout();
 
         const onAbort = () => {
           if (settled) return;
-          settled = true;
-          port.removeEventListener('message', handler);
+          settle();
           port.postMessage(createCancelMessage(task.taskId, 'Aborted'));
           resolve({
             taskId: task.taskId,
@@ -342,7 +387,13 @@ export class WorkerBus {
       }
       return result;
     } finally {
-      pool.release(worker);
+      // QNBS-v3: a timed-out worker is presumed wedged — force-terminate + respawn instead of
+      //          releasing it back to the idle pool where a future task would reuse it.
+      if (timedOut) {
+        pool.terminateWorker(worker.workerId);
+      } else {
+        pool.release(worker);
+      }
     }
   }
 

packages/worker-bus/src/workerPool.ts

@@ -82,6 +82,18 @@ export class WorkerPool {
     this.entries = [];
   }
 
+  /**
+   * Force-terminate one worker and respawn a replacement (pool stays at capacity).
+   * QNBS-v3: a task that missed its timeoutMs deadline may have wedged the worker in an
+   * unknown state — treat it like a crash instead of recycling it back to idle via release().
+   */
+  terminateWorker(workerId: string): void {
+    const entry = this.entries.find((e) => e.instance.workerId === workerId);
+    if (!entry) return;
+    this.setCrashed(workerId);
+    this.restartWorker(entry);
+  }
+
   getHealth(): {
     totalWorkers: number;
     idleWorkers: number;

packages/worker-bus/tests/workerBus.test.ts

@@ -626,4 +626,105 @@ describe('WorkerBus', () => {
     handle.cancel('test-abort');
     await expect(handle.result).rejects.toThrow('cancelled');
   });
+
+  it('times out a hung worker, records a circuit-breaker failure, and force-terminates it', async () => {
+    // Worker never sends PROGRESS or RESULT — simulates a wedged worker (not crashed, just silent).
+    const mockPort = {
+      addEventListener: vi.fn(),
+      removeEventListener: vi.fn(),
+      postMessage: vi.fn(),
+      start: vi.fn(),
+    };
+
+    const mockWorker = {
+      workerId: 'mock-worker-hung',
+      worker: {} as Worker,
+      channel: { port1: mockPort, port2: mockPort } as unknown as MessageChannel,
+      port: mockPort as unknown as MessagePort,
+      state: 'idle' as const,
+      capabilities: ['inference.text'] as const,
+      labels: {},
+    };
+
+    const pool = (bus as unknown as { pools: Map<string, WorkerPool> }).pools.get('fake')!;
+    vi.spyOn(pool, 'acquire').mockResolvedValue(
+      mockWorker as unknown as import('../src/workerPool').PooledWorkerInstance,
+    );
+    const terminateWorkerSpy = vi.spyOn(pool, 'terminateWorker').mockImplementation(() => {});
+    const releaseSpy = vi.spyOn(pool, 'release');
+
+    const handle = bus.enqueue(
+      'test.task',
+      { data: 1 },
+      { timeoutMs: 25, retryPolicy: { maxRetries: 0 } },
+    );
+
+    await expect(handle.result).rejects.toThrow(/exceeded its 25ms deadline/i);
+    expect(terminateWorkerSpy).toHaveBeenCalledWith('mock-worker-hung');
+    expect(releaseSpy).not.toHaveBeenCalled();
+    expect(bus.getTelemetry().failedTasks).toBe(1);
+    expect(bus.getTelemetry().deadLetterCount).toBe(1);
+    expect(bus.getTelemetry().circuitBreakerStates['test.task']).toBeDefined();
+  });
+
+  it('resets the timeout watchdog on PROGRESS so a slow-but-alive task is not killed early', async () => {
+    const mockPort = {
+      addEventListener: vi.fn((type: string, handler: EventListener) => {
+        if (type === 'message') {
+          // Two PROGRESS pings inside the timeout window, then RESULT after the original
+          // deadline would have expired — only survives if PROGRESS re-arms the watchdog.
+          setTimeout(() => {
+            handler(
+              new MessageEvent('message', {
+                data: { kind: 'PROGRESS', taskId: 'mock-task-id', stage: 'step1', progress: 0.3 },
+              }),
+            );
+          }, 15);
+          setTimeout(() => {
+            handler(
+              new MessageEvent('message', {
+                data: { kind: 'PROGRESS', taskId: 'mock-task-id', stage: 'step2', progress: 0.6 },
+              }),
+            );
+          }, 30);
+          setTimeout(() => {
+            handler(
+              new MessageEvent('message', {
+                data: {
+                  kind: 'RESULT',
+                  taskId: 'mock-task-id',
+                  success: true,
+                  result: 'slow-but-done',
+                  latencyMs: 45,
+                },
+              }),
+            );
+          }, 45);
+        }
+      }),
+      removeEventListener: vi.fn(),
+      postMessage: vi.fn(),
+      start: vi.fn(),
+    };
+
+    const mockWorker = {
+      workerId: 'mock-worker-slow',
+      worker: {} as Worker,
+      channel: { port1: mockPort, port2: mockPort } as unknown as MessageChannel,
+      port: mockPort as unknown as MessagePort,
+      state: 'idle' as const,
+      capabilities: ['inference.text'] as const,
+      labels: {},
+    };
+
+    const pool = (bus as unknown as { pools: Map<string, WorkerPool> }).pools.get('fake')!;
+    vi.spyOn(pool, 'acquire').mockResolvedValue(
+      mockWorker as unknown as import('../src/workerPool').PooledWorkerInstance,
+    );
+
+    // timeoutMs (20ms) is shorter than the total run (45ms), but each PROGRESS arrives
+    // well within 20ms of the previous reset, so the watchdog never fires.
+    const handle = bus.enqueue('test.task', { data: 1 }, { timeoutMs: 20 });
+    await expect(handle.result).resolves.toBe('slow-but-done');
+  });
 });

packages/worker-bus/tests/workerPool.test.ts

@@ -204,4 +204,41 @@ describe('WorkerPool', () => {
     pool.release(worker);
     expect(pool.getHealth().totalWorkers).toBe(0);
   });
+
+  it('terminateWorker force-terminates a specific worker and respawns a replacement', async () => {
+    const pool = new WorkerPool('test-pool', ['inference.text'], {
+      maxWorkers: 1,
+      minWorkers: 1,
+      idleTimeoutMs: 120_000,
+      workerScript: '/mock.worker.js',
+      capabilities: ['inference.text'],
+      labels: {},
+    });
+    const worker = await pool.acquire();
+    const mockWorker = worker.worker as unknown as MockWorker;
+
+    pool.terminateWorker(worker.workerId);
+
+    expect(mockWorker.terminated).toBe(true);
+    // QNBS-v3: like restartWorker() on crash — the pool respawns to stay at capacity,
+    //          but the new worker must be a distinct instance, not the wedged one.
+    expect(pool.getHealth().totalWorkers).toBe(1);
+    expect(pool.getHealth().crashedWorkers).toBe(0);
+    await pool.terminateAll();
+  });
+
+  it('terminateWorker is a no-op for an unknown workerId', async () => {
+    const pool = new WorkerPool('test-pool', ['inference.text'], {
+      maxWorkers: 1,
+      minWorkers: 1,
+      idleTimeoutMs: 120_000,
+      workerScript: '/mock.worker.js',
+      capabilities: ['inference.text'],
+      labels: {},
+    });
+    await pool.acquire();
+    expect(() => pool.terminateWorker('nonexistent-worker-id')).not.toThrow();
+    expect(pool.getHealth().totalWorkers).toBe(1);
+    await pool.terminateAll();
+  });
 });

services/ai/localAiDeviceProfiler.ts

@@ -172,7 +172,10 @@ function detectDirectML(
 // Memory Tier
 // ---------------------------------------------------------------------------
 
-function detectMemoryTier(): DeviceCapabilityProfile['memoryTier'] {
+// QNBS-v3: exported (previously module-private) so callers that only need a fast, synchronous
+//          memory-tier read — e.g. workerBusManager sizing the inference pool at init — don't
+//          have to pay for the full async generateDeviceProfile() (WebGPU/WebNN/battery probes).
+export function detectMemoryTier(): DeviceCapabilityProfile['memoryTier'] {
   const deviceMemory =
     typeof navigator !== 'undefined' && 'deviceMemory' in navigator
       ? (navigator as Navigator & { deviceMemory?: number }).deviceMemory

services/workerBusManager.ts

@@ -72,9 +72,12 @@ async function inferencePoolOptions() {
   const { MAX_WORKERS_INFERENCE, MIN_WORKERS, WORKER_IDLE_TIMEOUT_MS } = await import(
     '@domain/worker-bus'
   );
+  // QNBS-v3: [P1 — scale inference worker count to the device's memory tier instead of a fixed
+  //          cap. Each replica loads its own transformers.js pipeline (no cross-replica cache
+  //          sharing), so more replicas only help on devices with RAM headroom to spare.]
+  const maxWorkers = Math.min(await resolveInferenceMaxWorkers(), MAX_WORKERS_INFERENCE);
   return {
-    // QNBS-v3: [Capped below MAX_WORKERS_INFERENCE — each replica loads its own transformers.js pipeline (no cross-replica cache sharing), so 4 concurrent workers could mean 4x the model memory footprint under a burst.]
-    maxWorkers: Math.min(2, MAX_WORKERS_INFERENCE),
+    maxWorkers,
     minWorkers: MIN_WORKERS,
     idleTimeoutMs: WORKER_IDLE_TIMEOUT_MS,
     workerScript: new URL('../workers/v2/inference.worker.ts', import.meta.url).href,
@@ -83,6 +86,24 @@ async function inferencePoolOptions() {
   };
 }
 
+/**
+ * QNBS-v3: memory-tier-driven worker count — high:3, medium:2 (previous hardcoded default),
+ * low:1. Falls back to the previous default (2) if device profiling throws (e.g. non-browser
+ * test environment) so this can never block or fail pool initialization.
+ */
+async function resolveInferenceMaxWorkers(): Promise<number> {
+  try {
+    const { detectMemoryTier } = await import('./ai/localAiDeviceProfiler');
+    const tier = detectMemoryTier();
+    if (tier === 'high') return 3;
+    if (tier === 'low') return 1;
+    return 2;
+  } catch (err) {
+    log.warn('Failed to detect memory tier for inference pool sizing; using default', err);
+    return 2;
+  }
+}
+
 /** Re-register the 'inference' pool if it was removed via terminatePool() — a no-op if already present. */
 async function reRegisterInferencePool(bus: WorkerBus): Promise<void> {
   if (bus.hasPool('inference')) return;

tests/unit/workerBusManager.test.ts

@@ -295,4 +295,70 @@ describe('workerBusManager', () => {
       );
     });
   });
+
+  describe('inference pool sizing by memory tier', () => {
+    // QNBS-v3: [P1 — inference pool maxWorkers is now derived from
+    //          localAiDeviceProfiler.detectMemoryTier() instead of a hardcoded 2. Verified via
+    //          the same re-registration path as the "memory-safety cap" test above, since
+    //          initWorkerBus() only calls registry.register() (mocked, no-op) — bus.registerPool()
+    //          with the real computed options is only exercised by ensureInferencePool().]
+    afterEach(() => {
+      vi.doUnmock('../../services/ai/localAiDeviceProfiler');
+    });
+
+    async function expectMaxWorkersForTier(
+      detectMemoryTier: () => 'high' | 'medium' | 'low',
+      expectedMaxWorkers: number,
+    ): Promise<void> {
+      vi.doMock('../../services/ai/localAiDeviceProfiler', () => ({ detectMemoryTier }));
+      const { initWorkerBus, ensureInferencePool } = await import(
+        '../../services/workerBusManager'
+      );
+      await initWorkerBus();
+      mockRegisterPool.mockClear();
+      mockHasPool.mockReturnValue(false);
+
+      await ensureInferencePool();
+
+      expect(mockRegisterPool).toHaveBeenCalledWith(
+        'inference',
+        expect.arrayContaining(['inference.text', 'inference.embed']),
+        expect.objectContaining({ maxWorkers: expectedMaxWorkers }),
+      );
+    }
+
+    it('sizes maxWorkers to 3 on a high memory tier', async () => {
+      await expectMaxWorkersForTier(() => 'high', 3);
+    });
+
+    it('sizes maxWorkers to 2 on a medium memory tier (unchanged default)', async () => {
+      await expectMaxWorkersForTier(() => 'medium', 2);
+    });
+
+    it('sizes maxWorkers to 1 on a low memory tier', async () => {
+      await expectMaxWorkersForTier(() => 'low', 1);
+    });
+
+    it('falls back to maxWorkers 2 when memory-tier detection throws', async () => {
+      vi.doMock('../../services/ai/localAiDeviceProfiler', () => ({
+        detectMemoryTier: () => {
+          throw new Error('profiler boom');
+        },
+      }));
+      const { initWorkerBus, ensureInferencePool } = await import(
+        '../../services/workerBusManager'
+      );
+      await initWorkerBus();
+      mockRegisterPool.mockClear();
+      mockHasPool.mockReturnValue(false);
+
+      await ensureInferencePool();
+
+      expect(mockRegisterPool).toHaveBeenCalledWith(
+        'inference',
+        expect.arrayContaining(['inference.text', 'inference.embed']),
+        expect.objectContaining({ maxWorkers: 2 }),
+      );
+    });
+  });
 });