From c44b8c51603583c61b80eef96de8bebcdef83d3b Mon Sep 17 00:00:00 2001 From: ejclaw Date: Wed, 29 Apr 2026 22:45:09 +0900 Subject: [PATCH] fix message loop human interjection handling --- src/message-runtime-dispatch.test.ts | 147 +++++++++++++++++++++++++++ src/message-runtime-dispatch.ts | 26 +++++ src/message-runtime-loop.ts | 2 + src/message-runtime.ts | 3 + 4 files changed, 178 insertions(+) create mode 100644 src/message-runtime-dispatch.test.ts diff --git a/src/message-runtime-dispatch.test.ts b/src/message-runtime-dispatch.test.ts new file mode 100644 index 0000000..8c5571f --- /dev/null +++ b/src/message-runtime-dispatch.test.ts @@ -0,0 +1,147 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./db.js', () => ({ + getLastHumanMessageTimestamp: vi.fn(() => null), + getLatestOpenPairedTaskForChat: vi.fn(() => null), + getMessagesSinceSeq: vi.fn(() => []), + getPairedTurnOutputs: vi.fn(() => []), +})); + +vi.mock('./service-routing.js', () => ({ + hasReviewerLease: vi.fn(() => false), + resolveLeaseServiceId: vi.fn(() => null), +})); + +import { getLastHumanMessageTimestamp } from './db.js'; +import { processLoopGroupMessages } from './message-runtime-dispatch.js'; +import { hasReviewerLease } from './service-routing.js'; +import type { Channel, NewMessage, RegisteredGroup } from './types.js'; + +const getLastHumanMessageTimestampMock = + getLastHumanMessageTimestamp as unknown as { + mockReturnValue(value: string | null): void; + }; +const hasReviewerLeaseMock = hasReviewerLease as unknown as { + mockReturnValue(value: boolean): void; +}; + +const chatJid = 'group@test'; +const timestamp = '2026-04-29T13:25:28.000Z'; + +const group: RegisteredGroup = { + name: 'Test Group', + folder: 'test-group', + trigger: '@Andy', + added_at: timestamp, + requiresTrigger: false, +}; + +const channel: Channel = { + name: 'discord', + connect: vi.fn(), + sendMessage: vi.fn(), + isConnected: vi.fn(() => true), + ownsJid: vi.fn(() => true), + disconnect: vi.fn(), +}; + +function message(overrides: Partial): NewMessage { + return { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user', + sender_name: 'User', + content: '중간에 이거 먼저 봐줘', + timestamp, + seq: 10, + is_from_me: false, + is_bot_message: false, + ...overrides, + }; +} + +function defaultArgs( + overrides: Partial[0]> = {}, +) { + return { + chatJid, + group, + groupMessages: [message({})], + channel, + assistantName: '오너', + failureFinalText: 'FAILED', + triggerPattern: /^\/clear\b/, + hasImplicitContinuationWindow: vi.fn(() => false), + lastAgentTimestamps: {}, + saveState: vi.fn(), + timezone: 'Asia/Seoul', + executeTurn: vi.fn(), + schedulePairedFollowUp: vi.fn(() => true), + enqueueMessageCheck: vi.fn(), + sendQueuedMessage: vi.fn(() => true), + closeStdin: vi.fn(), + isRunningMessageTurn: vi.fn(() => true), + labelPairedSenders: vi.fn((_jid, messages) => messages), + formatMessages: vi.fn((messages: NewMessage[]) => + messages.map((item) => item.content).join('\n'), + ), + ...overrides, + } satisfies Parameters[0]; +} + +describe('processLoopGroupMessages', () => { + beforeEach(() => { + vi.clearAllMocks(); + getLastHumanMessageTimestampMock.mockReturnValue(null); + hasReviewerLeaseMock.mockReturnValue(false); + }); + + it('requeues external human messages instead of piping them into an active agent', async () => { + const args = defaultArgs(); + + await processLoopGroupMessages(args); + + expect(args.closeStdin).toHaveBeenCalledWith('human-message-detected'); + expect(args.enqueueMessageCheck).toHaveBeenCalledTimes(1); + expect(args.sendQueuedMessage).not.toHaveBeenCalled(); + expect(args.lastAgentTimestamps).toEqual({}); + expect(args.saveState).not.toHaveBeenCalled(); + }); + + it('does not close stdin for human messages when no message turn is running', async () => { + const args = defaultArgs({ + isRunningMessageTurn: vi.fn(() => false), + }); + + await processLoopGroupMessages(args); + + expect(args.closeStdin).not.toHaveBeenCalled(); + expect(args.enqueueMessageCheck).toHaveBeenCalledTimes(1); + expect(args.sendQueuedMessage).not.toHaveBeenCalled(); + }); + + it('still pipes bot-only messages when active stdin accepts them', async () => { + hasReviewerLeaseMock.mockReturnValue(true); + getLastHumanMessageTimestampMock.mockReturnValue(timestamp); + const args = defaultArgs({ + groupMessages: [ + message({ + id: 'bot-msg-1', + sender: '오너', + sender_name: '오너', + content: 'STEP_DONE', + is_from_me: true, + is_bot_message: true, + seq: 12, + }), + ], + }); + + await processLoopGroupMessages(args); + + expect(args.sendQueuedMessage).toHaveBeenCalledWith(chatJid, 'STEP_DONE'); + expect(args.closeStdin).not.toHaveBeenCalled(); + expect(args.enqueueMessageCheck).not.toHaveBeenCalled(); + expect(args.lastAgentTimestamps).toEqual({ [chatJid]: '12' }); + }); +}); diff --git a/src/message-runtime-dispatch.ts b/src/message-runtime-dispatch.ts index 62b9660..e6b3538 100644 --- a/src/message-runtime-dispatch.ts +++ b/src/message-runtime-dispatch.ts @@ -30,6 +30,12 @@ import type { type RuntimeLog = Pick; +function hasExternalHumanMessage(messages: NewMessage[]): boolean { + return messages.some( + (message) => message.is_from_me !== true && !message.is_bot_message, + ); +} + export function enqueueGenericFollowUpAfterDeliveryRetry(args: { chatJid: string; runId: string; @@ -235,6 +241,7 @@ export async function processLoopGroupMessages(args: { enqueueMessageCheck: () => void; sendQueuedMessage: (chatJid: string, text: string) => boolean; closeStdin: (reason: string) => void; + isRunningMessageTurn: (chatJid: string) => boolean; labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; formatMessages: (messages: NewMessage[], timezone: string) => string; }): Promise { @@ -295,6 +302,25 @@ export async function processLoopGroupMessages(args: { return; } + if (hasExternalHumanMessage(processableGroupMessages)) { + const interruptedActiveRun = args.isRunningMessageTurn(chatJid); + if (interruptedActiveRun) { + args.closeStdin('human-message-detected'); + } + args.enqueueMessageCheck(); + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + messageCount: processableGroupMessages.length, + interruptedActiveRun, + }, + 'Queued human message for a fresh turn instead of piping into active agent', + ); + return; + } + await processQueuedGroupDispatch({ chatJid, group, diff --git a/src/message-runtime-loop.ts b/src/message-runtime-loop.ts index a17c34c..31fd03e 100644 --- a/src/message-runtime-loop.ts +++ b/src/message-runtime-loop.ts @@ -40,6 +40,7 @@ export async function processMessageLoopTick(args: { ) => void; sendQueuedMessage: (chatJid: string, text: string) => boolean; closeStdin: (chatJid: string, reason: string) => void; + isRunningMessageTurn: (chatJid: string) => boolean; labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; }): Promise { args.enqueuePendingHandoffs(); @@ -105,6 +106,7 @@ export async function processMessageLoopTick(args: { args.enqueueScopedGroupMessageCheck(chatJid, group.folder), sendQueuedMessage: args.sendQueuedMessage, closeStdin: (reason) => args.closeStdin(chatJid, reason), + isRunningMessageTurn: args.isRunningMessageTurn, labelPairedSenders: args.labelPairedSenders, formatMessages, }); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 97e8d40..d074b8f 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -536,6 +536,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deps.queue.closeStdin(chatJid, { reason, }), + isRunningMessageTurn: (chatJid) => + deps.queue.getStatuses([chatJid])[0]?.runPhase === + 'running_messages', labelPairedSenders: labelPairedRuntimeSenders, }); } catch (err) {