From 0f751af151ce612fb02fa5deaa51b37783127491 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Wed, 8 Apr 2026 03:50:06 +0900 Subject: [PATCH] Stabilize paired follow-up routing --- src/message-agent-executor-paired.ts | 4 +- src/message-agent-executor-rules.ts | 5 +- src/message-agent-executor.test.ts | 2 +- src/message-runtime-dispatch.ts | 2 +- src/message-runtime-flow.test.ts | 2 +- src/message-runtime-flow.ts | 2 +- src/message-runtime-rules.test.ts | 4 +- src/message-runtime-rules.ts | 5 +- src/message-runtime.test.ts | 219 +++++++++++++++++++++++-- src/message-runtime.ts | 2 +- src/paired-follow-up-scheduler.test.ts | 30 +++- src/paired-follow-up-scheduler.ts | 1 - 12 files changed, 248 insertions(+), 30 deletions(-) diff --git a/src/message-agent-executor-paired.ts b/src/message-agent-executor-paired.ts index 446e594..f7aad5f 100644 --- a/src/message-agent-executor-paired.ts +++ b/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', ); }, }; diff --git a/src/message-agent-executor-rules.ts b/src/message-agent-executor-rules.ts index 2f2b96a..4fab503 100644 --- a/src/message-agent-executor-rules.ts +++ b/src/message-agent-executor-rules.ts @@ -23,10 +23,7 @@ export function resolvePairedFollowUpQueueAction(args: { }): PairedFollowUpQueueAction { const nextTurnAction = resolveNextTurnAction({ taskStatus: args.taskStatus, - lastTurnOutputRole: - args.executionStatus === 'succeeded' && args.sawOutput - ? args.completedRole - : null, + lastTurnOutputRole: args.sawOutput ? args.completedRole : null, }); const dispatch = resolveFollowUpDispatch({ source: 'executor-recovery', diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 8b5fab3..7948dfc 100644 --- a/src/message-agent-executor.test.ts +++ b/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', ); }); }); diff --git a/src/message-runtime-dispatch.ts b/src/message-runtime-dispatch.ts index a4dea46..8b46ab0 100644 --- a/src/message-runtime-dispatch.ts +++ b/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', ); } } diff --git a/src/message-runtime-flow.test.ts b/src/message-runtime-flow.test.ts index 2d1b55c..280f51b 100644 --- a/src/message-runtime-flow.test.ts +++ b/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', ); }); }); diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index d766caf..ca311e2 100644 --- a/src/message-runtime-flow.ts +++ b/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; } diff --git a/src/message-runtime-rules.test.ts b/src/message-runtime-rules.test.ts index 79e4a32..1504062 100644 --- a/src/message-runtime-rules.test.ts +++ b/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', diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index 1369d62..9444324 100644 --- a/src/message-runtime-rules.ts +++ b/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'; } diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index abd0dc8..90d2cb7 100644 --- a/src/message-runtime.test.ts +++ b/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([ - { - 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); + vi.mocked(db.getMessagesSince).mockReturnValue([] 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(), diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 391a28a..8a8d37e 100644 --- a/src/message-runtime.ts +++ b/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', ); } } diff --git a/src/paired-follow-up-scheduler.test.ts b/src/paired-follow-up-scheduler.test.ts index 65fb0f1..6d612ed 100644 --- a/src/paired-follow-up-scheduler.test.ts +++ b/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(); diff --git a/src/paired-follow-up-scheduler.ts b/src/paired-follow-up-scheduler.ts index 11b4574..dfae7fa 100644 --- a/src/paired-follow-up-scheduler.ts +++ b/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,