fix message loop human interjection handling

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>): 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<Parameters<typeof processLoopGroupMessages>[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<typeof processLoopGroupMessages>[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' });
+  });
+});

src/message-runtime-dispatch.ts

@@ -30,6 +30,12 @@ import type {
 
 type RuntimeLog = Pick<typeof logger, 'info' | 'debug'>;
 
+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<void> {
@@ -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,

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<void> {
   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,
     });

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) {