diff --git a/src/channels/discord.test.ts b/src/channels/discord.test.ts index 277098f..fb04df1 100644 --- a/src/channels/discord.test.ts +++ b/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'); - - // No error, no API call + await expect( + channel.sendMessage('dc:1234567890123456', 'No client'), + ).rejects.toThrow('Discord client not initialized'); }); it('splits messages exceeding 2000 characters', async () => { diff --git a/src/channels/discord.ts b/src/channels/discord.ts index 75072a9..d809096 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -410,8 +410,7 @@ export class DiscordChannel implements Channel { options: SendMessageOptions = {}, ): Promise { if (!this.client) { - logger.warn('Discord client not initialized'); - return; + throw new Error('Discord client not initialized'); } 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'); - return; + throw new Error(`Discord channel not found or not text-based: ${jid}`); } const textChannel = channel as TextChannel; @@ -627,11 +625,15 @@ export class DiscordChannel implements Channel { } async sendAndTrack(jid: string, text: string): Promise { - 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 { - 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( diff --git a/src/index.ts b/src/index.ts index 168fbec..bacc9ae 100644 --- a/src/index.ts +++ b/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 { }); startIpcWatcher({ sendMessage: async (jid, text, senderRole, runId, attachments) => { - if ( - runId && - (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( + await deliverIpcOutboundMessage( + { jid, text, senderRole, runId, attachments }, { - transition: 'ipc:route', - chatJid: jid, - runId: runId ?? null, - senderRole: senderRole ?? null, - requestedRoleChannel: route.requestedRoleChannelName, - selectedChannel: route.selectedChannelName, - usedRoleChannel: route.usedRoleChannel, - fallbackUsed: route.fallbackUsed, + channels, + roomBindings: runtimeState.getRoomBindings, + queue, + log: logger, }, - '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; diff --git a/src/ipc-outbound-delivery.test.ts b/src/ipc-outbound-delivery.test.ts new file mode 100644 index 0000000..3c28f15 --- /dev/null +++ b/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 { + 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(); + }); +}); diff --git a/src/ipc-outbound-delivery.ts b/src/ipc-outbound-delivery.ts new file mode 100644 index 0000000..371e3d0 --- /dev/null +++ b/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; + +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; + 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 { + 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'; +}