Merge pull request #87 from phj1081/codex/owner/ejclaw

Unify IPC outbound delivery through work items

src/channels/discord.test.ts

@@ -818,14 +818,14 @@ describe('DiscordChannel', () => {
       ).rejects.toThrow('Channel not found');
     });
 
-    it('does nothing when client is not initialized', async () => {
+    it('rejects when client is not initialized', async () => {
       const opts = createTestOpts();
       const channel = new DiscordChannel('test-token', opts);
 
       // Don't connect — client is null
-      await channel.sendMessage('dc:1234567890123456', 'No client');
+      await expect(
-
+        channel.sendMessage('dc:1234567890123456', 'No client'),
-      // No error, no API call
+      ).rejects.toThrow('Discord client not initialized');
     });
 
     it('splits messages exceeding 2000 characters', async () => {

src/channels/discord.ts

@@ -410,8 +410,7 @@ export class DiscordChannel implements Channel {
     options: SendMessageOptions = {},
   ): Promise<void> {
     if (!this.client) {
-      logger.warn('Discord client not initialized');
+      throw new Error('Discord client not initialized');
-      return;
     }
 
     try {
@@ -419,8 +418,7 @@ export class DiscordChannel implements Channel {
       const channel = await this.client.channels.fetch(channelId);
 
       if (!channel || !('send' in channel)) {
-        logger.warn({ jid }, 'Discord channel not found or not text-based');
+        throw new Error(`Discord channel not found or not text-based: ${jid}`);
-        return;
       }
 
       const textChannel = channel as TextChannel;
@@ -627,11 +625,15 @@ export class DiscordChannel implements Channel {
   }
 
   async sendAndTrack(jid: string, text: string): Promise<string | null> {
-    if (!this.client) return null;
+    if (!this.client) {
+      throw new Error('Discord client not initialized');
+    }
     try {
       const channelId = jid.replace(/^dc:/, '');
       const channel = await this.client.channels.fetch(channelId);
-      if (!channel || !('send' in channel)) return null;
+      if (!channel || !('send' in channel)) {
+        throw new Error(`Discord channel not found or not text-based: ${jid}`);
+      }
       const msg = await (channel as TextChannel).send(text);
       logger.info(
         {
@@ -763,11 +765,15 @@ export class DiscordChannel implements Channel {
     messageId: string,
     text: string,
   ): Promise<void> {
-    if (!this.client) return;
+    if (!this.client) {
+      throw new Error('Discord client not initialized');
+    }
     try {
       const channelId = jid.replace(/^dc:/, '');
       const channel = await this.client.channels.fetch(channelId);
-      if (!channel || !('messages' in channel)) return;
+      if (!channel || !('messages' in channel)) {
+        throw new Error(`Discord channel not found or not editable: ${jid}`);
+      }
       const msg = await (channel as TextChannel).messages.fetch(messageId);
       await msg.edit(text);
       logger.info(

src/index.ts

@@ -30,12 +30,12 @@ import { composeDashboardContent } from './dashboard-render.js';
 import { GroupQueue } from './group-queue.js';
 import { resolveGroupIpcPath } from './group-folder.js';
 import { startIpcWatcher } from './ipc.js';
+import { deliverIpcOutboundMessage } from './ipc-outbound-delivery.js';
 import {
   findChannel,
   findChannelByName,
   formatOutbound,
   normalizeMessageForDedupe,
-  resolveChannelForDeliveryRole,
 } from './router.js';
 import {
   buildRestartAnnouncement,
@@ -388,44 +388,15 @@ async function main(): Promise<void> {
   });
   startIpcWatcher({
     sendMessage: async (jid, text, senderRole, runId, attachments) => {
-      if (
+      await deliverIpcOutboundMessage(
-        runId &&
+        { jid, text, senderRole, runId, attachments },
-        (senderRole === 'reviewer' || senderRole === 'arbiter') &&
-        queue.hasRecordedDirectTerminalDeliveryForRun(jid, runId, senderRole)
-      ) {
-        logger.info(
-          {
-            transition: 'ipc:skip-post-terminal',
-            chatJid: jid,
-            senderRole,
-            runId,
-          },
-          'Skipped IPC relay message because the run already emitted a direct terminal verdict',
-        );
-        return;
-      }
-      const route = resolveChannelForDeliveryRole(channels, jid, senderRole);
-      if (!route.channel) throw new Error(`No channel for JID: ${jid}`);
-      logger.info(
         {
-          transition: 'ipc:route',
+          channels,
-          chatJid: jid,
+          roomBindings: runtimeState.getRoomBindings,
-          runId: runId ?? null,
+          queue,
-          senderRole: senderRole ?? null,
+          log: logger,
-          requestedRoleChannel: route.requestedRoleChannelName,
-          selectedChannel: route.selectedChannelName,
-          usedRoleChannel: route.usedRoleChannel,
-          fallbackUsed: route.fallbackUsed,
         },
-        'IPC relay routed message to channel',
       );
-      await route.channel.sendMessage(jid, text, { attachments });
-      if (
-        (senderRole === 'reviewer' || senderRole === 'arbiter') &&
-        isTerminalStatusMessage(text)
-      ) {
-        queue.noteDirectTerminalDelivery(jid, senderRole, text);
-      }
     },
     injectInboundMessage: async (payload) => {
       const jid = payload.chatJid;

src/ipc-outbound-delivery.test.ts

@@ -0,0 +1,190 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { deliverIpcOutboundMessage } from './ipc-outbound-delivery.js';
+import type { WorkItem } from './db/work-items.js';
+import type { Channel, RegisteredGroup } from './types.js';
+
+function makeChannel(name = 'discord', owns = true): Channel {
+  return {
+    name,
+    connect: async () => {},
+    sendMessage: vi.fn(async () => {}),
+    isConnected: () => true,
+    ownsJid: () => owns,
+    disconnect: async () => {},
+  };
+}
+
+function makeGroup(overrides: Partial<RegisteredGroup> = {}): RegisteredGroup {
+  return {
+    name: 'Room',
+    folder: 'room-folder',
+    added_at: '2026-04-28T00:00:00.000Z',
+    agentType: 'codex',
+    workDir: '/repo',
+    ...overrides,
+  };
+}
+
+function makeWorkItem(input: {
+  chat_jid: string;
+  group_folder: string;
+  agent_type?: WorkItem['agent_type'];
+  delivery_role?: WorkItem['delivery_role'];
+  result_payload: string;
+  attachments?: WorkItem['attachments'];
+}): WorkItem {
+  return {
+    id: 123,
+    group_folder: input.group_folder,
+    chat_jid: input.chat_jid,
+    agent_type: input.agent_type ?? 'codex',
+    service_id: 'codex-main',
+    delivery_role: input.delivery_role ?? null,
+    status: 'produced',
+    start_seq: null,
+    end_seq: null,
+    result_payload: input.result_payload,
+    attachments: input.attachments ?? [],
+    delivery_attempts: 0,
+    delivery_message_id: null,
+    last_error: null,
+    created_at: '2026-04-28T00:00:00.000Z',
+    updated_at: '2026-04-28T00:00:00.000Z',
+    delivered_at: null,
+  };
+}
+
+describe('deliverIpcOutboundMessage', () => {
+  it('funnels IPC outbound through canonical work item delivery', async () => {
+    const ownerChannel = makeChannel();
+    const reviewerChannel = makeChannel('discord-review', false);
+    const createdItem = makeWorkItem({
+      chat_jid: 'dc:room',
+      group_folder: 'room-folder',
+      delivery_role: 'reviewer',
+      result_payload: 'TASK_DONE\n검증 완료',
+    });
+    const createWorkItem = vi.fn(() => createdItem);
+    const deliverWorkItem = vi.fn(async () => true);
+    const noteDirectTerminalDelivery = vi.fn();
+
+    const result = await deliverIpcOutboundMessage(
+      {
+        jid: 'dc:room',
+        text: 'TASK_DONE\n검증 완료',
+        senderRole: 'reviewer',
+        runId: 'run-1',
+      },
+      {
+        channels: [ownerChannel, reviewerChannel],
+        roomBindings: () => ({ 'dc:room': makeGroup() }),
+        queue: { noteDirectTerminalDelivery },
+        createWorkItem,
+        deliverWorkItem,
+      },
+    );
+
+    expect(result).toBe('delivered');
+    expect(createWorkItem).toHaveBeenCalledWith({
+      group_folder: 'room-folder',
+      chat_jid: 'dc:room',
+      agent_type: 'codex',
+      delivery_role: 'reviewer',
+      start_seq: null,
+      end_seq: null,
+      result_payload: 'TASK_DONE\n검증 완료',
+      attachments: undefined,
+    });
+    expect(deliverWorkItem).toHaveBeenCalledWith(
+      expect.objectContaining({
+        channel: reviewerChannel,
+        item: createdItem,
+        attachmentBaseDirs: ['/repo'],
+      }),
+    );
+    expect(noteDirectTerminalDelivery).toHaveBeenCalledWith(
+      'dc:room',
+      'reviewer',
+      'TASK_DONE\n검증 완료',
+    );
+    expect(ownerChannel.sendMessage).not.toHaveBeenCalled();
+    expect(reviewerChannel.sendMessage).not.toHaveBeenCalled();
+  });
+
+  it('keeps failed IPC outbound in the delivery retry queue', async () => {
+    const channel = makeChannel();
+    const createWorkItem = vi.fn((input) => makeWorkItem(input as any));
+    const deliverWorkItem = vi.fn(async () => false);
+    const noteDirectTerminalDelivery = vi.fn();
+
+    const result = await deliverIpcOutboundMessage(
+      {
+        jid: 'dc:room',
+        text: 'TASK_DONE\n검증 완료',
+        senderRole: 'reviewer',
+        runId: 'run-1',
+      },
+      {
+        channels: [channel],
+        roomBindings: () => ({ 'dc:room': makeGroup() }),
+        queue: { noteDirectTerminalDelivery },
+        createWorkItem,
+        deliverWorkItem,
+      },
+    );
+
+    expect(result).toBe('queued_retry');
+    expect(createWorkItem).toHaveBeenCalledTimes(1);
+    expect(deliverWorkItem).toHaveBeenCalledTimes(1);
+    expect(noteDirectTerminalDelivery).not.toHaveBeenCalled();
+  });
+
+  it('does not create non-canonical outbound for unregistered rooms', async () => {
+    const createWorkItem = vi.fn((input) => makeWorkItem(input as any));
+    const deliverWorkItem = vi.fn(async () => true);
+
+    await expect(
+      deliverIpcOutboundMessage(
+        { jid: 'dc:missing', text: 'hello' },
+        {
+          channels: [makeChannel()],
+          roomBindings: () => ({}),
+          queue: {},
+          createWorkItem,
+          deliverWorkItem,
+        },
+      ),
+    ).rejects.toThrow('No registered room binding for IPC outbound JID');
+
+    expect(createWorkItem).not.toHaveBeenCalled();
+    expect(deliverWorkItem).not.toHaveBeenCalled();
+  });
+
+  it('skips duplicate terminal IPC messages already recorded for the run', async () => {
+    const createWorkItem = vi.fn((input) => makeWorkItem(input as any));
+    const deliverWorkItem = vi.fn(async () => true);
+
+    const result = await deliverIpcOutboundMessage(
+      {
+        jid: 'dc:room',
+        text: 'TASK_DONE\n검증 완료',
+        senderRole: 'reviewer',
+        runId: 'run-1',
+      },
+      {
+        channels: [makeChannel()],
+        roomBindings: () => ({ 'dc:room': makeGroup() }),
+        queue: {
+          hasRecordedDirectTerminalDeliveryForRun: () => true,
+        },
+        createWorkItem,
+        deliverWorkItem,
+      },
+    );
+
+    expect(result).toBe('skipped_recorded_terminal');
+    expect(createWorkItem).not.toHaveBeenCalled();
+    expect(deliverWorkItem).not.toHaveBeenCalled();
+  });
+});

src/ipc-outbound-delivery.ts

@@ -0,0 +1,156 @@
+import { createProducedWorkItem } from './db.js';
+import type { WorkItem } from './db/work-items.js';
+import { deliverOpenWorkItem } from './message-runtime-delivery.js';
+import { parseVisibleVerdict } from './paired-execution-context-shared.js';
+import { resolveChannelForDeliveryRole } from './router.js';
+import type {
+  Channel,
+  OutboundAttachment,
+  PairedRoomRole,
+  RegisteredGroup,
+} from './types.js';
+import { logger } from './logger.js';
+
+type IpcDeliveryLog = Pick<typeof logger, 'info' | 'warn' | 'error'>;
+
+interface IpcOutboundQueue {
+  hasRecordedDirectTerminalDeliveryForRun?(
+    groupJid: string,
+    runId: string,
+    senderRole?: string | null,
+  ): boolean;
+  noteDirectTerminalDelivery?(
+    groupJid: string,
+    senderRole?: string | null,
+    text?: string | null,
+  ): void;
+}
+
+export interface IpcOutboundDeliveryArgs {
+  jid: string;
+  text: string;
+  senderRole?: string;
+  runId?: string;
+  attachments?: OutboundAttachment[];
+}
+
+export interface IpcOutboundDeliveryDeps {
+  channels: Channel[];
+  roomBindings: () => Record<string, RegisteredGroup>;
+  queue: IpcOutboundQueue;
+  log?: IpcDeliveryLog;
+  createWorkItem?: typeof createProducedWorkItem;
+  deliverWorkItem?: typeof deliverOpenWorkItem;
+}
+
+export type IpcOutboundDeliveryResult =
+  | 'delivered'
+  | 'queued_retry'
+  | 'skipped_recorded_terminal';
+
+function normalizeDeliveryRole(
+  senderRole: string | undefined,
+): PairedRoomRole | undefined {
+  if (
+    senderRole === 'owner' ||
+    senderRole === 'reviewer' ||
+    senderRole === 'arbiter'
+  ) {
+    return senderRole;
+  }
+  return undefined;
+}
+
+function isTerminalStatusMessage(text: string): boolean {
+  return parseVisibleVerdict(text) !== 'continue';
+}
+
+export async function deliverIpcOutboundMessage(
+  args: IpcOutboundDeliveryArgs,
+  deps: IpcOutboundDeliveryDeps,
+): Promise<IpcOutboundDeliveryResult> {
+  const log = deps.log ?? logger;
+  const deliveryRole = normalizeDeliveryRole(args.senderRole);
+  if (
+    args.runId &&
+    (deliveryRole === 'reviewer' || deliveryRole === 'arbiter') &&
+    deps.queue.hasRecordedDirectTerminalDeliveryForRun?.(
+      args.jid,
+      args.runId,
+      deliveryRole,
+    )
+  ) {
+    log.info(
+      {
+        transition: 'ipc:skip-post-terminal',
+        chatJid: args.jid,
+        senderRole: deliveryRole,
+        runId: args.runId,
+      },
+      'Skipped IPC relay message because the run already emitted a direct terminal verdict',
+    );
+    return 'skipped_recorded_terminal';
+  }
+
+  const group = deps.roomBindings()[args.jid];
+  if (!group) {
+    throw new Error(
+      `No registered room binding for IPC outbound JID: ${args.jid}`,
+    );
+  }
+
+  const route = resolveChannelForDeliveryRole(
+    deps.channels,
+    args.jid,
+    deliveryRole,
+  );
+  if (!route.channel) throw new Error(`No channel for JID: ${args.jid}`);
+
+  log.info(
+    {
+      transition: 'ipc:route',
+      chatJid: args.jid,
+      runId: args.runId ?? null,
+      senderRole: deliveryRole ?? null,
+      requestedRoleChannel: route.requestedRoleChannelName,
+      selectedChannel: route.selectedChannelName,
+      usedRoleChannel: route.usedRoleChannel,
+      fallbackUsed: route.fallbackUsed,
+    },
+    'IPC relay routed message to canonical work item delivery',
+  );
+
+  const createWorkItem = deps.createWorkItem ?? createProducedWorkItem;
+  const workItem = createWorkItem({
+    group_folder: group.folder,
+    chat_jid: args.jid,
+    agent_type: group.agentType ?? 'claude-code',
+    delivery_role: deliveryRole ?? null,
+    start_seq: null,
+    end_seq: null,
+    result_payload: args.text,
+    attachments: args.attachments,
+  });
+
+  const deliverWorkItem = deps.deliverWorkItem ?? deliverOpenWorkItem;
+  const delivered = await deliverWorkItem({
+    channel: route.channel,
+    item: workItem as WorkItem,
+    log,
+    attachmentBaseDirs: group.workDir ? [group.workDir] : undefined,
+    isDuplicateOfLastBotFinal: () => false,
+    openContinuation: () => {},
+  });
+
+  if (!delivered) {
+    return 'queued_retry';
+  }
+
+  if (
+    (deliveryRole === 'reviewer' || deliveryRole === 'arbiter') &&
+    isTerminalStatusMessage(args.text)
+  ) {
+    deps.queue.noteDirectTerminalDelivery?.(args.jid, deliveryRole, args.text);
+  }
+  return 'delivered';
+}