diff --git a/src/channels/discord.test.ts b/src/channels/discord.test.ts index 9cba66d..e2eebca 100644 --- a/src/channels/discord.test.ts +++ b/src/channels/discord.test.ts @@ -113,6 +113,7 @@ vi.mock('discord.js', () => { }); import { DiscordChannel, DiscordChannelOpts } from './discord.js'; +import { logger } from '../logger.js'; // --- Test helpers --- @@ -815,6 +816,32 @@ describe('DiscordChannel', () => { flags: 1 << 2, }); }); + + it('logs channel name and Discord message ids after sending', async () => { + const opts = createTestOpts(); + const channel = new DiscordChannel('test-token', opts); + await channel.connect(); + + const mockChannel = { + send: vi.fn().mockResolvedValue({ id: 'discord-message-1' }), + sendTyping: vi.fn(), + }; + currentClient().channels.fetch.mockResolvedValue(mockChannel); + + await channel.sendMessage('dc:1234567890123456', 'Hello'); + + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + jid: 'dc:1234567890123456', + channelName: 'discord', + deliveryMode: 'send', + chunkCount: 1, + messageId: 'discord-message-1', + messageIds: ['discord-message-1'], + }), + 'Discord message sent', + ); + }); }); // --- ownsJid --- diff --git a/src/channels/discord.ts b/src/channels/discord.ts index 0856333..47186c7 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -533,9 +533,19 @@ export class DiscordChannel implements Channel { attachment: f, name: path.basename(f), })); + const sentMessageIds: string[] = []; + let chunkCount = 0; + + const recordSentMessage = (message: Message | null | undefined) => { + chunkCount += 1; + if (message?.id) sentMessageIds.push(message.id); + }; if (!cleaned && files.length === 0) { - logger.debug({ jid }, 'Skipping empty Discord outbound message'); + logger.debug( + { jid, channelName: this.name }, + 'Skipping empty Discord outbound message', + ); return; } @@ -547,17 +557,21 @@ export class DiscordChannel implements Channel { if (cleaned.length <= MAX_LENGTH) { // Send text with first batch of files - await textChannel.send({ - content: cleaned || undefined, - files: fileBatches[0]?.length ? fileBatches[0] : undefined, - flags: MessageFlags.SuppressEmbeds, - }); + recordSentMessage( + await textChannel.send({ + content: cleaned || undefined, + files: fileBatches[0]?.length ? fileBatches[0] : undefined, + flags: MessageFlags.SuppressEmbeds, + }), + ); // Send remaining file batches as follow-up messages for (let b = 1; b < fileBatches.length; b++) { - await textChannel.send({ - files: fileBatches[b], - flags: MessageFlags.SuppressEmbeds, - }); + recordSentMessage( + await textChannel.send({ + files: fileBatches[b], + flags: MessageFlags.SuppressEmbeds, + }), + ); } } else { // Send text in chunks, attach first batch to the first chunk @@ -565,24 +579,43 @@ export class DiscordChannel implements Channel { for (let i = 0; i < cleaned.length; i += MAX_LENGTH) { const chunk = cleaned.slice(i, i + MAX_LENGTH); const batch = fileBatches[fileBatchIndex]; - await textChannel.send({ - content: chunk, - files: batch?.length ? batch : undefined, - flags: MessageFlags.SuppressEmbeds, - }); + recordSentMessage( + await textChannel.send({ + content: chunk, + files: batch?.length ? batch : undefined, + flags: MessageFlags.SuppressEmbeds, + }), + ); if (batch?.length) fileBatchIndex++; } // Send any remaining file batches for (let b = fileBatchIndex; b < fileBatches.length; b++) { - await textChannel.send({ - files: fileBatches[b], - flags: MessageFlags.SuppressEmbeds, - }); + recordSentMessage( + await textChannel.send({ + files: fileBatches[b], + flags: MessageFlags.SuppressEmbeds, + }), + ); } } - logger.info({ jid, length: text.length }, 'Discord message sent'); + logger.info( + { + jid, + channelName: this.name, + length: text.length, + deliveryMode: 'send', + chunkCount, + attachmentCount: files.length, + messageId: sentMessageIds[0] ?? null, + messageIds: sentMessageIds, + }, + 'Discord message sent', + ); } catch (err) { - logger.error({ jid, err }, 'Failed to send Discord message'); + logger.error( + { jid, channelName: this.name, err }, + 'Failed to send Discord message', + ); throw err; } } @@ -780,8 +813,21 @@ export class DiscordChannel implements Channel { if (!channel || !('messages' in channel)) return; const msg = await (channel as TextChannel).messages.fetch(messageId); await msg.edit(text); + logger.info( + { + jid, + channelName: this.name, + deliveryMode: 'edit', + messageId, + length: text.length, + }, + 'Discord message edited', + ); } catch (err) { - logger.debug({ jid, messageId, err }, 'Failed to edit Discord message'); + logger.debug( + { jid, channelName: this.name, messageId, err }, + 'Failed to edit Discord message', + ); throw err; // Re-throw so callers (e.g. dashboard) can reset message ID } } diff --git a/src/index.ts b/src/index.ts index 48eb6a0..7e1d162 100644 --- a/src/index.ts +++ b/src/index.ts @@ -46,9 +46,9 @@ import { startIpcWatcher } from './ipc.js'; import { findChannel, findChannelByName, - findChannelForDeliveryRole, formatOutbound, normalizeMessageForDedupe, + resolveChannelForDeliveryRole, } from './router.js'; import { buildRestartAnnouncement, @@ -473,9 +473,21 @@ async function main(): Promise { }); startIpcWatcher({ sendMessage: (jid, text, senderRole) => { - const channel = findChannelForDeliveryRole(channels, jid, senderRole); - if (!channel) throw new Error(`No channel for JID: ${jid}`); - return channel.sendMessage(jid, text); + const route = resolveChannelForDeliveryRole(channels, jid, senderRole); + if (!route.channel) throw new Error(`No channel for JID: ${jid}`); + logger.info( + { + transition: 'ipc:route', + chatJid: jid, + senderRole: senderRole ?? null, + requestedRoleChannel: route.requestedRoleChannelName, + selectedChannel: route.selectedChannelName, + usedRoleChannel: route.usedRoleChannel, + fallbackUsed: route.fallbackUsed, + }, + 'IPC relay routed message to channel', + ); + return route.channel.sendMessage(jid, text); }, nudgeScheduler: nudgeSchedulerLoop, registeredGroups: () => registeredGroups, diff --git a/src/ipc-auth.test.ts b/src/ipc-auth.test.ts index dc9f9ec..0d03bc6 100644 --- a/src/ipc-auth.test.ts +++ b/src/ipc-auth.test.ts @@ -540,7 +540,7 @@ describe('IPC message authorization', () => { it('forwards senderRole through authorized IPC messages', async () => { const sendMessage = vi.fn(async () => {}); - await forwardAuthorizedIpcMessage( + const result = await forwardAuthorizedIpcMessage( { type: 'message', chatJid: 'other@g.us', @@ -558,6 +558,12 @@ describe('IPC message authorization', () => { 'review text', 'reviewer', ); + expect(result).toEqual( + expect.objectContaining({ + outcome: 'sent', + senderRole: 'reviewer', + }), + ); }); it('does not forward unauthorized IPC messages even when senderRole exists', async () => { @@ -577,6 +583,7 @@ describe('IPC message authorization', () => { ); expect(result.outcome).toBe('blocked'); + expect(result.senderRole).toBe('reviewer'); expect(sendMessage).not.toHaveBeenCalled(); }); }); diff --git a/src/ipc.ts b/src/ipc.ts index 874a7e2..2e6b567 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -63,6 +63,7 @@ export interface IpcMessageForwardResult { chatJid?: string; targetGroup?: string | null; isMainOverride?: boolean; + senderRole?: string | null; } export async function forwardAuthorizedIpcMessage( @@ -73,7 +74,7 @@ export async function forwardAuthorizedIpcMessage( sendMessage: IpcDeps['sendMessage'], ): Promise { if (!(msg.type === 'message' && msg.chatJid && msg.text)) { - return { outcome: 'ignored' }; + return { outcome: 'ignored', senderRole: msg.senderRole ?? null }; } const targetGroup = registeredGroups[msg.chatJid]; @@ -86,6 +87,7 @@ export async function forwardAuthorizedIpcMessage( chatJid: msg.chatJid, targetGroup: targetGroup?.folder ?? null, isMainOverride, + senderRole: msg.senderRole ?? null, }; } @@ -95,6 +97,7 @@ export async function forwardAuthorizedIpcMessage( chatJid: msg.chatJid, targetGroup: targetGroup?.folder ?? null, isMainOverride, + senderRole: msg.senderRole ?? null, }; } @@ -253,6 +256,7 @@ export function startIpcWatcher(deps: IpcDeps): void { sourceGroup, targetGroup: forwardResult.targetGroup ?? null, isMainOverride: forwardResult.isMainOverride, + senderRole: forwardResult.senderRole ?? null, }, 'IPC message sent', ); @@ -264,6 +268,7 @@ export function startIpcWatcher(deps: IpcDeps): void { sourceGroup, targetGroup: forwardResult.targetGroup ?? null, isMainOverride: forwardResult.isMainOverride, + senderRole: forwardResult.senderRole ?? null, }, 'Unauthorized IPC message attempt blocked', ); diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 3195b82..53b7474 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -214,6 +214,7 @@ import { resolveHandoffRoleOverride, } from './message-runtime.js'; import * as config from './config.js'; +import { logger } from './logger.js'; import * as serviceRouting from './service-routing.js'; import type { Channel, RegisteredGroup } from './types.js'; @@ -628,6 +629,90 @@ describe('createMessageRuntime', () => { expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid); }); + it('suppresses duplicate stale work item delivery and logs the suppression reason', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const enqueueMessageCheck = vi.fn(); + + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + vi.mocked(db.getOpenWorkItem).mockReturnValue({ + id: 199, + group_folder: group.folder, + chat_jid: chatJid, + agent_type: 'codex', + service_id: 'claude', + delivery_role: 'owner', + status: 'delivery_retry', + start_seq: 1, + end_seq: 1, + result_payload: '같은 최종 답변입니다.', + delivery_attempts: 1, + created_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + delivered_at: null, + delivery_message_id: null, + last_error: 'discord send failed', + }); + vi.mocked(db.getLastBotFinalMessage).mockReturnValue([ + { + id: 'last-final', + chat_jid: chatJid, + sender: 'reviewer-bot@test', + sender_name: '리뷰어', + content: '같은 최종 답변입니다.', + timestamp: new Date().toISOString(), + is_from_me: false, + is_bot_message: true, + } as any, + ]); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + enqueueMessageCheck, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-open-work-item-duplicate-suppressed', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendMessage).not.toHaveBeenCalled(); + expect(db.markWorkItemDelivered).toHaveBeenCalledWith(199, null); + expect(agentRunner.runAgentProcess).not.toHaveBeenCalled(); + expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid, + channelName: 'discord', + workItemId: 199, + deliveryRole: 'owner', + suppressionReason: 'paired-final-duplicate', + preview: '같은 최종 답변입니다.', + }), + 'Suppressed duplicate final message in paired room (marked as delivered)', + ); + }); + it('retries a stale reviewer work item when the reviewer channel is missing', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); @@ -1325,6 +1410,17 @@ describe('createMessageRuntime', () => { chatJid, resolveGroupIpcPath(group.folder), ); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid, + taskId: 'task-merge-ready-inline-loop', + taskStatus: 'merge_ready', + handoffMode: 'inline-finalize', + nextRole: 'owner', + cursor: 42, + }), + 'Executing merge_ready finalize turn inline after bot-only reviewer follow-up', + ); }); it('reuses the shared arbiter prompt builder for pending arbiter turns', async () => { diff --git a/src/message-runtime.ts b/src/message-runtime.ts index c3c2251..3a4c323 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -319,6 +319,18 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return isDuplicateOfLastBotFinal(chatJid, text); }; + const buildDeliveryLogContext = ( + channel: Channel, + item: WorkItem, + extra: Record = {}, + ): Record => ({ + chatJid: item.chat_jid, + channelName: channel.name, + workItemId: item.id, + deliveryRole: item.delivery_role ?? null, + ...extra, + }); + const deliverOpenWorkItem = async ( channel: Channel, item: WorkItem, @@ -338,11 +350,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { // Mark as delivered without sending, and don't open continuation markWorkItemDelivered(item.id, null); logger.info( - { - chatJid: item.chat_jid, - workItemId: item.id, + buildDeliveryLogContext(channel, item, { preview: item.result_payload.slice(0, 100), - }, + suppressionReason: 'paired-final-duplicate', + }), 'Suppressed duplicate final message in paired room (marked as delivered)', ); return true; @@ -350,6 +361,14 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { try { if (replaceMessageId && channel.editMessage) { + logger.info( + buildDeliveryLogContext(channel, item, { + deliveryAttempts: item.delivery_attempts + 1, + deliveryMode: 'edit', + replacedMessageId: replaceMessageId, + }), + 'Attempting to deliver produced work item by replacing tracked progress message', + ); await channel.editMessage( item.chat_jid, replaceMessageId, @@ -358,39 +377,43 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { markWorkItemDelivered(item.id, replaceMessageId); continuationTracker.open(item.chat_jid); logger.info( - { - chatJid: item.chat_jid, - workItemId: item.id, + buildDeliveryLogContext(channel, item, { deliveryAttempts: item.delivery_attempts + 1, + deliveryMode: 'edit', replacedMessageId: replaceMessageId, - }, + }), 'Delivered produced work item by replacing tracked progress message', ); return true; } } catch (err) { logger.warn( - { - chatJid: item.chat_jid, - workItemId: item.id, + buildDeliveryLogContext(channel, item, { deliveryAttempts: item.delivery_attempts + 1, + deliveryMode: 'edit', replacedMessageId: replaceMessageId, err, - }, + }), 'Failed to replace tracked progress message; falling back to a new message', ); } try { + logger.info( + buildDeliveryLogContext(channel, item, { + deliveryAttempts: item.delivery_attempts + 1, + deliveryMode: 'send', + }), + 'Attempting to deliver produced work item as a new message', + ); await channel.sendMessage(item.chat_jid, item.result_payload); markWorkItemDelivered(item.id); continuationTracker.open(item.chat_jid); logger.info( - { - chatJid: item.chat_jid, - workItemId: item.id, + buildDeliveryLogContext(channel, item, { deliveryAttempts: item.delivery_attempts + 1, - }, + deliveryMode: 'send', + }), 'Delivered produced work item', ); return true; @@ -398,12 +421,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const errorMessage = getErrorMessage(err); markWorkItemDeliveryRetry(item.id, errorMessage); logger.warn( - { - chatJid: item.chat_jid, - workItemId: item.id, + buildDeliveryLogContext(channel, item, { deliveryAttempts: item.delivery_attempts + 1, + deliveryMode: 'send', err, - }, + }), 'Failed to deliver produced work item', ); return false; @@ -1311,6 +1333,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { groupFolder: group.folder, taskId: loopPendingTask?.id ?? null, taskStatus: loopPendingTask?.status ?? null, + handoffMode: 'inline-finalize', + nextRole: 'owner', + cursor: mergeReadyCursor, }, 'Executing merge_ready finalize turn inline after bot-only reviewer follow-up', ); @@ -1374,6 +1399,16 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { groupFolder: group.folder, taskId: loopPendingTask?.id ?? null, taskStatus: loopPendingTask?.status ?? null, + handoffMode: 'requeue', + nextRole: + loopPendingTask?.status === 'review_ready' + ? 'reviewer' + : loopPendingTask?.status === 'arbiter_requested' || + loopPendingTask?.status === 'in_arbitration' + ? 'arbiter' + : 'reviewer', + cursor: botOnlyPendingTurn.cursor, + cursorKey: botOnlyPendingTurn.cursorKey, }, 'Queued fresh paired pending turn instead of piping bot-only follow-up into the active agent', ); diff --git a/src/router.test.ts b/src/router.test.ts index 6bb6b18..f97b6ca 100644 --- a/src/router.test.ts +++ b/src/router.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it } from 'vitest'; -import { findChannelForDeliveryRole } from './router.js'; +import { + findChannelForDeliveryRole, + resolveChannelForDeliveryRole, +} from './router.js'; import { type Channel } from './types.js'; function createChannel(name: string, ownedJids: string[]): Channel { @@ -51,4 +54,41 @@ describe('findChannelForDeliveryRole', () => { ownerChannel, ); }); + + it('reports fallback metadata when a role-specific channel is unavailable', () => { + const ownerChannel = createChannel('discord-main', [jid]); + + expect( + resolveChannelForDeliveryRole([ownerChannel], jid, 'reviewer'), + ).toEqual( + expect.objectContaining({ + channel: ownerChannel, + requestedRoleChannelName: 'discord-review', + selectedChannelName: 'discord-main', + usedRoleChannel: false, + fallbackUsed: true, + }), + ); + }); + + it('reports direct role routing metadata when the role channel exists', () => { + const ownerChannel = createChannel('discord-main', [jid]); + const reviewerChannel = createChannel('discord-review', [jid]); + + expect( + resolveChannelForDeliveryRole( + [ownerChannel, reviewerChannel], + jid, + 'reviewer', + ), + ).toEqual( + expect.objectContaining({ + channel: reviewerChannel, + requestedRoleChannelName: 'discord-review', + selectedChannelName: 'discord-review', + usedRoleChannel: true, + fallbackUsed: false, + }), + ); + }); }); diff --git a/src/router.ts b/src/router.ts index 1a8eced..5cbc161 100644 --- a/src/router.ts +++ b/src/router.ts @@ -93,24 +93,64 @@ export function findChannel( return channels.find((c) => c.ownsJid(jid)); } +export interface DeliveryRouteResolution { + channel?: Channel; + requestedRoleChannelName: string | null; + selectedChannelName: string | null; + usedRoleChannel: boolean; + fallbackUsed: boolean; +} + +function resolveRequestedRoleChannelName(senderRole?: string): string | null { + if (senderRole === 'reviewer') return 'discord-review'; + if (senderRole === 'arbiter') return 'discord-arbiter'; + return null; +} + +export function resolveChannelForDeliveryRole( + channels: Channel[], + jid: string, + senderRole?: string, +): DeliveryRouteResolution { + const requestedRoleChannelName = resolveRequestedRoleChannelName(senderRole); + if (!requestedRoleChannelName) { + const channel = findChannel(channels, jid); + return { + channel, + requestedRoleChannelName, + selectedChannelName: channel?.name ?? null, + usedRoleChannel: false, + fallbackUsed: false, + }; + } + + const roleChannel = findChannelByName(channels, requestedRoleChannelName); + if (roleChannel) { + return { + channel: roleChannel, + requestedRoleChannelName, + selectedChannelName: roleChannel.name, + usedRoleChannel: true, + fallbackUsed: false, + }; + } + + const fallbackChannel = findChannel(channels, jid); + return { + channel: fallbackChannel, + requestedRoleChannelName, + selectedChannelName: fallbackChannel?.name ?? null, + usedRoleChannel: false, + fallbackUsed: true, + }; +} + export function findChannelForDeliveryRole( channels: Channel[], jid: string, senderRole?: string, ): Channel | undefined { - if (senderRole === 'reviewer') { - return ( - findChannelByName(channels, 'discord-review') || - findChannel(channels, jid) - ); - } - if (senderRole === 'arbiter') { - return ( - findChannelByName(channels, 'discord-arbiter') || - findChannel(channels, jid) - ); - } - return findChannel(channels, jid); + return resolveChannelForDeliveryRole(channels, jid, senderRole).channel; } export function findChannelByName(