Connor-Finn · Oct 4, 2026 · Oct 4, 2026
diff --git a/backend/src/__tests__/events/PgBossEventBus.test.ts b/backend/src/__tests__/events/PgBossEventBus.test.ts
@@ -0,0 +1,60 @@
+import { PgBossEventBus } from '../../events/PgBossEventBus';
+import { createTestEvent } from '../helpers/testUtils';
+
+function setup() {
+  const subscriptions = new Map<string, (message: any) => Promise<void>>();
+  const queue = {
+    publish: jest.fn().mockResolvedValue('job-1'),
+    subscribe: jest.fn(async (name: string, handler: any) => { subscriptions.set(name, handler); }),
+  } as any;
+  const prisma = { domainEventLog: { create: jest.fn().mockResolvedValue({}) } } as any;
+  const bus = new PgBossEventBus(prisma, queue);
+  const published = () => queue.publish.mock.calls.map((c: any[]) => c[0]);
+  return { bus, queue, subscriptions, published };
+}
+
+const event = () => createTestEvent('shipment.created', 'shipment', 'ship-1', {});
+
+describe('PgBossEventBus fan-out (#327)', () => {
+  it('hands events to the dispatch queue in a process that does not run handlers', async () => {
+    const { bus, published } = setup();
+    // Routing registered but never started: the API with DISABLE_EMBEDDED_WORKERS=true.
+    await bus.subscribe('projection.shipment', ['shipment.*'], jest.fn());
+
+    await bus.publish(event());
+
+    expect(published()).toEqual(['evt.__dispatch']);
+  });
+
+  it('fans out directly to matching handler queues in a process that runs handlers', async () => {
+    const { bus, published, subscriptions } = setup();
+    await bus.subscribe('projection.shipment', ['shipment.*'], jest.fn());
+    await bus.subscribe('projection.order', ['order.*'], jest.fn());
+    await bus.start();
+
+    await bus.publish(event());
+
+    expect(published()).toEqual(['evt.projection.shipment']);
+    expect([...subscriptions.keys()]).toEqual(['evt.projection.shipment', 'evt.projection.order', 'evt.__dispatch']);
+  });
+
+  it('fans dispatched events out to its own handler queues', async () => {
+    const { bus, published, subscriptions } = setup();
+    await bus.subscribe('audit', ['*'], jest.fn());
+    await bus.start();
+
+    await subscriptions.get('evt.__dispatch')!({ type: 'shipment.created', payload: event() });
+
+    expect(published()).toEqual(['evt.audit']);
+  });
+
+  it('keeps dispatching when started with no handlers, so nothing is dropped', async () => {
+    const { bus, published, subscriptions } = setup();
+    await bus.start();
+
+    await bus.publish(event());
+
+    expect(published()).toEqual(['evt.__dispatch']);
+    expect(subscriptions.has('evt.__dispatch')).toBe(false);
+  });
+});
diff --git a/backend/src/events/PgBossEventBus.ts b/backend/src/events/PgBossEventBus.ts
@@ -19,6 +19,14 @@ import { IQueueAdapter, QueueMessage } from '../queue/IQueueAdapter.js';
 
 const QUEUE_PREFIX = 'evt.';
 
+/**
+ * Events published by a process that doesn't run handlers (the API with DISABLE_EMBEDDED_WORKERS,
+ * or an integrations-only worker) can't be fanned out there: fan-out only knows the handlers
+ * registered in its own process (#327). They go to this queue instead, and whichever process does
+ * run handlers consumes it and fans them out to its handler queues.
+ */
+const DISPATCH_QUEUE = QUEUE_PREFIX + '__dispatch';
+
 interface HandlerRegistration {
   patterns: string[];
   handler: EventHandler;
@@ -31,6 +39,8 @@ type TxClient = Omit<PrismaClient, '$connect' | '$disconnect' | '$on' | '$transa
 export class PgBossEventBus implements IEventBus {
   private registry = new Map<string, HandlerRegistration>();
   private started = false;
+  /** True once start() wired at least one handler: this process consumes events itself. */
+  private consuming = false;
 
   constructor(
     private prisma: PrismaClient,
@@ -61,22 +71,26 @@ export class PgBossEventBus implements IEventBus {
    * Fan out an already-persisted event to matching handler queues.
    * Called after transaction commit. Failures are logged but don't throw
    * because the event is already safely in DomainEventLog.
+   *
+   * A process that runs handlers fans out itself; one that doesn't hands the event to the dispatch
+   * queue, for a process that does (#327).
    */
   async fanOut<T>(event: DomainEvent<T>): Promise<void> {
+    if (!this.consuming) {
+      await this.queue.publish(DISPATCH_QUEUE, toMessage(event)).catch((err) => {
+        console.error(`[EventBus] Dispatch failed for event ${event.id} (${event.type}): ${(err as Error).message}`);
+      });
+      return;
+    }
+    await this.fanOutLocally(event);
+  }
+
+  private async fanOutLocally<T>(event: DomainEvent<T>): Promise<void> {
     const fanOutPromises: Promise<string>[] = [];
 
     for (const [handlerName, registration] of this.registry) {
       if (this.matchesAny(event.type, registration.patterns)) {
-        const queueName = QUEUE_PREFIX + handlerName;
-        const message: QueueMessage = {
-          type: event.type,
-          payload: event,
-          metadata: {
-            timestamp: event.timestamp,
-            sourceId: event.id,
-          },
-        };
-        fanOutPromises.push(this.queue.publish(queueName, message));
+        fanOutPromises.push(this.queue.publish(QUEUE_PREFIX + handlerName, toMessage(event)));
       }
     }
 
@@ -141,11 +155,20 @@ export class PgBossEventBus implements IEventBus {
       await this.wireHandler(handlerName, reg.handler, reg.options);
     }
     this.started = true;
+
+    // A process with handlers also takes events dispatched by processes without any.
+    if (this.registry.size > 0) {
+      await this.queue.subscribe(DISPATCH_QUEUE, async (message: QueueMessage) => {
+        await this.fanOutLocally(message.payload as DomainEvent);
+      });
+      this.consuming = true;
+    }
     console.log(`[EventBus] Started with ${this.registry.size} handler(s)`);
   }
 
   async stop(): Promise<void> {
     this.started = false;
+    this.consuming = false;
     console.log('[EventBus] Stopped');
   }
 
@@ -190,3 +213,12 @@ export class PgBossEventBus implements IEventBus {
     });
   }
 }
+
+function toMessage<T>(event: DomainEvent<T>): QueueMessage {
+  return {
+    type: event.type,
+    payload: event,
+    metadata: { timestamp: event.timestamp, sourceId: event.id },
+  };
+}
+
diff --git a/backend/src/events/registerHandlers.ts b/backend/src/events/registerHandlers.ts
@@ -250,14 +250,3 @@ function shipmentRoutePlanningHandler(prisma: PrismaClient, commandBus: ICommand
   ));
 }
 
-/**
- * Route planning for the embedded (API process) workers, which register the other handlers without
- * a command bus (#328). The standalone worker gets it through registerEventHandlers instead, so it
- * is never registered twice.
- */
-export async function registerShipmentRoutePlanning(eventBus: IEventBus, prisma: PrismaClient, commandBus: ICommandBus): Promise<void> {
-  const handler = shipmentRoutePlanningHandler(prisma, commandBus);
-  await eventBus.subscribe(handler.name, handler.eventPatterns, (event) => handler.handle(event), handler.options);
-  console.log(`[EventBus] Registered handler: ${handler.name} (patterns: ${handler.eventPatterns.join(', ')})`);
-}
-
diff --git a/backend/src/index.ts b/backend/src/index.ts
@@ -24,7 +24,7 @@ import { createOrderAutoArchiveWorker, registerOrderAutoArchiveSchedule, ORDER_A
 import { OrderAutoArchiveService } from './services/OrderAutoArchiveService.js';
 import { createCarrierUserAnonymizeWorker, registerCarrierUserAnonymizeSchedule, CARRIER_USER_ANONYMIZE_QUEUE } from './workers/carrierUserAnonymizeWorker.js';
 import { CarrierUserAnonymizationService } from './services/CarrierUserAnonymizationService.js';
-import { registerEventHandlers, registerShipmentRoutePlanning } from './events/registerHandlers.js';
+import { registerEventHandlers } from './events/registerHandlers.js';
 import { ICommandBus } from './commands/CommandBus.js';
 import type { IEventBus } from './events/IEventBus.js';
 import { createWebhookRetryWorker, registerWebhookRetrySchedule, WEBHOOK_RETRY_QUEUE } from './workers/webhookRetryWorker.js';
@@ -153,16 +153,19 @@ async function start() {
     if (process.env.DISABLE_EMBEDDED_WORKERS !== 'true') {
       // CQRS event handlers (projections, audit, notifications) — wired in the
       // backend process for dev. In prod with a separate worker container,
-      // DISABLE_EMBEDDED_WORKERS=true keeps these out so they aren't processed twice.
+      // DISABLE_EMBEDDED_WORKERS=true keeps these out so they aren't processed twice; this process
+      // then runs no handlers, so its events go to the dispatch queue for the worker (#327).
       try {
         const eventBus = container.resolve<IEventBus>(TOKENS.IEventBus);
         const documentService = container.has(TOKENS.IDocumentGenerationService)
           ? container.resolve<IDocumentGenerationService>(TOKENS.IDocumentGenerationService)
           : undefined;
+        // With the command bus, the command-driven handlers (issue engine, auto-replenishment,
+        // route planning) run here too, as they do in the standalone worker (#327).
         await registerEventHandlers(
-          eventBus, server.prisma, undefined, undefined, undefined, undefined, undefined, documentService
+          eventBus, server.prisma, undefined, undefined, undefined,
+          container.resolve<ICommandBus>(TOKENS.ICommandBus), undefined, documentService
         );
-        await registerShipmentRoutePlanning(eventBus, server.prisma, container.resolve<ICommandBus>(TOKENS.ICommandBus));
         await eventBus.start();
         server.log.info('Embedded event handlers registered (projections + audit + notifications)');
       } catch (err) {

diff --git a/backend/src/worker.ts b/backend/src/worker.ts
@@ -4,7 +4,7 @@
  * This process:
  * - Does NOT start Fastify or listen on any HTTP port
  * - Creates its own PrismaClient with its own connection pool
- * - Creates its own PgBossQueueAdapter
+ * - Sets up the same DI container as the API (queue adapter, event bus, full command bus)
  * - Registers event handlers (audit, notifications, email, webhooks, triage)
  * - Optionally runs the existing operational workers (inbound webhook)
  *
@@ -20,17 +20,10 @@
  */
 
 import { PrismaClient } from '@prisma/client';
-import { PgBossQueueAdapter } from './queue/PgBossQueueAdapter.js';
 import { PgBossEventBus } from './events/PgBossEventBus.js';
 import { registerEventHandlers } from './events/registerHandlers.js';
 import { QUEUES } from './queue/events.js';
 import { createInboundWebhookWorker } from './workers/inboundWebhookWorker.js';
-import { OrderDeliveryService } from './services/OrderDeliveryService.js';
-import { ArrivalCriteriaEvaluationService } from './services/ArrivalCriteriaEvaluationService.js';
-import { RecordGeofenceArrivalCommandHandler } from './commands/tracking/RecordGeofenceArrivalCommand.js';
-import { RecordGeofenceDepartureCommandHandler } from './commands/tracking/RecordGeofenceDepartureCommand.js';
-import { RecordJourneyCheckpointCommandHandler } from './commands/tracking/RecordJourneyCheckpointCommand.js';
-import { RecordStopOrdersDeliveryCommandHandler } from './commands/orders/RecordStopOrdersDeliveryCommand.js';
 import { IEmailService } from './services/IEmailService.js';
 import { SmtpEmailService } from './services/SmtpEmailService.js';
 import { ConsoleEmailService } from './services/ConsoleEmailService.js';
@@ -39,14 +32,13 @@ import { DatabaseBinaryStorage } from './storage/DatabaseBinaryStorage.js';
 import { S3FileStorage } from './storage/S3FileStorage.js';
 import { AnthropicLlmProvider } from './services/llm/AnthropicLlmProvider.js';
 import { ILlmProvider } from './services/llm/ILlmProvider.js';
-import { CommandBus, ICommandBus } from './commands/CommandBus.js';
-import { CreateAgentDecisionCommandHandler } from './commands/agentDecisions/CreateAgentDecisionCommand.js';
-import { RecordDecisionOutcomeCommandHandler } from './commands/agentDecisions/RecordDecisionOutcomeCommand.js';
-import { PromoteDecisionCommandHandler } from './commands/agentDecisions/PromoteDecisionCommand.js';
-import { CreateIssueCommandHandler } from './commands/issues/CreateIssueCommand.js';
-import { UpdateIssueCommandHandler } from './commands/issues/UpdateIssueCommand.js';
-import { EscalateIssueCommandHandler } from './commands/issues/EscalateIssueCommand.js';
-import { SetShipmentRouteCommandHandler } from './commands/shipments/SetShipmentRouteCommand.js';
+import { ICommandBus } from './commands/CommandBus.js';
+import { registerDependencies } from './di/index.js';
+import { container } from './di/container.js';
+import { TOKENS } from './di/tokens.js';
+import type { IQueueAdapter } from './queue/IQueueAdapter.js';
+import type { IOrderDeliveryService } from './services/OrderDeliveryService.js';
+import type { IArrivalCriteriaEvaluationService } from './services/ArrivalCriteriaEvaluationService.js';
 import { DEFAULT_TRIAGE_PROMPT, DEFAULT_TRIAGE_EVENTS } from './events/handlers/TriageAgentHandler.js';
 import { SkillRegistry } from './services/skills/SkillRegistry.js';
 import { DocumentGenerationService, IDocumentGenerationService } from './services/DocumentGenerationService.js';
@@ -103,15 +95,14 @@ async function startWorker() {
   await prisma.$connect();
   console.log('[Worker] Database connected');
 
-  // Own queue adapter
-  const dbUrl = process.env.DATABASE_URL || '';
-  const queue = new PgBossQueueAdapter(dbUrl);
+  // The same container as the API (#327): one queue adapter, one event bus and the full command
+  // bus, so every command-driven handler runs here whether or not an LLM is configured.
+  registerDependencies(prisma);
+  const queue = container.resolve<IQueueAdapter>(TOKENS.IQueueAdapter);
   await queue.start();
   console.log('[Worker] Queue adapter started');
-
-  // One event bus for the process. Fan-out only reaches handlers registered on this bus, so the
-  // integration workers below publish through the same instance the event handlers register on.
-  const eventBus = new PgBossEventBus(prisma, queue);
+  const eventBus = container.resolve<PgBossEventBus>(TOKENS.IEventBus);
+  const commandBus = container.resolve<ICommandBus>(TOKENS.ICommandBus);
 
   // Event handlers (audit, notifications, email, webhooks, triage)
   if (WORKER_MODE === 'all' || WORKER_MODE === 'events') {
@@ -156,7 +147,6 @@ async function startWorker() {
     // LLM key may only drive it when that org is the only tenant. With several orgs, one tenant's
     // key would pay for, and see, every other tenant's agent traffic, so the environment is used.
     let llmProvider: ILlmProvider | undefined;
-    let workerCommandBus: ICommandBus | undefined;
 
     const soleOrgId = await resolveSoleOrganizationId(prisma);
     const org = soleOrgId
@@ -177,17 +167,6 @@ async function startWorker() {
         baseURL: process.env.ANTHROPIC_BASE_URL,
       });
 
-      // Worker-local command bus for agent handlers to dispatch commands
-      const bus = new CommandBus();
-      bus.register(new CreateAgentDecisionCommandHandler(prisma, eventBus));
-      bus.register(new RecordDecisionOutcomeCommandHandler(prisma, eventBus));
-      bus.register(new PromoteDecisionCommandHandler(prisma, eventBus));
-      bus.register(new CreateIssueCommandHandler(prisma, eventBus));
-      bus.register(new UpdateIssueCommandHandler(prisma, eventBus));
-      bus.register(new EscalateIssueCommandHandler(prisma, eventBus));
-      // The route planning handler stores routes through this bus (#328).
-      bus.register(new SetShipmentRouteCommandHandler(prisma, eventBus));
-      workerCommandBus = bus;
 
       const source = org?.llmApiKey ? 'org config' : 'env var';
       console.log(`[Worker] LLM provider configured (Anthropic via ${source}), AI agents enabled`);
@@ -199,10 +178,8 @@ async function startWorker() {
 
     // Build skill registry for automation rules (always available, even without LLM)
     const skillRegistry = new SkillRegistry();
-    if (workerCommandBus) {
-      skillRegistry.register(new CreateIssueSkill(workerCommandBus));
-      skillRegistry.register(new EscalateIssueSkill(workerCommandBus));
-    }
+    skillRegistry.register(new CreateIssueSkill(commandBus));
+    skillRegistry.register(new EscalateIssueSkill(commandBus));
     skillRegistry.register(new CallWebhookSkill());
     if (emailService) {
       skillRegistry.register(new SendEmailSkill(emailService));
@@ -221,26 +198,19 @@ async function startWorker() {
         )
       : undefined;
 
-    await registerEventHandlers(eventBus, prisma, emailService, storageProvider, llmProvider, workerCommandBus, skillRegistry, documentService);
+    await registerEventHandlers(eventBus, prisma, emailService, storageProvider, llmProvider, commandBus, skillRegistry, documentService);
     await eventBus.start();
     console.log('[Worker] Event handlers registered and started');
   }
 
   // Integration workers (inbound webhook). The legacy outbound carrier and
   // outbound tracking workers were removed — outbound EDI is now driven by
   // Edi856AutoSendHandler and Edi810AutoSendHandler off domain events.
-  // Registered after the event handlers so the bus already knows where to fan out (#287).
+  // In integrations-only mode this process runs no handlers, so its events go to the dispatch queue
+  // for an events worker to fan out (#327).
   if (WORKER_MODE === 'all' || WORKER_MODE === 'integrations') {
-    if (WORKER_MODE === 'integrations') {
-      console.warn('[Worker] integrations-only mode: tracking events are recorded but only fan out to handlers registered in this process');
-    }
-    const trackingCommandBus = new CommandBus();
-    trackingCommandBus.register(new RecordGeofenceArrivalCommandHandler(prisma, eventBus));
-    trackingCommandBus.register(new RecordGeofenceDepartureCommandHandler(prisma, eventBus));
-    trackingCommandBus.register(new RecordJourneyCheckpointCommandHandler(prisma, eventBus));
-    trackingCommandBus.register(new RecordStopOrdersDeliveryCommandHandler(prisma, eventBus));
-    const deliveryService = new OrderDeliveryService(prisma, trackingCommandBus);
-    const arrivalCriteriaService = new ArrivalCriteriaEvaluationService(prisma, deliveryService, trackingCommandBus);
+    const deliveryService = container.resolve<IOrderDeliveryService>(TOKENS.IOrderDeliveryService);
+    const arrivalCriteriaService = container.resolve<IArrivalCriteriaEvaluationService>(TOKENS.IArrivalCriteriaEvaluationService);
     await queue.subscribe(
       QUEUES.INBOUND_WEBHOOK,
       createInboundWebhookWorker(prisma, deliveryService, arrivalCriteriaService, eventBus),