diff --git a/src/index.test.ts b/src/index.test.ts index a4fa1fa..32817cd 100644 --- a/src/index.test.ts +++ b/src/index.test.ts @@ -3,7 +3,6 @@ import { describe, expect, it, vi } from 'vitest'; import { editFormattedTrackedChannelMessage, isTerminalStatusMessage, - sendFormattedChannelMessage, sendFormattedTrackedChannelMessage, } from './index.js'; import { composeDashboardContent } from './dashboard-render.js'; @@ -81,12 +80,6 @@ describe('index scheduler messaging helpers', () => { 'msg-123', 'only hidden', ); - await sendFormattedChannelMessage( - [channel], - 'dc:test', - 'only hidden', - ); - expect(trackedResult).toBeNull(); expect(sendAndTrack).not.toHaveBeenCalled(); expect(editMessage).not.toHaveBeenCalled(); diff --git a/src/index.ts b/src/index.ts index bacc9ae..1137ee0 100644 --- a/src/index.ts +++ b/src/index.ts @@ -30,10 +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 { + deliverCanonicalOutboundMessage, + deliverIpcOutboundMessage, +} from './ipc-outbound-delivery.js'; import { findChannel, - findChannelByName, formatOutbound, normalizeMessageForDedupe, } from './router.js'; @@ -90,20 +92,6 @@ import { FAILOVER_MIN_DURATION_MS } from './config.js'; // Token rotation is initialized lazily on first use or at startup below -export async function sendFormattedChannelMessage( - channels: Channel[], - jid: string, - rawText: string, -): Promise { - const channel = findChannel(channels, jid); - if (!channel) { - logger.warn({ jid }, 'No channel owns JID, cannot send message'); - return; - } - const text = formatOutbound(rawText); - if (text) await channel.sendMessage(jid, text); -} - export async function sendFormattedTrackedChannelMessage( channels: Channel[], jid: string, @@ -156,6 +144,23 @@ const runtime = createMessageRuntime({ clearSession: runtimeState.clearSession, }); +async function deliverFormattedCanonicalMessage( + jid: string, + rawText: string, + deliveryRole?: 'owner' | 'reviewer' | 'arbiter', +): Promise { + const text = formatOutbound(rawText); + if (!text) return; + await deliverCanonicalOutboundMessage( + { jid, text, deliveryRole }, + { + channels, + roomBindings: runtimeState.getRoomBindings, + log: logger, + }, + ); +} + /** * Get available groups list for the agent. * Returns groups ordered by most recent activity. @@ -192,8 +197,7 @@ async function announceRestartRecovery( return explicitContext; } - await sendFormattedChannelMessage( - channels, + await deliverFormattedCanonicalMessage( explicitContext.chatJid, buildRestartAnnouncement(explicitContext), ); @@ -207,8 +211,7 @@ async function announceRestartRecovery( if (hasRecentRestartAnnouncement(interrupted.chatJid, dedupeSince)) { continue; } - await sendFormattedChannelMessage( - channels, + await deliverFormattedCanonicalMessage( interrupted.chatJid, buildInterruptedRestartAnnouncement(interrupted), ); @@ -230,8 +233,7 @@ async function announceRestartRecovery( return null; } - await sendFormattedChannelMessage( - channels, + await deliverFormattedCanonicalMessage( inferred.chatJid, inferred.lines.join('\n'), ); @@ -360,13 +362,6 @@ async function main(): Promise { } // Start subsystems (independently of connection handler) - // Paired-room scheduler output goes through the reviewer bot slot. - const reviewerChannelName = 'discord-review'; - const reviewerChannelForCron = findChannelByName( - channels, - reviewerChannelName, - ); - startSchedulerLoop({ roomBindings: runtimeState.getRoomBindings, getSessions: runtimeState.getSessions, @@ -374,13 +369,9 @@ async function main(): Promise { onProcess: (groupJid, proc, processName, ipcDir) => queue.registerProcess(groupJid, proc, processName, ipcDir), sendMessage: (jid, rawText) => - sendFormattedChannelMessage(channels, jid, rawText), - sendMessageViaReviewerBot: reviewerChannelForCron - ? async (jid, rawText) => { - const text = formatOutbound(rawText); - if (text) await reviewerChannelForCron.sendMessage(jid, text); - } - : undefined, + deliverFormattedCanonicalMessage(jid, rawText), + sendMessageViaReviewerBot: (jid, rawText) => + deliverFormattedCanonicalMessage(jid, rawText, 'reviewer'), sendTrackedMessage: (jid, rawText) => sendFormattedTrackedChannelMessage(channels, jid, rawText), editTrackedMessage: (jid, messageId, rawText) => diff --git a/src/ipc-outbound-delivery.test.ts b/src/ipc-outbound-delivery.test.ts index 3c28f15..baad6bd 100644 --- a/src/ipc-outbound-delivery.test.ts +++ b/src/ipc-outbound-delivery.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it, vi } from 'vitest'; -import { deliverIpcOutboundMessage } from './ipc-outbound-delivery.js'; +import { + deliverCanonicalOutboundMessage, + deliverIpcOutboundMessage, +} from './ipc-outbound-delivery.js'; import type { WorkItem } from './db/work-items.js'; import type { Channel, RegisteredGroup } from './types.js'; @@ -56,6 +59,47 @@ function makeWorkItem(input: { } describe('deliverIpcOutboundMessage', () => { + it('funnels ordinary outbound through canonical work item delivery', async () => { + const channel = makeChannel(); + const createdItem = makeWorkItem({ + chat_jid: 'dc:room', + group_folder: 'room-folder', + result_payload: '일반 안내', + }); + const createWorkItem = vi.fn(() => createdItem); + const deliverWorkItem = vi.fn(async () => true); + + const result = await deliverCanonicalOutboundMessage( + { jid: 'dc:room', text: '일반 안내' }, + { + channels: [channel], + roomBindings: () => ({ 'dc:room': makeGroup() }), + createWorkItem, + deliverWorkItem, + }, + ); + + expect(result).toBe('delivered'); + expect(createWorkItem).toHaveBeenCalledWith({ + group_folder: 'room-folder', + chat_jid: 'dc:room', + agent_type: 'codex', + delivery_role: null, + start_seq: null, + end_seq: null, + result_payload: '일반 안내', + attachments: undefined, + }); + expect(deliverWorkItem).toHaveBeenCalledWith( + expect.objectContaining({ + channel, + item: createdItem, + attachmentBaseDirs: ['/repo'], + }), + ); + expect(channel.sendMessage).not.toHaveBeenCalled(); + }); + it('funnels IPC outbound through canonical work item delivery', async () => { const ownerChannel = makeChannel(); const reviewerChannel = makeChannel('discord-review', false); @@ -155,7 +199,7 @@ describe('deliverIpcOutboundMessage', () => { deliverWorkItem, }, ), - ).rejects.toThrow('No registered room binding for IPC outbound JID'); + ).rejects.toThrow('No registered room binding for outbound JID'); expect(createWorkItem).not.toHaveBeenCalled(); expect(deliverWorkItem).not.toHaveBeenCalled(); diff --git a/src/ipc-outbound-delivery.ts b/src/ipc-outbound-delivery.ts index 371e3d0..1c00b1f 100644 --- a/src/ipc-outbound-delivery.ts +++ b/src/ipc-outbound-delivery.ts @@ -43,6 +43,23 @@ export interface IpcOutboundDeliveryDeps { deliverWorkItem?: typeof deliverOpenWorkItem; } +export interface CanonicalOutboundDeliveryArgs { + jid: string; + text: string; + deliveryRole?: PairedRoomRole; + attachments?: OutboundAttachment[]; +} + +export interface CanonicalOutboundDeliveryDeps { + channels: Channel[]; + roomBindings: () => Record; + log?: IpcDeliveryLog; + createWorkItem?: typeof createProducedWorkItem; + deliverWorkItem?: typeof deliverOpenWorkItem; +} + +export type CanonicalOutboundDeliveryResult = 'delivered' | 'queued_retry'; + export type IpcOutboundDeliveryResult = | 'delivered' | 'queued_retry' @@ -65,6 +82,61 @@ function isTerminalStatusMessage(text: string): boolean { return parseVisibleVerdict(text) !== 'continue'; } +export async function deliverCanonicalOutboundMessage( + args: CanonicalOutboundDeliveryArgs, + deps: CanonicalOutboundDeliveryDeps, +): Promise { + const log = deps.log ?? logger; + const group = deps.roomBindings()[args.jid]; + if (!group) { + throw new Error(`No registered room binding for outbound JID: ${args.jid}`); + } + + const route = resolveChannelForDeliveryRole( + deps.channels, + args.jid, + args.deliveryRole, + ); + if (!route.channel) throw new Error(`No channel for JID: ${args.jid}`); + + log.info( + { + transition: 'outbound:route', + chatJid: args.jid, + deliveryRole: args.deliveryRole ?? null, + requestedRoleChannel: route.requestedRoleChannelName, + selectedChannel: route.selectedChannelName, + usedRoleChannel: route.usedRoleChannel, + fallbackUsed: route.fallbackUsed, + }, + 'Routed outbound 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: args.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: () => {}, + }); + + return delivered ? 'delivered' : 'queued_retry'; +} + export async function deliverIpcOutboundMessage( args: IpcOutboundDeliveryArgs, deps: IpcOutboundDeliveryDeps, @@ -92,58 +164,34 @@ export async function deliverIpcOutboundMessage( 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 result = await deliverCanonicalOutboundMessage( + { + jid: args.jid, + text: args.text, + deliveryRole, + attachments: args.attachments, + }, + { + channels: deps.channels, + roomBindings: deps.roomBindings, + log, + createWorkItem: deps.createWorkItem, + deliverWorkItem: deps.deliverWorkItem, + }, + ); - 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 (result !== 'delivered') { + return result; } if ( diff --git a/src/message-runtime.ts b/src/message-runtime.ts index f139436..b2645dd 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -20,6 +20,7 @@ import { deliverOpenWorkItem, processOpenWorkItemDelivery, } from './message-runtime-delivery.js'; +import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js'; import { handleQueuedRunGates } from './message-runtime-gating.js'; import { enqueuePendingHandoffs as enqueueClaimedServiceHandoffs, @@ -93,6 +94,21 @@ export interface MessageRuntimeDeps { clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void; } +async function deliverSessionCommandMessage( + deps: Pick, + chatJid: string, + text: string, +): Promise { + await deliverCanonicalOutboundMessage( + { jid: chatJid, text }, + { + channels: deps.channels, + roomBindings: deps.getRoomBindings, + log: logger, + }, + ); +} + export function createMessageRuntime(deps: MessageRuntimeDeps): { processGroupMessages: ( chatJid: string, @@ -110,7 +126,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { chatJid: string, messages: NewMessage[], ): NewMessage[] => labelPairedSenders(deps.channels, chatJid, messages); - const enqueueScopedGroupMessageCheck = buildScopedMessageCheckEnqueuer( deps.queue, ); @@ -587,7 +602,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { timezone: deps.timezone, hasImplicitContinuationWindow: continuationTracker.has, sessionCommandDeps: { - sendMessage: (text) => channel.sendMessage(chatJid, text), + sendMessage: (text) => + deliverSessionCommandMessage(deps, chatJid, text), setTyping: (typing) => channel.setTyping?.(chatJid, typing) ?? Promise.resolve(), runAgent: (prompt, onOutput) =>