queue/docs/DESKTOP_CHAT_APP_INTEGRATION.md

Desktop Chat App - @queue Integration Guide

Integration guide for adding BullMQ-based job queues to the desktop-chat-app Electron application.

Overview

The desktop-chat-app is an Electron + React application with several ML-intensive operations that would benefit from queue-based processing:

Feature	Current Implementation	Queue Benefit
Voice/TTS	Sync IPC, 60s timeout	Batch TTS, voice cloning prioritization
Image Generation	HTTP, 300s timeout	Batch requests, GPU load management
Indexing Service	In-process with Redis	Distributed processing for large codebases
Agent Client	SSE streaming	Retry logic, rate limiting

Existing Infrastructure: Redis is already running on port 41224, making BullMQ integration straightforward.

---

1. Setup for Electron Main Process

Unlike NestJS applications, Electron main processes use BullMQ directly. Install the required packages:

pnpm add bullmq @lilith/queue/core @lilith/queue/ml

Queue Manager Service

Create a centralized queue manager for the Electron main process:

// src/main/services/queue-manager.ts

import { Queue, Worker, QueueEvents, type ConnectionOptions } from 'bullmq';
import {
  JobPriority,
  generateCorrelationId,
  isPeakHour,
  shouldDeferJob,
  type BaseJobData,
  type QueueMetrics,
} from '@lilith/queue/core';

/** Redis connection for the existing instance */
const REDIS_CONNECTION: ConnectionOptions = {
  host: 'localhost',
  port: 41224,
  maxRetriesPerRequest: null, // Required for BullMQ workers
};

/** Active queues registry */
const queues = new Map<string, Queue>();
const workers = new Map<string, Worker>();
const queueEvents = new Map<string, QueueEvents>();

/**
 * Get or create a queue
 */
export function getQueue<TData extends BaseJobData>(
  name: string,
  defaultJobOptions?: Parameters<Queue['add']>[2]
): Queue<TData> {
  if (!queues.has(name)) {
    const queue = new Queue<TData>(name, {
      connection: REDIS_CONNECTION,
      defaultJobOptions: {
        attempts: 3,
        backoff: { type: 'exponential', delay: 1000 },
        removeOnComplete: { count: 100, age: 86400 },
        removeOnFail: { count: 500, age: 604800 },
        ...defaultJobOptions,
      },
    });

    queues.set(name, queue);
  }

  return queues.get(name) as Queue<TData>;
}

/**
 * Register a worker for a queue
 */
export function registerWorker<TData extends BaseJobData, TResult>(
  queueName: string,
  processor: (job: { data: TData; id?: string; name: string }) => Promise<TResult>,
  options?: { concurrency?: number }
): Worker<TData, TResult> {
  const worker = new Worker<TData, TResult>(
    queueName,
    async (job) => processor(job),
    {
      connection: REDIS_CONNECTION,
      concurrency: options?.concurrency ?? 1,
    }
  );

  workers.set(queueName, worker);
  return worker;
}

/**
 * Get queue events for monitoring
 */
export function getQueueEvents(queueName: string): QueueEvents {
  if (!queueEvents.has(queueName)) {
    const events = new QueueEvents(queueName, { connection: REDIS_CONNECTION });
    queueEvents.set(queueName, events);
  }

  return queueEvents.get(queueName)!;
}

/**
 * Add a job with priority and peak-hour awareness
 */
export async function addJob<TData extends BaseJobData>(
  queueName: string,
  jobName: string,
  data: Omit<TData, 'createdAt' | 'correlationId'> & { correlationId?: string },
  options?: {
    priority?: JobPriority;
    delay?: number;
    jobId?: string;
  }
): Promise<string> {
  const queue = getQueue<TData>(queueName);
  const priority = options?.priority ?? JobPriority.NORMAL;

  // Check peak-hour deferral
  const { shouldDefer, delay: peakDelay } = shouldDeferJob(priority);
  const finalDelay = shouldDefer ? Math.max(options?.delay ?? 0, peakDelay) : (options?.delay ?? 0);

  const jobData = {
    ...data,
    createdAt: Date.now(),
    correlationId: data.correlationId ?? generateCorrelationId(queueName),
  } as TData;

  const job = await queue.add(jobName, jobData, {
    priority,
    delay: finalDelay,
    jobId: options?.jobId,
  });

  return job.id!;
}

/**
 * Get queue metrics for IPC status reporting
 */
export async function getQueueMetrics(queueName: string): Promise<QueueMetrics> {
  const queue = getQueue(queueName);
  const counts = await queue.getJobCounts();
  const isPaused = await queue.isPaused();

  return {
    name: queueName,
    counts: {
      waiting: counts.waiting ?? 0,
      active: counts.active ?? 0,
      completed: counts.completed ?? 0,
      failed: counts.failed ?? 0,
      delayed: counts.delayed ?? 0,
      paused: counts.paused ?? 0,
    },
    isPaused,
    timestamp: Date.now(),
  };
}

/**
 * Get all queue summaries
 */
export async function getAllQueueMetrics(): Promise<QueueMetrics[]> {
  const metrics: QueueMetrics[] = [];

  for (const [name] of queues) {
    metrics.push(await getQueueMetrics(name));
  }

  return metrics;
}

/**
 * Graceful shutdown
 */
export async function shutdownQueues(): Promise<void> {
  // Close workers first
  for (const [name, worker] of workers) {
    console.log(`[Queue] Closing worker: ${name}`);
    await worker.close();
  }

  // Close queue events
  for (const [name, events] of queueEvents) {
    console.log(`[Queue] Closing events: ${name}`);
    await events.close();
  }

  // Close queues
  for (const [name, queue] of queues) {
    console.log(`[Queue] Closing queue: ${name}`);
    await queue.close();
  }

  queues.clear();
  workers.clear();
  queueEvents.clear();
}

Lifecycle Integration

Register queue shutdown in the Electron app lifecycle:

// src/main/index.ts (or wherever app lifecycle is managed)

import { app } from 'electron';
import { shutdownQueues } from './services/queue-manager.js';

app.on('before-quit', async (event) => {
  event.preventDefault();

  console.log('[App] Shutting down queues...');
  await shutdownQueues();

  app.exit(0);
});

---

2. Voice Synthesis Queue with Priorities

The current voice-handlers.ts uses synchronous IPC with a 60-second timeout. With queuing, we can:

1. Batch multiple TTS requests efficiently
2. Prioritize voice cloning over regular TTS
3. Handle long-running requests without blocking

Job Types and Data

// src/main/services/queues/voice-queue.ts

import { Queue, Worker } from 'bullmq';
import {
  JobPriority,
  type BaseJobData,
  DEFAULT_ML_TIMEOUT,
} from '@lilith/queue/core';
import { RequestBatchingStrategy } from '@lilith/queue/ml';
import { BrowserWindow } from 'electron';

import { getQueue, registerWorker, addJob } from '../queue-manager.js';

/** Voice synthesis job data */
export interface VoiceSynthesizeJobData extends BaseJobData {
  provider: 'chatterbox' | 'piper';
  endpoint: string;
  text: string;
  // Chatterbox-specific
  voiceId?: string | null;
  exaggeration?: number;
  cfgWeight?: number;
  // Piper-specific
  voice?: string;
  speed?: number;
  // Priority indicator
  isVoiceCloning?: boolean;
}

/** Voice synthesis result */
export interface VoiceSynthesizeResult {
  success: boolean;
  audioBase64?: string;
  format?: string;
  error?: string;
  cacheKey?: string;
}

const VOICE_QUEUE_NAME = 'voice-synthesis';

/** TTS Request batching for efficient processing */
const ttsBatcher = new RequestBatchingStrategy<
  VoiceSynthesizeJobData,
  VoiceSynthesizeResult
>({
  maxBatchSize: 10,       // Batch up to 10 requests
  maxWaitMs: 200,         // Wait max 200ms to collect batch
  processBatch: async (requests) => {
    // Process each request individually but in parallel
    // (TTS services typically don't support true batching)
    return Promise.all(requests.map(synthesizeSingle));
  },
  getKey: (req) => `${req.provider}:${req.text}:${req.voiceId ?? 'default'}`,
  onBatchProcessed: (size, durationMs) => {
    console.log(`[Voice] Batch of ${size} processed in ${durationMs}ms`);
  },
});

/**
 * Synthesize a single TTS request
 */
async function synthesizeSingle(
  data: VoiceSynthesizeJobData
): Promise<VoiceSynthesizeResult> {
  const controller = new AbortController();
  const timeoutId = setTimeout(() => controller.abort(), DEFAULT_ML_TIMEOUT);

  try {
    let url: string;
    let body: Record<string, unknown>;

    if (data.provider === 'chatterbox') {
      url = `${data.endpoint}/synthesize`;
      body = {
        text: data.text,
        voice_id: data.voiceId ?? undefined,
        exaggeration: data.exaggeration,
        cfg_weight: data.cfgWeight,
      };
    } else {
      url = `${data.endpoint}/api/tts/synthesize`;
      body = {
        text: data.text,
        voice: data.voice,
        speed: data.speed,
      };
    }

    const response = await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(body),
      signal: controller.signal,
    });

    clearTimeout(timeoutId);

    if (!response.ok) {
      return { success: false, error: `HTTP ${response.status}` };
    }

    if (data.provider === 'chatterbox') {
      const json = await response.json() as { audio_base64: string; format?: string };
      return {
        success: true,
        audioBase64: json.audio_base64,
        format: json.format ?? 'wav',
      };
    } else {
      const buffer = await response.arrayBuffer();
      return {
        success: true,
        audioBase64: Buffer.from(buffer).toString('base64'),
        format: 'wav',
      };
    }
  } catch (error) {
    clearTimeout(timeoutId);
    return {
      success: false,
      error: error instanceof Error ? error.message : 'Unknown error',
    };
  }
}

/**
 * Initialize the voice synthesis queue and worker
 */
export function initializeVoiceQueue(): void {
  // Create the queue
  getQueue<VoiceSynthesizeJobData>(VOICE_QUEUE_NAME, {
    attempts: 2,
    backoff: { type: 'fixed', delay: 500 },
  });

  // Register worker with concurrency matching GPU capacity
  registerWorker<VoiceSynthesizeJobData, VoiceSynthesizeResult>(
    VOICE_QUEUE_NAME,
    async (job) => {
      // Use batcher for efficient processing
      return ttsBatcher.add(job.data);
    },
    { concurrency: 3 } // Process up to 3 jobs in parallel
  );

  console.log('[Voice Queue] Initialized');
}

/**
 * Queue a voice synthesis job
 *
 * @param data - Synthesis parameters
 * @returns Promise resolving to job ID
 */
export async function queueVoiceSynthesis(
  data: Omit<VoiceSynthesizeJobData, 'createdAt' | 'correlationId'>,
  options?: { priority?: JobPriority }
): Promise<string> {
  // Voice cloning gets HIGH priority by default
  const priority = options?.priority ??
    (data.isVoiceCloning ? JobPriority.HIGH : JobPriority.NORMAL);

  return addJob<VoiceSynthesizeJobData>(
    VOICE_QUEUE_NAME,
    data.isVoiceCloning ? 'voice-clone' : 'synthesize',
    data,
    { priority }
  );
}

/**
 * Queue voice synthesis and wait for result
 * (For IPC handlers that need synchronous-style responses)
 */
export async function synthesizeAndWait(
  data: Omit<VoiceSynthesizeJobData, 'createdAt' | 'correlationId'>
): Promise<VoiceSynthesizeResult> {
  const queue = getQueue<VoiceSynthesizeJobData>(VOICE_QUEUE_NAME);

  const priority = data.isVoiceCloning ? JobPriority.HIGH : JobPriority.NORMAL;

  const job = await queue.add(
    data.isVoiceCloning ? 'voice-clone' : 'synthesize',
    {
      ...data,
      createdAt: Date.now(),
      correlationId: `voice-${Date.now()}`,
    },
    { priority }
  );

  // Wait for completion with timeout
  const result = await job.waitUntilFinished(
    (await import('../queue-manager.js')).getQueueEvents(VOICE_QUEUE_NAME),
    DEFAULT_ML_TIMEOUT
  );

  return result;
}

/**
 * Broadcast voice queue status to renderer
 */
export async function broadcastVoiceQueueStatus(): Promise<void> {
  const metrics = await (await import('../queue-manager.js')).getQueueMetrics(VOICE_QUEUE_NAME);
  const batcherStats = ttsBatcher.getStats();

  for (const window of BrowserWindow.getAllWindows()) {
    window.webContents.send('queue:voice-status', {
      ...metrics,
      batcher: batcherStats,
    });
  }
}

Updated IPC Handler

// src/main/ipc/voice-handlers.ts (updated synthesize handler)

import { ipcMain } from 'electron';
import { synthesizeAndWait, queueVoiceSynthesis } from '../services/queues/voice-queue.js';
import { ttsCache } from '../services/tts-cache.js';

// ... existing handlers ...

// Replace the 'voice:synthesize' handler with queue-backed version
ipcMain.handle(
  'voice:synthesize',
  async (_event, request: TTSSynthesizeRequest): Promise<TTSSynthesizeResponse> => {
    try {
      // Check cache first (keep existing cache logic)
      await ttsCache.initialize();
      const cacheKey = ttsCache.getCacheKey(request);
      const cached = await ttsCache.get(cacheKey);

      if (cached) {
        return {
          success: true,
          audioBase64: cached.audioBase64,
          format: cached.format,
        };
      }

      // Use queue for processing
      const result = await synthesizeAndWait({
        provider: request.provider,
        endpoint: request.endpoint,
        text: request.text,
        voiceId: request.voiceId,
        exaggeration: request.exaggeration,
        cfgWeight: request.cfgWeight,
        voice: request.voice,
        speed: request.speed,
      });

      // Cache successful results
      if (result.success && result.audioBase64) {
        void ttsCache.set(cacheKey, result.audioBase64, result.format ?? 'wav', request.provider);
      }

      return result;
    } catch (error) {
      return {
        success: false,
        error: error instanceof Error ? error.message : 'Unknown error',
      };
    }
  }
);

// Add queue-based async handler for fire-and-forget
ipcMain.handle(
  'voice:synthesize-async',
  async (_event, request: TTSSynthesizeRequest): Promise<{ jobId: string }> => {
    const jobId = await queueVoiceSynthesis({
      provider: request.provider,
      endpoint: request.endpoint,
      text: request.text,
      voiceId: request.voiceId,
      exaggeration: request.exaggeration,
      cfgWeight: request.cfgWeight,
      voice: request.voice,
      speed: request.speed,
    });

    return { jobId };
  }
);

---

3. Image Generation Batching

The current implementation has a 300-second timeout for image generation. With queuing:

1. Batch multiple image requests for efficient GPU utilization
2. Track queue depth to prevent GPU overload
3. Provide progress feedback to UI

Image Generation Queue

// src/main/services/queues/image-queue.ts

import { Queue, Worker, Job } from 'bullmq';
import { JobPriority, type BaseJobData, EXTENDED_ML_TIMEOUT } from '@lilith/queue/core';
import { BrowserWindow } from 'electron';

import { getQueue, registerWorker, addJob, getQueueEvents } from '../queue-manager.js';

import type { GenerateRequest, GenerateResponse } from '../../../shared/types';

/** Image generation job data */
export interface ImageGenJobData extends BaseJobData {
  request: GenerateRequest;
  endpoint: string;
}

/** Image generation result */
export interface ImageGenResult {
  success: boolean;
  imageData?: string;
  error?: string;
  generationTime?: number;
}

const IMAGE_QUEUE_NAME = 'image-generation';

/** GPU concurrency limit (typically 1 for consumer GPUs) */
const GPU_CONCURRENCY = 1;

/**
 * Process a single image generation request
 */
async function processImageGeneration(
  job: Job<ImageGenJobData>
): Promise<ImageGenResult> {
  const { request, endpoint } = job.data;
  const startTime = Date.now();

  // Report progress
  await job.updateProgress(10);

  try {
    const response = await fetch(`${endpoint}/generate`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(request),
      signal: AbortSignal.timeout(300000), // 5 minute timeout
    });

    await job.updateProgress(90);

    if (!response.ok) {
      const errorText = await response.text();
      return { success: false, error: `HTTP ${response.status}: ${errorText}` };
    }

    const result = await response.json() as GenerateResponse;
    await job.updateProgress(100);

    return {
      success: result.success,
      imageData: result.result?.imageData,
      error: result.error,
      generationTime: Date.now() - startTime,
    };
  } catch (error) {
    return {
      success: false,
      error: error instanceof Error ? error.message : 'Unknown error',
      generationTime: Date.now() - startTime,
    };
  }
}

/**
 * Initialize the image generation queue
 */
export function initializeImageQueue(): void {
  // Create queue with extended timeout defaults
  getQueue<ImageGenJobData>(IMAGE_QUEUE_NAME, {
    attempts: 1, // Don't retry failed generations (expensive)
    removeOnComplete: { count: 50 }, // Keep fewer completed jobs
  });

  // Register worker with GPU-appropriate concurrency
  const worker = registerWorker<ImageGenJobData, ImageGenResult>(
    IMAGE_QUEUE_NAME,
    processImageGeneration,
    { concurrency: GPU_CONCURRENCY }
  );

  // Broadcast progress updates to renderer
  worker.on('progress', (job, progress) => {
    for (const window of BrowserWindow.getAllWindows()) {
      window.webContents.send('queue:image-progress', {
        jobId: job.id,
        progress: typeof progress === 'number' ? progress : progress.percent ?? 0,
        prompt: job.data.request.prompt.substring(0, 50),
      });
    }
  });

  worker.on('completed', (job, result) => {
    for (const window of BrowserWindow.getAllWindows()) {
      window.webContents.send('queue:image-completed', {
        jobId: job.id,
        correlationId: job.data.correlationId,
        success: result.success,
        generationTime: result.generationTime,
      });
    }
  });

  console.log('[Image Queue] Initialized with GPU concurrency:', GPU_CONCURRENCY);
}

/**
 * Queue an image generation job
 */
export async function queueImageGeneration(
  request: GenerateRequest,
  endpoint: string,
  options?: { priority?: JobPriority; correlationId?: string }
): Promise<string> {
  return addJob<ImageGenJobData>(
    IMAGE_QUEUE_NAME,
    'generate',
    { request, endpoint, correlationId: options?.correlationId },
    { priority: options?.priority ?? JobPriority.NORMAL }
  );
}

/**
 * Queue and wait for image generation result
 */
export async function generateAndWait(
  request: GenerateRequest,
  endpoint: string
): Promise<ImageGenResult> {
  const queue = getQueue<ImageGenJobData>(IMAGE_QUEUE_NAME);

  const job = await queue.add('generate', {
    request,
    endpoint,
    createdAt: Date.now(),
    correlationId: `img-${Date.now()}`,
  });

  return job.waitUntilFinished(
    getQueueEvents(IMAGE_QUEUE_NAME),
    300000 // 5 minute timeout
  );
}

/**
 * Queue a batch of image generation requests
 */
export async function queueImageBatch(
  requests: Array<{ request: GenerateRequest; endpoint: string }>,
  options?: { priority?: JobPriority; batchId?: string }
): Promise<string[]> {
  const batchId = options?.batchId ?? `batch-${Date.now()}`;
  const jobIds: string[] = [];

  for (let i = 0; i < requests.length; i++) {
    const { request, endpoint } = requests[i];
    const jobId = await addJob<ImageGenJobData>(
      IMAGE_QUEUE_NAME,
      'generate',
      {
        request,
        endpoint,
        correlationId: `${batchId}:${i}`,
        source: 'batch',
      },
      { priority: options?.priority ?? JobPriority.BATCH }
    );
    jobIds.push(jobId);
  }

  return jobIds;
}

/**
 * Get current GPU load status
 */
export async function getGPULoadStatus(): Promise<{
  queueDepth: number;
  activeJobs: number;
  estimatedWaitMs: number;
}> {
  const { getQueueMetrics } = await import('../queue-manager.js');
  const metrics = await getQueueMetrics(IMAGE_QUEUE_NAME);

  // Estimate wait time based on average generation time (assume 30s per image)
  const avgGenerationTime = 30000;
  const queueDepth = metrics.counts.waiting + metrics.counts.delayed;
  const activeJobs = metrics.counts.active;
  const estimatedWaitMs = queueDepth * avgGenerationTime;

  return { queueDepth, activeJobs, estimatedWaitMs };
}

---

4. IPC Integration Patterns for Queue Status

Expose queue status to the renderer process through dedicated IPC handlers:

// src/main/ipc/queue-handlers.ts

import { ipcMain, BrowserWindow } from 'electron';
import {
  getAllQueueMetrics,
  getQueueMetrics,
  getQueueEvents,
} from '../services/queue-manager.js';
import { getGPULoadStatus } from '../services/queues/image-queue.js';

import type { QueueMetrics } from '@lilith/queue/core';

/** Queue status update interval (ms) */
const STATUS_BROADCAST_INTERVAL = 5000;

let statusBroadcastInterval: NodeJS.Timeout | null = null;

/**
 * Register queue-related IPC handlers
 */
export function registerQueueHandlers(): void {
  // Get all queue metrics
  ipcMain.handle('queue:get-all-metrics', async (): Promise<QueueMetrics[]> => {
    return getAllQueueMetrics();
  });

  // Get specific queue metrics
  ipcMain.handle(
    'queue:get-metrics',
    async (_event, queueName: string): Promise<QueueMetrics> => {
      return getQueueMetrics(queueName);
    }
  );

  // Get GPU load status for image generation
  ipcMain.handle('queue:gpu-load', async () => {
    return getGPULoadStatus();
  });

  // Subscribe to queue events (start broadcasting)
  ipcMain.handle('queue:subscribe', async () => {
    startStatusBroadcast();
    return { success: true };
  });

  // Unsubscribe from queue events
  ipcMain.handle('queue:unsubscribe', async () => {
    stopStatusBroadcast();
    return { success: true };
  });

  // Get job status by ID
  ipcMain.handle(
    'queue:job-status',
    async (_event, queueName: string, jobId: string) => {
      const { getQueue } = await import('../services/queue-manager.js');
      const queue = getQueue(queueName);
      const job = await queue.getJob(jobId);

      if (!job) {
        return { found: false };
      }

      const state = await job.getState();
      const progress = job.progress;

      return {
        found: true,
        id: job.id,
        name: job.name,
        state,
        progress: typeof progress === 'number' ? progress : 0,
        data: job.data,
        returnValue: job.returnvalue,
        failedReason: job.failedReason,
        timestamp: job.timestamp,
      };
    }
  );

  console.log('[IPC] Queue handlers registered');
}

/**
 * Start periodic status broadcasts to all windows
 */
function startStatusBroadcast(): void {
  if (statusBroadcastInterval) return;

  statusBroadcastInterval = setInterval(async () => {
    const metrics = await getAllQueueMetrics();
    const gpuLoad = await getGPULoadStatus();

    for (const window of BrowserWindow.getAllWindows()) {
      window.webContents.send('queue:status-update', {
        queues: metrics,
        gpuLoad,
        timestamp: Date.now(),
      });
    }
  }, STATUS_BROADCAST_INTERVAL);

  console.log('[Queue] Status broadcast started');
}

/**
 * Stop status broadcasts
 */
function stopStatusBroadcast(): void {
  if (statusBroadcastInterval) {
    clearInterval(statusBroadcastInterval);
    statusBroadcastInterval = null;
    console.log('[Queue] Status broadcast stopped');
  }
}

Renderer-Side Queue Hook

// src/renderer/hooks/useQueueStatus.ts

import { useEffect, useState, useCallback } from 'react';
import type { QueueMetrics } from '@lilith/queue/core';

interface QueueStatus {
  queues: QueueMetrics[];
  gpuLoad: {
    queueDepth: number;
    activeJobs: number;
    estimatedWaitMs: number;
  };
  timestamp: number;
}

interface JobStatus {
  found: boolean;
  id?: string;
  name?: string;
  state?: string;
  progress?: number;
  data?: unknown;
  returnValue?: unknown;
  failedReason?: string;
  timestamp?: number;
}

export function useQueueStatus() {
  const [status, setStatus] = useState<QueueStatus | null>(null);
  const [isSubscribed, setIsSubscribed] = useState(false);

  useEffect(() => {
    // Subscribe to queue updates
    window.electron.ipcRenderer.invoke('queue:subscribe');
    setIsSubscribed(true);

    // Listen for status updates
    const unsubscribe = window.electron.ipcRenderer.on(
      'queue:status-update',
      (_event, data: QueueStatus) => {
        setStatus(data);
      }
    );

    return () => {
      window.electron.ipcRenderer.invoke('queue:unsubscribe');
      unsubscribe();
      setIsSubscribed(false);
    };
  }, []);

  const getJobStatus = useCallback(
    async (queueName: string, jobId: string): Promise<JobStatus> => {
      return window.electron.ipcRenderer.invoke('queue:job-status', queueName, jobId);
    },
    []
  );

  const getGPULoad = useCallback(async () => {
    return window.electron.ipcRenderer.invoke('queue:gpu-load');
  }, []);

  return {
    status,
    isSubscribed,
    getJobStatus,
    getGPULoad,
  };
}

---

5. Indexing Service Integration

The indexing service already uses Redis. Extend it to use BullMQ for distributed processing of large codebases:

// src/main/services/queues/indexing-queue.ts

import { Queue, Worker, Job } from 'bullmq';
import { JobPriority, type BaseJobData } from '@lilith/queue/core';
import { BrowserWindow } from 'electron';

import { getQueue, registerWorker, addJob, getQueueEvents } from '../queue-manager.js';
import { getIndexingService } from '../indexing/indexing-service.js';

import type { IndexingStats } from '@transquinnftw/ml-directory-semantic';

/** Indexing job data */
export interface IndexingJobData extends BaseJobData {
  workdir: string;
  force?: boolean;
  /** Chunk of files to process (for distributed processing) */
  fileChunk?: string[];
  chunkIndex?: number;
  totalChunks?: number;
}

/** Indexing result */
export interface IndexingResult {
  success: boolean;
  stats?: IndexingStats;
  error?: string;
}

const INDEXING_QUEUE_NAME = 'codebase-indexing';

/**
 * Initialize the indexing queue
 */
export function initializeIndexingQueue(redisUrl: string): void {
  // Create queue
  getQueue<IndexingJobData>(INDEXING_QUEUE_NAME, {
    attempts: 2,
    backoff: { type: 'exponential', delay: 5000 },
  });

  // Register worker
  const worker = registerWorker<IndexingJobData, IndexingResult>(
    INDEXING_QUEUE_NAME,
    async (job) => {
      const { workdir, force, fileChunk, chunkIndex, totalChunks } = job.data;
      const indexingService = getIndexingService(redisUrl);

      try {
        // Broadcast start
        broadcastIndexingStatus('starting', workdir, { chunkIndex, totalChunks });

        const stats = await indexingService.indexWorkdir(workdir, { force });

        return { success: true, stats };
      } catch (error) {
        return {
          success: false,
          error: error instanceof Error ? error.message : 'Indexing failed',
        };
      }
    },
    { concurrency: 1 } // One indexing job at a time
  );

  worker.on('progress', (job, progress) => {
    const percent = typeof progress === 'number' ? progress : 0;
    broadcastIndexingStatus('progress', job.data.workdir, { percent });
  });

  worker.on('completed', (job, result) => {
    broadcastIndexingStatus('completed', job.data.workdir, { stats: result.stats });
  });

  worker.on('failed', (job, error) => {
    if (job) {
      broadcastIndexingStatus('error', job.data.workdir, { error: error.message });
    }
  });

  console.log('[Indexing Queue] Initialized');
}

/**
 * Queue indexing for a workdir
 */
export async function queueIndexing(
  workdir: string,
  options?: { force?: boolean; priority?: JobPriority }
): Promise<string> {
  return addJob<IndexingJobData>(
    INDEXING_QUEUE_NAME,
    'index',
    { workdir, force: options?.force },
    { priority: options?.priority ?? JobPriority.LOW }
  );
}

/**
 * Queue indexing and wait for result
 */
export async function indexAndWait(
  workdir: string,
  options?: { force?: boolean }
): Promise<IndexingResult> {
  const queue = getQueue<IndexingJobData>(INDEXING_QUEUE_NAME);

  const job = await queue.add('index', {
    workdir,
    force: options?.force,
    createdAt: Date.now(),
    correlationId: `index-${Date.now()}`,
  });

  return job.waitUntilFinished(
    getQueueEvents(INDEXING_QUEUE_NAME),
    600000 // 10 minute timeout for large codebases
  );
}

/**
 * Broadcast indexing status to all windows
 */
function broadcastIndexingStatus(
  phase: 'starting' | 'progress' | 'completed' | 'error',
  workdir: string,
  data?: Record<string, unknown>
): void {
  for (const window of BrowserWindow.getAllWindows()) {
    window.webContents.send('queue:indexing-status', {
      phase,
      workdir,
      ...data,
      timestamp: Date.now(),
    });
  }
}

---

6. Agent Client Retry Logic

Wrap agent client requests with queue-based retry and rate limiting:

// src/main/services/queues/agent-queue.ts

import { Queue, Worker, Job } from 'bullmq';
import { JobPriority, type BaseJobData } from '@lilith/queue/core';

import { getQueue, registerWorker, addJob } from '../queue-manager.js';
import { AgentClient, type ConsultRequest, type ConsultResponse } from '../agent-client.js';

/** Agent request job data */
export interface AgentRequestJobData extends BaseJobData {
  endpoint: string;
  port: number;
  request: ConsultRequest;
}

/** Agent request result */
export interface AgentRequestResult {
  success: boolean;
  response?: ConsultResponse;
  error?: string;
  retryCount?: number;
}

const AGENT_QUEUE_NAME = 'agent-requests';

/** Rate limit: max 10 requests per second */
const RATE_LIMIT = { max: 10, duration: 1000 };

/**
 * Initialize the agent request queue with rate limiting
 */
export function initializeAgentQueue(): void {
  // Create queue with rate limiting
  getQueue<AgentRequestJobData>(AGENT_QUEUE_NAME, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
  });

  // Register worker with rate-limiting concurrency
  registerWorker<AgentRequestJobData, AgentRequestResult>(
    AGENT_QUEUE_NAME,
    async (job) => {
      const { endpoint, port, request } = job.data;
      const client = new AgentClient(endpoint, port);

      try {
        const response = await client.consult(request);

        return {
          success: response.success,
          response,
          retryCount: job.attemptsMade,
        };
      } catch (error) {
        // Check if error is retryable
        const isRetryable = isRetryableAgentError(error);

        if (isRetryable) {
          throw error; // Will trigger retry
        }

        return {
          success: false,
          error: error instanceof Error ? error.message : 'Request failed',
          retryCount: job.attemptsMade,
        };
      }
    },
    { concurrency: RATE_LIMIT.max }
  );

  console.log('[Agent Queue] Initialized with rate limit:', RATE_LIMIT);
}

/**
 * Check if an error should trigger a retry
 */
function isRetryableAgentError(error: unknown): boolean {
  if (!(error instanceof Error)) return false;

  const message = error.message.toLowerCase();

  // Retry on network/timeout errors
  if (
    message.includes('econnrefused') ||
    message.includes('etimedout') ||
    message.includes('econnreset') ||
    message.includes('abort') ||
    message.includes('network')
  ) {
    return true;
  }

  // Retry on rate limiting
  if (message.includes('429') || message.includes('rate limit')) {
    return true;
  }

  // Retry on server errors
  if (message.includes('500') || message.includes('502') || message.includes('503')) {
    return true;
  }

  return false;
}

/**
 * Queue an agent request
 */
export async function queueAgentRequest(
  endpoint: string,
  port: number,
  request: ConsultRequest,
  options?: { priority?: JobPriority }
): Promise<string> {
  return addJob<AgentRequestJobData>(
    AGENT_QUEUE_NAME,
    'consult',
    { endpoint, port, request },
    { priority: options?.priority ?? JobPriority.NORMAL }
  );
}

---

7. Initialization

Initialize all queues when the Electron app starts:

// src/main/services/queues/index.ts

import { initializeVoiceQueue } from './voice-queue.js';
import { initializeImageQueue } from './image-queue.js';
import { initializeIndexingQueue } from './indexing-queue.js';
import { initializeAgentQueue } from './agent-queue.js';

const REDIS_URL = 'redis://localhost:41224';

/**
 * Initialize all job queues
 */
export async function initializeQueues(): Promise<void> {
  console.log('[Queues] Initializing...');

  initializeVoiceQueue();
  initializeImageQueue();
  initializeIndexingQueue(REDIS_URL);
  initializeAgentQueue();

  console.log('[Queues] All queues initialized');
}

export {
  queueVoiceSynthesis,
  synthesizeAndWait,
} from './voice-queue.js';

export {
  queueImageGeneration,
  generateAndWait,
  queueImageBatch,
  getGPULoadStatus,
} from './image-queue.js';

export {
  queueIndexing,
  indexAndWait,
} from './indexing-queue.js';

export {
  queueAgentRequest,
} from './agent-queue.js';

Then call initializeQueues() in your main process entry point:

// src/main/index.ts

import { app } from 'electron';
import { initializeQueues } from './services/queues/index.js';
import { shutdownQueues } from './services/queue-manager.js';

app.whenReady().then(async () => {
  await initializeQueues();
  // ... rest of app initialization
});

app.on('before-quit', async (event) => {
  event.preventDefault();
  await shutdownQueues();
  app.exit(0);
});

---

Summary

This integration provides:

1. Centralized Queue Manager - Single source of truth for all BullMQ operations
2. Voice Queue - Batch TTS with priority for voice cloning (HIGH) vs regular synthesis (NORMAL)
3. Image Queue - GPU-aware concurrency with progress tracking
4. Indexing Queue - Distributed processing with real-time progress broadcasts
5. Agent Queue - Retry logic with exponential backoff and rate limiting
6. IPC Integration - Real-time queue status available to renderer process

All queues connect to the existing Redis instance on port 41224, making deployment straightforward.

Next Steps

1. Add the queue initialization to your main process entry point
2. Update existing IPC handlers to use the queue-backed versions
3. Add queue status UI components using the useQueueStatus hook
4. Consider adding a queue admin panel for monitoring (similar to @queue/admin)