From 6502eb65868453c6092a0ac65294fedb27406b54 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Wed, 29 Apr 2026 11:42:40 +0900 Subject: [PATCH] Extract message runtime final delivery --- src/message-runtime-final-delivery.test.ts | 216 +++++++++++++++++++++ src/message-runtime-final-delivery.ts | 73 +++++++ src/message-runtime.ts | 51 ++--- 3 files changed, 303 insertions(+), 37 deletions(-) create mode 100644 src/message-runtime-final-delivery.test.ts create mode 100644 src/message-runtime-final-delivery.ts diff --git a/src/message-runtime-final-delivery.test.ts b/src/message-runtime-final-delivery.test.ts new file mode 100644 index 0000000..e12718a --- /dev/null +++ b/src/message-runtime-final-delivery.test.ts @@ -0,0 +1,216 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./db.js', () => ({ + createProducedWorkItem: vi.fn((args) => ({ + id: 123, + chat_jid: args.chat_jid, + delivery_role: args.delivery_role, + result_payload: args.result_payload, + attachments: args.attachments, + delivery_attempts: 0, + })), +})); + +vi.mock('./logger.js', () => ({ + logger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +vi.mock('./message-runtime-delivery.js', () => ({ + deliverOpenWorkItem: vi.fn(() => Promise.resolve(true)), +})); + +import { createProducedWorkItem } from './db.js'; +import { logger } from './logger.js'; +import { deliverOpenWorkItem } from './message-runtime-delivery.js'; +import { deliverMessageRuntimeFinalText } from './message-runtime-final-delivery.js'; +import type { Channel, RegisteredGroup } from './types.js'; + +function makeChannel(): Channel { + return { + name: 'discord-owner', + connect: vi.fn(), + sendMessage: vi.fn(), + isConnected: vi.fn(() => true), + ownsJid: vi.fn(() => true), + disconnect: vi.fn(), + } as unknown as Channel; +} + +function makeGroup(overrides: Partial = {}): RegisteredGroup { + return { + name: 'Room', + folder: 'room-folder', + added_at: '2026-04-29T01:00:00.000Z', + ...overrides, + }; +} + +describe('message-runtime-final-delivery', () => { + beforeEach(() => { + vi.resetAllMocks(); + vi.mocked(deliverOpenWorkItem).mockResolvedValue(true); + vi.mocked(createProducedWorkItem).mockImplementation( + (args: any) => + ({ + id: 123, + chat_jid: args.chat_jid, + delivery_role: args.delivery_role, + result_payload: args.result_payload, + attachments: args.attachments, + delivery_attempts: 0, + }) as any, + ); + }); + + it.each(['reviewer', 'arbiter'] as const)( + 'skips %s work item delivery after direct terminal IPC delivery', + async (deliveryRole) => { + const hasDirectTerminalDeliveryForRun = vi.fn(() => true); + + const result = await deliverMessageRuntimeFinalText({ + text: `${deliveryRole} final`, + chatJid: 'chat-1', + runId: 'run-1', + channel: makeChannel(), + group: makeGroup(), + startSeq: 10, + endSeq: 11, + deliveryRole, + deliveryServiceId: 'codex-main', + hasDirectTerminalDeliveryForRun, + isDuplicateOfLastBotFinal: vi.fn(), + openContinuation: vi.fn(), + }); + + expect(result).toBe(true); + expect(hasDirectTerminalDeliveryForRun).toHaveBeenCalledWith( + 'chat-1', + 'run-1', + deliveryRole, + ); + expect(createProducedWorkItem).not.toHaveBeenCalled(); + expect(deliverOpenWorkItem).not.toHaveBeenCalled(); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid: 'chat-1', + runId: 'run-1', + deliveryRole, + }), + 'Skipping final work item delivery because this run already sent a direct terminal IPC message', + ); + }, + ); + + it('creates and delivers a produced work item with default agent type and attachment roots', async () => { + const channel = makeChannel(); + const isDuplicateOfLastBotFinal = vi.fn(() => false); + const openContinuation = vi.fn(); + const attachments = [ + { + path: '/tmp/image.png', + name: 'image.png', + mime: 'image/png', + }, + ]; + + const result = await deliverMessageRuntimeFinalText({ + text: 'owner final', + attachments, + chatJid: 'chat-1', + runId: 'run-owner', + channel, + group: makeGroup({ workDir: '/work/room' }), + startSeq: 1, + endSeq: 2, + deliveryRole: 'owner', + deliveryServiceId: null, + hasDirectTerminalDeliveryForRun: vi.fn(() => true), + isDuplicateOfLastBotFinal, + openContinuation, + }); + + expect(result).toBe(true); + expect(createProducedWorkItem).toHaveBeenCalledWith({ + group_folder: 'room-folder', + chat_jid: 'chat-1', + agent_type: 'claude-code', + service_id: undefined, + delivery_role: 'owner', + start_seq: 1, + end_seq: 2, + result_payload: 'owner final', + attachments, + }); + expect(deliverOpenWorkItem).toHaveBeenCalledWith( + expect.objectContaining({ + channel, + log: logger, + attachmentBaseDirs: ['/work/room'], + replaceMessageId: undefined, + isDuplicateOfLastBotFinal, + openContinuation, + }), + ); + }); + + it('preserves forced agent, delivery service, and replacement message metadata', async () => { + const channel = makeChannel(); + + await deliverMessageRuntimeFinalText({ + text: 'forced final', + chatJid: 'chat-1', + runId: 'run-forced', + channel, + group: makeGroup({ agentType: 'claude-code' }), + startSeq: null, + endSeq: null, + forcedAgentType: 'codex', + deliveryRole: 'arbiter', + deliveryServiceId: 'codex-main', + replaceMessageId: 'discord-message-1', + hasDirectTerminalDeliveryForRun: vi.fn(() => false), + isDuplicateOfLastBotFinal: vi.fn(), + openContinuation: vi.fn(), + }); + + expect(createProducedWorkItem).toHaveBeenCalledWith( + expect.objectContaining({ + agent_type: 'codex', + service_id: 'codex-main', + delivery_role: 'arbiter', + start_seq: null, + end_seq: null, + }), + ); + expect(deliverOpenWorkItem).toHaveBeenCalledWith( + expect.objectContaining({ + replaceMessageId: 'discord-message-1', + attachmentBaseDirs: undefined, + }), + ); + }); + + it('returns the deliverOpenWorkItem result', async () => { + vi.mocked(deliverOpenWorkItem).mockResolvedValue(false); + + await expect( + deliverMessageRuntimeFinalText({ + text: 'failed final', + chatJid: 'chat-1', + runId: 'run-failed', + channel: makeChannel(), + group: makeGroup(), + startSeq: 1, + endSeq: 2, + deliveryRole: null, + deliveryServiceId: null, + isDuplicateOfLastBotFinal: vi.fn(), + openContinuation: vi.fn(), + }), + ).resolves.toBe(false); + }); +}); diff --git a/src/message-runtime-final-delivery.ts b/src/message-runtime-final-delivery.ts new file mode 100644 index 0000000..977044d --- /dev/null +++ b/src/message-runtime-final-delivery.ts @@ -0,0 +1,73 @@ +import { createProducedWorkItem } from './db.js'; +import { logger } from './logger.js'; +import { deliverOpenWorkItem } from './message-runtime-delivery.js'; +import type { + AgentType, + Channel, + OutboundAttachment, + PairedRoomRole, + RegisteredGroup, +} from './types.js'; + +export async function deliverMessageRuntimeFinalText(args: { + text: string; + attachments?: OutboundAttachment[]; + chatJid: string; + runId: string; + channel: Channel; + group: RegisteredGroup; + startSeq: number | null; + endSeq: number | null; + forcedAgentType?: AgentType; + deliveryRole: PairedRoomRole | null; + deliveryServiceId: string | null; + replaceMessageId?: string | null; + hasDirectTerminalDeliveryForRun?: ( + chatJid: string, + runId: string, + deliveryRole: PairedRoomRole | null, + ) => boolean; + isDuplicateOfLastBotFinal: (chatJid: string, text: string) => boolean; + openContinuation: (chatJid: string) => void; +}): Promise { + if ( + (args.deliveryRole === 'reviewer' || args.deliveryRole === 'arbiter') && + args.hasDirectTerminalDeliveryForRun?.( + args.chatJid, + args.runId, + args.deliveryRole, + ) + ) { + logger.info( + { + chatJid: args.chatJid, + runId: args.runId, + deliveryRole: args.deliveryRole, + }, + 'Skipping final work item delivery because this run already sent a direct terminal IPC message', + ); + return true; + } + + const workItem = createProducedWorkItem({ + group_folder: args.group.folder, + chat_jid: args.chatJid, + agent_type: args.forcedAgentType ?? args.group.agentType ?? 'claude-code', + service_id: args.deliveryServiceId ?? undefined, + delivery_role: args.deliveryRole, + start_seq: args.startSeq, + end_seq: args.endSeq, + result_payload: args.text, + attachments: args.attachments, + }); + + return deliverOpenWorkItem({ + channel: args.channel, + item: workItem, + log: logger, + attachmentBaseDirs: args.group.workDir ? [args.group.workDir] : undefined, + replaceMessageId: args.replaceMessageId, + isDuplicateOfLastBotFinal: args.isDuplicateOfLastBotFinal, + openContinuation: args.openContinuation, + }); +} diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 23a5b4a..f128588 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1,5 +1,4 @@ import { - createProducedWorkItem, getOpenWorkItemForChat, getMessagesSinceSeq, getLatestOpenPairedTaskForChat, @@ -15,10 +14,7 @@ import { GroupQueue, GroupRunContext } from './group-queue.js'; import { findChannel, formatMessages } from './router.js'; import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; import { enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp } from './message-runtime-dispatch.js'; -import { - deliverOpenWorkItem, - processOpenWorkItemDelivery, -} from './message-runtime-delivery.js'; +import { processOpenWorkItemDelivery } from './message-runtime-delivery.js'; import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js'; import { handleQueuedRunGates } from './message-runtime-gating.js'; import { @@ -74,6 +70,7 @@ import { hasHumanMessageAfterWorkItem, } from './message-runtime-preflight-messages.js'; import { handleMessageRuntimeAfterDeliverySuccess } from './message-runtime-after-delivery.js'; +import { deliverMessageRuntimeFinalText } from './message-runtime-final-delivery.js'; export { isDuplicateOfLastBotFinal }; @@ -191,41 +188,21 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deliveryServiceId, replaceMessageId, }) => { - if ( - (deliveryRole === 'reviewer' || deliveryRole === 'arbiter') && - deps.queue.hasDirectTerminalDeliveryForRun?.( - chatJid, - runId, - deliveryRole, - ) - ) { - logger.info( - { - chatJid, - runId, - deliveryRole, - }, - 'Skipping final work item delivery because this run already sent a direct terminal IPC message', - ); - return true; - } - const workItem = createProducedWorkItem({ - group_folder: group.folder, - chat_jid: chatJid, - agent_type: forcedAgentType ?? group.agentType ?? 'claude-code', - service_id: deliveryServiceId ?? undefined, - delivery_role: deliveryRole, - start_seq: startSeq, - end_seq: endSeq, - result_payload: text, + return deliverMessageRuntimeFinalText({ + text, attachments, - }); - return deliverOpenWorkItem({ + chatJid, + runId, channel, - item: workItem, - log: logger, - attachmentBaseDirs: group.workDir ? [group.workDir] : undefined, + group, + startSeq, + endSeq, + forcedAgentType, + deliveryRole, + deliveryServiceId, replaceMessageId, + hasDirectTerminalDeliveryForRun: + deps.queue.hasDirectTerminalDeliveryForRun?.bind(deps.queue), isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal, openContinuation: (targetChatJid) => continuationTracker.open(targetChatJid),