Stabilize paired follow-up routing

src/message-agent-executor-paired.ts

@@ -216,7 +216,7 @@ export function createPairedExecutionLifecycle(args: {
 
       const nextTurnAction = resolveNextTurnAction({
         taskStatus: finishedTask.status,
-        lastTurnOutputRole: null,
+        lastTurnOutputRole: pairedSawOutput ? completedRole : null,
       });
       if (nextTurnAction.kind === 'none') {
         return;
@@ -240,7 +240,7 @@ export function createPairedExecutionLifecycle(args: {
         },
         scheduled
           ? 'Queued paired follow-up after failed reviewer/arbiter execution left a pending task state'
-          : 'Skipped duplicate paired follow-up after failed reviewer/arbiter execution in the same run',
+          : 'Skipped duplicate paired follow-up after failed reviewer/arbiter execution while task state was unchanged',
       );
     },
   };

src/message-agent-executor-rules.ts

@@ -23,10 +23,7 @@ export function resolvePairedFollowUpQueueAction(args: {
 }): PairedFollowUpQueueAction {
   const nextTurnAction = resolveNextTurnAction({
     taskStatus: args.taskStatus,
-    lastTurnOutputRole:
+    lastTurnOutputRole: args.sawOutput ? args.completedRole : null,
-      args.executionStatus === 'succeeded' && args.sawOutput
-        ? args.completedRole
-        : null,
   });
   const dispatch = resolveFollowUpDispatch({
     source: 'executor-recovery',

src/message-agent-executor.test.ts

@@ -1866,7 +1866,7 @@ describe('runAgentForGroup room memory', () => {
         intentKind: 'reviewer-turn',
         scheduled: false,
       }),
-      'Skipped duplicate paired follow-up after failed reviewer/arbiter execution in the same run',
+      'Skipped duplicate paired follow-up after failed reviewer/arbiter execution while task state was unchanged',
     );
   });
 });

src/message-runtime-dispatch.ts

@@ -100,7 +100,7 @@ export function enqueueGenericFollowUpAfterDeliveryRetry(args: {
         taskStatus: args.pendingTask?.status ?? null,
         intentKind: nextTurnAction.kind,
       },
-      'Skipped duplicate paired follow-up enqueue after delivery retry in the same run',
+      'Skipped duplicate paired follow-up enqueue after delivery retry while task state was unchanged',
     );
   }
 }

src/message-runtime-flow.test.ts

@@ -114,7 +114,7 @@ describe('executeBotOnlyPairedFollowUpAction', () => {
         intentKind: 'owner-follow-up',
         scheduled: false,
       }),
-      'Skipped duplicate paired pending turn requeue in the same run',
+      'Skipped duplicate paired pending turn requeue while task state was unchanged',
     );
   });
 });

src/message-runtime-flow.ts

@@ -418,7 +418,7 @@ export async function executeBotOnlyPairedFollowUpAction(args: {
     },
     scheduled
       ? 'Queued fresh paired pending turn instead of piping bot-only follow-up into the active agent'
-      : 'Skipped duplicate paired pending turn requeue in the same run',
+      : 'Skipped duplicate paired pending turn requeue while task state was unchanged',
   );
   return true;
 }

src/message-runtime-rules.test.ts

@@ -153,7 +153,7 @@ describe('message-runtime-rules', () => {
     });
   });
 
-  it('routes fresh human input to owner even while review is pending', () => {
+  it('routes fresh human input to owner while review is pending or running', () => {
     expect(
       resolveQueuedTurnRole({
         taskStatus: 'review_ready',
@@ -165,7 +165,7 @@ describe('message-runtime-rules', () => {
         taskStatus: 'in_review',
         hasHumanMessage: true,
       }),
-    ).toBe('reviewer');
+    ).toBe('owner');
     expect(
       resolveQueuedTurnRole({
         taskStatus: 'review_ready',

src/message-runtime-rules.ts

@@ -65,7 +65,10 @@ export function resolveQueuedTurnRole(args: {
   taskStatus?: PairedTaskStatus | null;
   hasHumanMessage: boolean;
 }): 'owner' | 'reviewer' | 'arbiter' {
-  if (args.hasHumanMessage && args.taskStatus === 'review_ready') {
+  if (
+    args.hasHumanMessage &&
+    (args.taskStatus === 'review_ready' || args.taskStatus === 'in_review')
+  ) {
     return 'owner';
   }
 

src/message-runtime.test.ts

@@ -340,6 +340,7 @@ describe('createMessageRuntime', () => {
       triggerPattern: /^@Andy\b/i,
       channels: [channel],
       queue: {
+        enqueueMessageCheck: vi.fn(),
         registerProcess: vi.fn(),
         closeStdin: vi.fn(),
         notifyIdle: vi.fn(),
@@ -409,6 +410,7 @@ describe('createMessageRuntime', () => {
       triggerPattern: /^@Andy\b/i,
       channels: [channel],
       queue: {
+        enqueueMessageCheck: vi.fn(),
         registerProcess: vi.fn(),
         closeStdin: vi.fn(),
         notifyIdle: vi.fn(),
@@ -1914,7 +1916,116 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
         taskStatus: 'review_ready',
         scheduled: false,
       }),
-      'Skipped duplicate paired follow-up after successful owner delivery in the same run',
+      'Skipped duplicate paired follow-up after successful owner delivery while task state was unchanged',
+    );
+  });
+
+  it('does not enqueue the same reviewer follow-up twice across different runs while task state is unchanged', async () => {
+    const chatJid = 'group@test';
+    const group = makeGroup('codex');
+    const ownerChannel: Channel = {
+      ...makeChannel(chatJid),
+      isOwnMessage: vi.fn((msg) => msg.sender === 'owner-bot@test'),
+    };
+    const reviewerChannel = makeChannel(chatJid, 'discord-review', false);
+    const enqueueMessageCheck = vi.fn();
+    const pairedTask = {
+      id: 'task-owner-delivery-dedup-across-runs',
+      chat_jid: chatJid,
+      group_folder: group.folder,
+      owner_service_id: 'claude',
+      reviewer_service_id: 'codex-review',
+      title: null,
+      source_ref: 'HEAD',
+      plan_notes: null,
+      review_requested_at: null,
+      round_trip_count: 0,
+      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',
+    } as any;
+
+    vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true);
+    vi.mocked(db.getLatestOpenPairedTaskForChat).mockImplementation(
+      () => pairedTask,
+    );
+    vi.mocked(db.getMessagesSince).mockReturnValue([
+      {
+        id: 'human-dedup-across-runs-1',
+        chat_jid: chatJid,
+        sender: 'user@test',
+        sender_name: 'User',
+        content: '이 구현 진행해줘',
+        timestamp: '2026-03-30T00:00:00.000Z',
+        seq: 1,
+        is_bot_message: false,
+      } as any,
+    ]);
+    vi.mocked(agentRunner.runAgentProcess).mockImplementation(
+      async (_group, _input, _onProcess, onOutput) => {
+        pairedTask.status = 'review_ready';
+        pairedTask.review_requested_at = '2026-03-30T00:00:01.000Z';
+        pairedTask.round_trip_count = 1;
+        await onOutput?.({
+          status: 'success',
+          phase: 'final',
+          result: 'DONE_WITH_CONCERNS\nowner complete',
+          newSessionId: 'session-owner-delivery-dedup-across-runs',
+        });
+        return {
+          status: 'success',
+          result: 'DONE_WITH_CONCERNS\nowner complete',
+          newSessionId: 'session-owner-delivery-dedup-across-runs',
+        };
+      },
+    );
+
+    const runtime = createMessageRuntime({
+      assistantName: 'Andy',
+      idleTimeout: 1_000,
+      pollInterval: 1_000,
+      timezone: 'UTC',
+      triggerPattern: /^@Andy\b/i,
+      channels: [ownerChannel, reviewerChannel],
+      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(),
+    });
+
+    await runtime.processGroupMessages(chatJid, {
+      runId: 'run-owner-delivery-dedup-1',
+      reason: 'messages',
+    });
+    await runtime.processGroupMessages(chatJid, {
+      runId: 'run-owner-delivery-dedup-2',
+      reason: 'drain',
+    });
+
+    expect(enqueueMessageCheck).toHaveBeenCalledTimes(1);
+    expect(logger.info).toHaveBeenCalledWith(
+      expect.objectContaining({
+        chatJid,
+        runId: 'run-owner-delivery-dedup-2',
+        completedRole: 'owner',
+        taskId: 'task-owner-delivery-dedup-across-runs',
+        taskStatus: 'review_ready',
+        scheduled: false,
+      }),
+      'Skipped duplicate paired follow-up after successful owner delivery while task state was unchanged',
     );
   });
 
@@ -2007,6 +2118,95 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
     expect(reviewerChannel.sendMessage).not.toHaveBeenCalled();
   });
 
+  it('routes fresh human input to owner even when the latest task is in_review', async () => {
+    const chatJid = 'group@test';
+    const group = makeGroup('codex');
+    const ownerChannel = makeChannel(chatJid);
+    const reviewerChannel = makeChannel(chatJid, 'discord-review', false);
+
+    vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true);
+    vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
+      id: 'task-human-owner-override-in-review',
+      chat_jid: chatJid,
+      group_folder: group.folder,
+      owner_service_id: 'claude',
+      reviewer_service_id: 'codex-main',
+      title: null,
+      source_ref: 'HEAD',
+      plan_notes: null,
+      review_requested_at: '2026-03-30T00:00:00.000Z',
+      round_trip_count: 1,
+      status: 'in_review',
+      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',
+    } as any);
+    vi.mocked(db.getMessagesSince).mockReturnValue([
+      {
+        id: 'human-override-in-review-1',
+        chat_jid: chatJid,
+        sender: 'user@test',
+        sender_name: 'User',
+        content: '방금 말한 조건도 같이 반영해줘',
+        timestamp: '2026-03-30T00:00:01.000Z',
+        seq: 1,
+        is_bot_message: false,
+      } as any,
+    ]);
+    vi.mocked(agentRunner.runAgentProcess).mockImplementation(
+      async (_group, _input, _onProcess, onOutput) => {
+        await onOutput?.({
+          status: 'success',
+          phase: 'final',
+          result: 'DONE_WITH_CONCERNS\nowner handled fresh in_review input',
+          newSessionId: 'session-human-owner-override-in-review',
+        });
+        return {
+          status: 'success',
+          result: 'DONE_WITH_CONCERNS\nowner handled fresh in_review input',
+          newSessionId: 'session-human-owner-override-in-review',
+        };
+      },
+    );
+
+    const runtime = createMessageRuntime({
+      assistantName: 'Andy',
+      idleTimeout: 1_000,
+      pollInterval: 1_000,
+      timezone: 'UTC',
+      triggerPattern: /^@Andy\b/i,
+      channels: [ownerChannel, reviewerChannel],
+      queue: {
+        registerProcess: vi.fn(),
+        closeStdin: vi.fn(),
+        notifyIdle: vi.fn(),
+        enqueueMessageCheck: vi.fn(),
+      } 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-human-owner-override-in-review',
+      reason: 'messages',
+    });
+
+    expect(result).toBe(true);
+    expect(ownerChannel.sendMessage).toHaveBeenCalledWith(
+      chatJid,
+      'DONE_WITH_CONCERNS\nowner handled fresh in_review input',
+    );
+    expect(reviewerChannel.sendMessage).not.toHaveBeenCalled();
+  });
+
   it('consumes stale bot-only owner messages once the finalize turn output is already persisted', () => {
     const chatJid = 'group@test';
     const group = makeGroup('codex');
@@ -2696,7 +2896,7 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
     );
   });
 
-  it('fails closed for in-review turns when the reviewer channel is missing', async () => {
+  it('fails closed for in-review turns without fresh human input when the reviewer channel is missing', async () => {
     const chatJid = 'group@test';
     const group = makeGroup('codex');
     const ownerChannel = makeChannel(chatJid);
@@ -2722,18 +2922,7 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
       created_at: '2026-03-30T00:00:00.000Z',
       updated_at: '2026-03-30T00:00:00.000Z',
     });
-    vi.mocked(db.getMessagesSince).mockReturnValue([
+    vi.mocked(db.getMessagesSince).mockReturnValue([] as any);
-      {
-        id: 'human-1',
-        chat_jid: chatJid,
-        sender: 'user@test',
-        sender_name: 'User',
-        content: '리뷰해줘',
-        timestamp: '2026-03-30T00:00:11.000Z',
-        seq: 12,
-        is_bot_message: false,
-      },
-    ] as any);
 
     const runtime = createMessageRuntime({
       assistantName: 'Andy',
@@ -2743,6 +2932,7 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
       triggerPattern: /^@Andy\b/i,
       channels: [ownerChannel],
       queue: {
+        enqueueMessageCheck: vi.fn(),
         registerProcess: vi.fn(),
         closeStdin: vi.fn(),
         notifyIdle: vi.fn(),
@@ -3000,6 +3190,7 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
       triggerPattern: /^@Andy\b/i,
       channels: [ownerChannel, reviewerChannel],
       queue: {
+        enqueueMessageCheck: vi.fn(),
         registerProcess: vi.fn(),
         closeStdin: vi.fn(),
         notifyIdle: vi.fn(),

src/message-runtime.ts

@@ -377,7 +377,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
             },
             scheduled
               ? 'Queued paired follow-up after successful owner delivery'
-              : 'Skipped duplicate paired follow-up after successful owner delivery in the same run',
+              : 'Skipped duplicate paired follow-up after successful owner delivery while task state was unchanged',
           );
         }
       }

src/paired-follow-up-scheduler.test.ts

@@ -13,7 +13,7 @@ describe('paired follow-up scheduler', () => {
     vi.useRealTimers();
   });
 
-  it('deduplicates the same follow-up intent within one run', () => {
+  it('deduplicates the same follow-up intent while task state is unchanged', () => {
     const enqueue = vi.fn();
     const task = {
       id: 'task-1',
@@ -41,6 +41,34 @@ describe('paired follow-up scheduler', () => {
     expect(enqueue).toHaveBeenCalledTimes(1);
   });
 
+  it('deduplicates the same follow-up intent across different runs', () => {
+    const enqueue = vi.fn();
+    const task = {
+      id: 'task-1',
+      status: 'review_ready',
+      round_trip_count: 1,
+    } as const;
+
+    const first = schedulePairedFollowUpOnce({
+      chatJid: 'group@test',
+      runId: 'run-1',
+      task,
+      intentKind: 'reviewer-turn',
+      enqueue,
+    });
+    const second = schedulePairedFollowUpOnce({
+      chatJid: 'group@test',
+      runId: 'run-2',
+      task,
+      intentKind: 'reviewer-turn',
+      enqueue,
+    });
+
+    expect(first).toBe(true);
+    expect(second).toBe(false);
+    expect(enqueue).toHaveBeenCalledTimes(1);
+  });
+
   it('keeps different round trips schedulable', () => {
     const enqueue = vi.fn();
 

src/paired-follow-up-scheduler.ts

@@ -48,7 +48,6 @@ export function schedulePairedFollowUpOnce(args: {
 
   const key = [
     args.chatJid,
-    args.runId,
     buildPairedFollowUpKey({
       taskId: args.task.id,
       taskStatus: args.task.status,