Fix reviewer-to-owner follow-up race

src/message-agent-executor.test.ts

@@ -1315,6 +1315,107 @@ describe('runAgentForGroup room memory', () => {
     expect(deps.queue.enqueueMessageCheck).toHaveBeenCalledWith('group@test');
   });
 
+  it('stores reviewer turn output before transitioning the paired task back to active', async () => {
+    const group = { ...makeGroup(), folder: 'test-group' };
+    const deps = makeDeps();
+
+    vi.mocked(serviceRouting.getEffectiveChannelLease).mockReturnValue({
+      chat_jid: 'group@test',
+      owner_agent_type: 'codex',
+      reviewer_agent_type: 'claude-code',
+      arbiter_agent_type: null,
+      owner_service_id: 'codex-main',
+      reviewer_service_id: 'claude',
+      arbiter_service_id: null,
+      activated_at: null,
+      reason: null,
+      explicit: false,
+    });
+    vi.mocked(
+      pairedExecutionContext.preparePairedExecutionContext,
+    ).mockReturnValue({
+      task: {
+        id: 'paired-task-reviewer-output-order',
+        chat_jid: 'group@test',
+        group_folder: 'test-group',
+        owner_service_id: 'codex-main',
+        reviewer_service_id: 'claude',
+        title: null,
+        source_ref: 'HEAD',
+        plan_notes: null,
+        round_trip_count: 1,
+        review_requested_at: '2026-03-31T00:00:00.000Z',
+        status: 'in_review',
+        arbiter_verdict: null,
+        arbiter_requested_at: null,
+        completion_reason: null,
+        created_at: '2026-03-31T00:00:00.000Z',
+        updated_at: '2026-03-31T00:00:00.000Z',
+      },
+      workspace: null,
+      envOverrides: {},
+    });
+    vi.mocked(db.getPairedTaskById).mockReturnValue({
+      id: 'paired-task-reviewer-output-order',
+      chat_jid: 'group@test',
+      group_folder: 'test-group',
+      owner_service_id: 'codex-main',
+      reviewer_service_id: 'claude',
+      title: null,
+      source_ref: 'HEAD',
+      plan_notes: null,
+      round_trip_count: 1,
+      review_requested_at: '2026-03-31T00:00:00.000Z',
+      status: 'active',
+      arbiter_verdict: null,
+      arbiter_requested_at: null,
+      completion_reason: null,
+      created_at: '2026-03-31T00:00:00.000Z',
+      updated_at: '2026-03-31T00:00:00.000Z',
+    });
+    vi.mocked(agentRunner.runAgentProcess).mockImplementation(
+      async (_group, _input, _onProcess, onOutput) => {
+        await onOutput?.({
+          status: 'success',
+          result: 'DONE_WITH_CONCERNS\nreviewer feedback',
+          output: {
+            visibility: 'public',
+            text: 'DONE_WITH_CONCERNS\nreviewer feedback',
+          },
+          phase: 'final',
+        });
+        return {
+          status: 'success',
+          result: 'DONE_WITH_CONCERNS\nreviewer feedback',
+        };
+      },
+    );
+
+    const result = await runAgentForGroup(deps, {
+      group,
+      prompt: 'please review',
+      chatJid: 'group@test',
+      runId: 'run-reviewer-output-order',
+      forcedRole: 'reviewer',
+      onOutput: async () => {},
+    });
+
+    expect(result).toBe('success');
+    expect(db.insertPairedTurnOutput).toHaveBeenCalledWith(
+      'paired-task-reviewer-output-order',
+      1,
+      'reviewer',
+      'DONE_WITH_CONCERNS\nreviewer feedback',
+    );
+    expect(
+      vi.mocked(db.insertPairedTurnOutput).mock.invocationCallOrder[0],
+    ).toBeLessThan(
+      vi.mocked(
+        pairedExecutionContext.completePairedExecutionContext,
+      ).mock.invocationCallOrder[0],
+    );
+  });
+
   it('does not enqueue a generic follow-up when reviewer output already returned the task to active', async () => {
     const group = { ...makeGroup(), folder: 'test-group' };
     const deps = makeDeps();

src/message-agent-executor.ts

@@ -939,13 +939,6 @@ export async function runAgentForGroup(
         !pairedSawOutput
           ? 'failed'
           : completionStatus;
-      completePairedExecutionContext({
-        taskId: pairedExecutionContext.task.id,
-        role: completedRole,
-        status: effectiveStatus,
-        summary: pairedExecutionSummary,
-      });
-
       // Store full output for direct inter-agent data passing (Discord-independent).
       if (pairedFullOutput && effectiveStatus === 'succeeded') {
         try {
@@ -964,6 +957,13 @@ export async function runAgentForGroup(
           );
         }
       }
+
+      completePairedExecutionContext({
+        taskId: pairedExecutionContext.task.id,
+        role: completedRole,
+        status: effectiveStatus,
+        summary: pairedExecutionSummary,
+      });
     }
 
     // Notify user when paired task reaches a terminal state that requires attention.

src/message-runtime-flow.ts

@@ -18,6 +18,7 @@ import {
 import { hasReviewerLease } from './service-routing.js';
 import type {
   Channel,
+  NewMessage,
   PairedTask,
   PairedRoomRole,
   RegisteredGroup,
@@ -46,6 +47,25 @@ export type BotOnlyPairedFollowUpAction =
       nextRole: 'owner' | 'reviewer' | 'arbiter';
     };
 
+export function resolveLastDeliveredBotRole(
+  messages: NewMessage[],
+): PairedRoomRole | null {
+  for (let index = messages.length - 1; index >= 0; index -= 1) {
+    const message = messages[index];
+    if (!message?.is_bot_message) {
+      continue;
+    }
+    if (
+      message.sender_name === 'owner' ||
+      message.sender_name === 'reviewer' ||
+      message.sender_name === 'arbiter'
+    ) {
+      return message.sender_name;
+    }
+  }
+  return null;
+}
+
 export function buildPendingPairedTurn(args: {
   chatJid: string;
   timezone: string;
@@ -57,6 +77,7 @@ export function buildPendingPairedTurn(args: {
   labeledRecentMessages: Parameters<
     typeof buildArbiterPromptForTask
   >[0]['labeledRecentMessages'];
+  lastDeliveredMessages?: NewMessage[];
   resolveChannel: (taskStatus?: string | null) => Channel | null;
 }): PendingPairedTurn {
   const {
@@ -75,7 +96,10 @@ export function buildPendingPairedTurn(args: {
   const lastTurnOutput = turnOutputs[turnOutputs.length - 1];
   const nextTurnAction = resolveNextTurnAction({
     taskStatus,
-    lastTurnOutputRole: lastTurnOutput?.role ?? null,
+    lastTurnOutputRole:
+      resolveLastDeliveredBotRole(args.lastDeliveredMessages ?? []) ??
+      lastTurnOutput?.role ??
+      null,
   });
   const recentMessages = getRecentChatMessages(chatJid, 20);
   const lastHumanMessage = getLastHumanMessageContent(chatJid);
@@ -210,6 +234,7 @@ export function resolveBotOnlyPairedFollowUpAction(args: {
   chatJid: string;
   task: PairedTask | null | undefined;
   isBotOnlyPairedFollowUp: boolean;
+  lastDeliveredMessages?: NewMessage[];
   pendingCursorSource:
     | { seq?: number | null; timestamp?: string | null }
     | undefined;
@@ -224,7 +249,10 @@ export function resolveBotOnlyPairedFollowUpAction(args: {
   const lastTurnOutput = getPairedTurnOutputs(task.id).at(-1);
   const nextTurnAction = resolveNextTurnAction({
     taskStatus: task.status,
-    lastTurnOutputRole: lastTurnOutput?.role ?? null,
+    lastTurnOutputRole:
+      resolveLastDeliveredBotRole(args.lastDeliveredMessages ?? []) ??
+      lastTurnOutput?.role ??
+      null,
   });
 
   if (nextTurnAction.kind === 'finalize-owner-turn') {

src/message-runtime.test.ts

@@ -1827,6 +1827,124 @@ describe('createMessageRuntime', () => {
     expect(sendMessage).not.toHaveBeenCalled();
   });
 
+  it('requeues owner follow-ups from reviewer bot-only messages even when paired turn output storage is stale', async () => {
+    const chatJid = 'group@test';
+    const group = makeGroup('codex');
+    const ownerChannel = makeChannel(chatJid);
+    const reviewerChannel = {
+      ...makeChannel(chatJid, 'discord-review', false),
+      isOwnMessage: vi.fn((message) => message.sender === 'reviewer-bot@test'),
+    };
+    const enqueueMessageCheck = vi.fn();
+    const closeStdin = vi.fn();
+    const sendMessage = vi.fn(() => false);
+    const setLastTimestamp = vi.fn();
+    const saveState = vi.fn();
+    const lastAgentTimestamps: Record<string, string> = {};
+    const stopLoop = new Error('stop-message-loop');
+
+    vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true);
+    vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
+      id: 'task-active-owner-follow-up-stale-output',
+      chat_jid: chatJid,
+      group_folder: group.folder,
+      owner_service_id: 'codex-main',
+      reviewer_service_id: 'claude',
+      title: null,
+      source_ref: 'HEAD',
+      plan_notes: null,
+      review_requested_at: '2026-03-30T00:00:00.000Z',
+      round_trip_count: 1,
+      status: 'active',
+      arbiter_verdict: null,
+      arbiter_requested_at: null,
+      completion_reason: null,
+      created_at: '2026-03-30T00:00:00.000Z',
+      updated_at: '2026-03-30T00:00:00.000Z',
+    });
+    vi.mocked(db.getPairedTurnOutputs).mockReturnValue([
+      {
+        id: 1,
+        task_id: 'task-active-owner-follow-up-stale-output',
+        turn_number: 1,
+        role: 'owner',
+        output_text: 'owner 초안',
+        created_at: '2026-03-30T00:00:01.000Z',
+      },
+    ]);
+    vi.mocked(db.getNewMessages).mockReturnValue({
+      messages: [
+        {
+          id: 'reviewer-bot-message-owner-follow-up-stale-output',
+          chat_jid: chatJid,
+          sender: 'reviewer-bot@test',
+          sender_name: '리뷰어',
+          content: 'DONE_WITH_CONCERNS\n\nreviewer direct message',
+          timestamp: '2026-03-30T00:00:04.000Z',
+          seq: 42,
+          is_bot_message: true,
+        } as any,
+      ],
+      newTimestamp: '42',
+    });
+
+    const runtime = createMessageRuntime({
+      assistantName: 'Andy',
+      idleTimeout: 60_000,
+      pollInterval: 123,
+      timezone: 'UTC',
+      triggerPattern: /^@Andy\b/i,
+      channels: [ownerChannel, reviewerChannel],
+      queue: {
+        registerProcess: vi.fn(),
+        closeStdin,
+        enqueueMessageCheck,
+        notifyIdle: vi.fn(),
+        sendMessage,
+      } as any,
+      getRegisteredGroups: () => ({ [chatJid]: group }),
+      getSessions: () => ({}),
+      getLastTimestamp: () => '',
+      setLastTimestamp,
+      getLastAgentTimestamps: () => lastAgentTimestamps,
+      saveState,
+      persistSession: vi.fn(),
+      clearSession: vi.fn(),
+    });
+
+    const originalSetTimeout = global.setTimeout;
+    const setTimeoutSpy = vi.spyOn(global, 'setTimeout').mockImplementation(((
+      handler: any,
+      timeout?: number,
+      ...args: any[]
+    ) => {
+      if (timeout === 123) {
+        throw stopLoop;
+      }
+      return (originalSetTimeout as any)(handler, timeout, ...args);
+    }) as typeof setTimeout);
+
+    try {
+      await expect(runtime.startMessageLoop()).rejects.toThrow(
+        stopLoop.message,
+      );
+    } finally {
+      setTimeoutSpy.mockRestore();
+    }
+
+    expect(setLastTimestamp).toHaveBeenCalledWith('42');
+    expect(agentRunner.runAgentProcess).not.toHaveBeenCalled();
+    expect(closeStdin).toHaveBeenCalledWith(
+      chatJid,
+      expect.objectContaining({ reason: 'paired-pending-turn-follow-up' }),
+    );
+    expect(enqueueMessageCheck).toHaveBeenCalledWith(
+      chatJid,
+      resolveGroupIpcPath(group.folder),
+    );
+    expect(sendMessage).not.toHaveBeenCalled();
+  });
+
   it('auto-runs an owner follow-up when a task returns to active after reviewer feedback', async () => {
     const chatJid = 'group@test';
     const group = makeGroup('codex');

src/message-runtime.ts

@@ -47,6 +47,7 @@ import {
   executeBotOnlyPairedFollowUpAction,
   executePendingPairedTurn,
   resolveBotOnlyPairedFollowUpAction,
+  resolveLastDeliveredBotRole,
   shouldSkipGenericFollowUpAfterDeliveryRetry,
 } from './message-runtime-flow.js';
 import { MessageTurnController } from './message-turn-controller.js';
@@ -858,6 +859,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
                 chatJid,
                 getRecentChatMessages(chatJid, 20),
               ),
+              lastDeliveredMessages: labelPairedSenders(chatJid, missedMessages),
               resolveChannel,
             })
           : null;
@@ -1217,22 +1219,20 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
               pendingMessages.length > 0
                 ? pendingMessages
                 : processableGroupMessages;
+            const labeledMessagesToSend = labelPairedSenders(
+              chatJid,
+              messagesToSend,
+            );
             const formatted = loopPendingTask
               ? buildPairedTurnPrompt({
                   taskId: loopPendingTask.id,
                   chatJid,
                   timezone: deps.timezone,
                   missedMessages: messagesToSend,
-                  labeledFallbackMessages: labelPairedSenders(
+                  labeledFallbackMessages: labeledMessagesToSend,
-                    chatJid,
-                    messagesToSend,
-                  ),
                   turnOutputs: getPairedTurnOutputs(loopPendingTask.id),
                 })
-              : formatMessages(
+              : formatMessages(labeledMessagesToSend, deps.timezone);
-                  labelPairedSenders(chatJid, messagesToSend),
-                  deps.timezone,
-                );
             const isBotOnlyPairedFollowUp = isBotOnlyPairedRoomTurn(
               chatJid,
               messagesToSend,
@@ -1245,6 +1245,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
               chatJid,
               task: loopPendingTask,
               isBotOnlyPairedFollowUp,
+              lastDeliveredMessages: labeledMessagesToSend,
               pendingCursorSource,
             });
             if (