From 75066f345dc6a3e83983d5f9f885bc9771260ba9 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Tue, 7 Apr 2026 08:07:25 +0900 Subject: [PATCH] Deduplicate paired reviewer handoffs --- src/message-agent-executor-rules.test.ts | 8 +- src/message-agent-executor-rules.ts | 11 +- src/message-agent-executor.test.ts | 109 +++++++++++++++- src/message-agent-executor.ts | 19 +-- src/message-runtime-flow.ts | 30 +++++ src/message-runtime-rules.test.ts | 18 +++ src/message-runtime-rules.ts | 12 +- src/message-runtime.test.ts | 158 +++++++++++++++++++++++ 8 files changed, 330 insertions(+), 35 deletions(-) diff --git a/src/message-agent-executor-rules.test.ts b/src/message-agent-executor-rules.test.ts index bb14e43..0662955 100644 --- a/src/message-agent-executor-rules.test.ts +++ b/src/message-agent-executor-rules.test.ts @@ -172,7 +172,7 @@ describe('message-agent-executor-rules', () => { ).toBe('pending'); }); - it('requests a generic follow-up after successful owner output moved the task to review_ready', () => { + it('does not request an executor-side follow-up after successful owner output moved the task to review_ready', () => { expect( resolvePairedFollowUpQueueAction({ completedRole: 'owner', @@ -180,10 +180,10 @@ describe('message-agent-executor-rules', () => { sawOutput: true, taskStatus: 'review_ready', }), - ).toBe('generic'); + ).toBe('none'); }); - it('skips generic requeue when reviewer success moved the task to merge_ready for inline finalize', () => { + it('does not request an executor-side follow-up when reviewer success moved the task to merge_ready', () => { expect( resolvePairedFollowUpQueueAction({ completedRole: 'reviewer', @@ -191,7 +191,7 @@ describe('message-agent-executor-rules', () => { sawOutput: true, taskStatus: 'merge_ready', }), - ).toBe('skip-inline-finalize'); + ).toBe('none'); }); it('does not request a generic follow-up after successful reviewer output moved the task back to active', () => { diff --git a/src/message-agent-executor-rules.ts b/src/message-agent-executor-rules.ts index daef4f8..cc66fc0 100644 --- a/src/message-agent-executor-rules.ts +++ b/src/message-agent-executor-rules.ts @@ -11,9 +11,7 @@ export { } from './agent-attempt-retry.js'; export type PairedFollowUpQueueAction = - | 'generic' | 'pending' - | 'skip-inline-finalize' | 'none'; export function resolvePairedFollowUpQueueAction(args: { @@ -31,14 +29,7 @@ export function resolvePairedFollowUpQueueAction(args: { }); if (args.executionStatus === 'succeeded' && args.sawOutput) { - if (nextTurnAction.kind === 'reviewer-turn') { - return 'generic'; - } - - return args.completedRole === 'reviewer' && - nextTurnAction.kind === 'finalize-owner-turn' - ? 'skip-inline-finalize' - : 'none'; + return 'none'; } const shouldRequeuePendingPairedTurn = diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index b9d7aa2..9ff75c8 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -1238,7 +1238,7 @@ describe('runAgentForGroup room memory', () => { expect(deps.queue.enqueueMessageCheck).not.toHaveBeenCalled(); }); - it('enqueues a generic follow-up when owner output moved the task to review_ready', async () => { + it('does not enqueue an executor-side follow-up when owner output moved the task to review_ready', async () => { const group = { ...makeGroup(), folder: 'test-group' }; const deps = makeDeps(); @@ -1311,8 +1311,7 @@ describe('runAgentForGroup room memory', () => { }); expect(result).toBe('success'); - expect(deps.queue.enqueueMessageCheck).toHaveBeenCalledTimes(1); - expect(deps.queue.enqueueMessageCheck).toHaveBeenCalledWith('group@test'); + expect(deps.queue.enqueueMessageCheck).not.toHaveBeenCalled(); }); it('stores reviewer turn output before transitioning the paired task back to active', async () => { @@ -1419,6 +1418,110 @@ describe('runAgentForGroup room memory', () => { ); }); + it('stores reviewer approval output before transitioning the paired task to merge_ready', async () => { + const group = { ...makeGroup(), folder: 'test-group' }; + const deps = makeDeps(); + const onOutput = vi.fn(async () => {}); + + 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-approval-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-approval-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: 'merge_ready', + 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\nreviewer approval', + output: { + visibility: 'public', + text: 'DONE\nreviewer approval', + }, + phase: 'final', + }); + return { + status: 'success', + result: 'DONE\nreviewer approval', + }; + }, + ); + + const result = await runAgentForGroup(deps, { + group, + prompt: 'please review', + chatJid: 'group@test', + runId: 'run-reviewer-approval-order', + forcedRole: 'reviewer', + onOutput, + }); + + expect(result).toBe('success'); + expect(db.insertPairedTurnOutput).toHaveBeenCalledWith( + 'paired-task-reviewer-approval-order', + 1, + 'reviewer', + 'DONE\nreviewer approval', + ); + expect( + vi.mocked(db.insertPairedTurnOutput).mock.invocationCallOrder[0], + ).toBeLessThan(onOutput.mock.invocationCallOrder[0]); + 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(); diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 28ef15a..039ad01 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -991,9 +991,9 @@ export async function runAgentForGroup( } } - // After owner/reviewer completes, enqueue the next turn so - // the message loop picks it up without waiting for a new message. - // Skip if: no output (interrupted), or task already completed (ESCALATE, done, etc.) + // Failed reviewer/arbiter runs can leave a pending paired task without a + // visible bot message. Requeue only those recovery cases here; successful + // paired handoffs are driven by the delivered bot message itself. if (pairedExecutionContext) { const completedRole = roomRoleContext?.role ?? 'owner'; const finishedCheck = getPairedTaskById(pairedExecutionContext.task.id); @@ -1003,18 +1003,7 @@ export async function runAgentForGroup( sawOutput: pairedSawOutput, taskStatus: finishedCheck?.status ?? null, }); - if (queueAction === 'generic') { - deps.queue.enqueueMessageCheck(chatJid); - } else if (queueAction === 'skip-inline-finalize') { - log.info( - { - taskId: pairedExecutionContext.task.id, - role: completedRole, - taskStatus: finishedCheck?.status ?? null, - }, - 'Skipping generic follow-up after reviewer approval because inline finalize will handle the owner handoff', - ); - } else if (queueAction === 'pending') { + if (queueAction === 'pending') { deps.queue.enqueueMessageCheck(chatJid); log.info( { diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index 261b43a..ab49fab 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -35,6 +35,12 @@ export type PendingPairedTurn = { export type BotOnlyPairedFollowUpAction = | { kind: 'none' } + | { + kind: 'consume-stale-bot-message'; + task: PairedTask; + cursor: string | number | null; + currentStatus: PairedTask['status']; + } | { kind: 'inline-finalize'; task: PairedTask; @@ -249,6 +255,15 @@ export function resolveBotOnlyPairedFollowUpAction(args: { lastTurnOutputRole: lastTurnOutput?.role ?? null, }); + if (nextTurnAction.kind === 'none') { + return { + kind: 'consume-stale-bot-message', + task, + cursor, + currentStatus: task.status, + }; + } + if (nextTurnAction.kind === 'finalize-owner-turn') { return { kind: 'inline-finalize', @@ -328,6 +343,21 @@ export async function executeBotOnlyPairedFollowUpAction(args: { ); } + if (action.kind === 'consume-stale-bot-message') { + log.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + taskId: action.task.id, + taskStatus: action.currentStatus, + cursor: action.cursor, + }, + 'Consumed stale bot-only paired message because no follow-up turn is pending', + ); + return true; + } + if (action.kind === 'inline-finalize') { log.info( { diff --git a/src/message-runtime-rules.test.ts b/src/message-runtime-rules.test.ts index 41cdeff..e49a061 100644 --- a/src/message-runtime-rules.test.ts +++ b/src/message-runtime-rules.test.ts @@ -49,6 +49,24 @@ describe('message-runtime-rules', () => { ).toEqual({ kind: 'finalize-owner-turn' }); }); + it('does not schedule a second reviewer turn when the latest turn already belongs to the reviewer', () => { + expect( + resolveNextTurnAction({ + taskStatus: 'review_ready', + lastTurnOutputRole: 'reviewer', + }), + ).toEqual({ kind: 'none' }); + }); + + it('does not schedule a second finalize turn when the latest turn already belongs to the owner', () => { + expect( + resolveNextTurnAction({ + taskStatus: 'merge_ready', + lastTurnOutputRole: 'owner', + }), + ).toEqual({ kind: 'none' }); + }); + it('maps active tasks with reviewer output to an owner follow-up', () => { expect( resolveNextTurnAction({ diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index 842875c..70d2e8f 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -75,12 +75,18 @@ export function resolveNextTurnAction(args: { switch (args.taskStatus) { case 'review_ready': case 'in_review': - return { kind: 'reviewer-turn' }; + return args.lastTurnOutputRole === 'reviewer' + ? { kind: 'none' } + : { kind: 'reviewer-turn' }; case 'arbiter_requested': case 'in_arbitration': - return { kind: 'arbiter-turn' }; + return args.lastTurnOutputRole === 'arbiter' + ? { kind: 'none' } + : { kind: 'arbiter-turn' }; case 'merge_ready': - return { kind: 'finalize-owner-turn' }; + return args.lastTurnOutputRole === 'owner' + ? { kind: 'none' } + : { kind: 'finalize-owner-turn' }; case 'active': return args.lastTurnOutputRole === 'reviewer' || args.lastTurnOutputRole === 'arbiter' diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 009bcfe..b05e509 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -213,6 +213,10 @@ import { resolveHandoffCursorKey, resolveHandoffRoleOverride, } from './message-runtime.js'; +import { + buildPendingPairedTurn, + resolveBotOnlyPairedFollowUpAction, +} from './message-runtime-flow.js'; import * as config from './config.js'; import { logger } from './logger.js'; import * as serviceRouting from './service-routing.js'; @@ -1564,6 +1568,160 @@ describe('createMessageRuntime', () => { expect(saveState).toHaveBeenCalled(); }); + it('does not build a second reviewer pending turn once the latest persisted turn already belongs to the reviewer', () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + const task = { + id: 'task-stale-reviewer-bot-message', + 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: 'review_ready', + 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.getPairedTurnOutputs).mockReturnValue([ + { + id: 1, + task_id: 'task-stale-reviewer-bot-message', + turn_number: 1, + role: 'owner', + output_text: 'owner 응답', + created_at: '2026-03-30T00:00:01.000Z', + }, + { + id: 2, + task_id: 'task-stale-reviewer-bot-message', + turn_number: 2, + role: 'reviewer', + output_text: 'reviewer 승인', + created_at: '2026-03-30T00:00:02.000Z', + }, + ]); + + expect( + buildPendingPairedTurn({ + chatJid, + timezone: 'UTC', + task, + rawMissedMessages: [ + { + seq: 42, + timestamp: '2026-03-30T00:00:04.000Z', + }, + ], + recentHumanMessages: [], + labeledRecentMessages: [], + resolveChannel: () => makeChannel(chatJid, 'discord-review', false), + }), + ).toBeNull(); + + expect( + resolveBotOnlyPairedFollowUpAction({ + chatJid, + task, + isBotOnlyPairedFollowUp: true, + pendingCursorSource: { + seq: 42, + timestamp: '2026-03-30T00:00:04.000Z', + }, + }), + ).toEqual({ + kind: 'consume-stale-bot-message', + task, + cursor: 42, + currentStatus: 'review_ready', + }); + }); + + it('consumes stale bot-only owner messages once the finalize turn output is already persisted', () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + const task = { + id: 'task-stale-owner-bot-message', + 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: 'merge_ready', + 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.getPairedTurnOutputs).mockReturnValue([ + { + id: 1, + task_id: 'task-stale-owner-bot-message', + turn_number: 1, + role: 'reviewer', + output_text: 'reviewer 승인', + created_at: '2026-03-30T00:00:01.000Z', + }, + { + id: 2, + task_id: 'task-stale-owner-bot-message', + turn_number: 2, + role: 'owner', + output_text: 'owner 최종 보고', + created_at: '2026-03-30T00:00:02.000Z', + }, + ]); + + expect( + buildPendingPairedTurn({ + chatJid, + timezone: 'UTC', + task, + rawMissedMessages: [ + { + seq: 43, + timestamp: '2026-03-30T00:00:04.000Z', + }, + ], + recentHumanMessages: [], + labeledRecentMessages: [], + resolveChannel: () => makeChannel(chatJid), + }), + ).toBeNull(); + + expect( + resolveBotOnlyPairedFollowUpAction({ + chatJid, + task, + isBotOnlyPairedFollowUp: true, + pendingCursorSource: { + seq: 43, + timestamp: '2026-03-30T00:00:04.000Z', + }, + }), + ).toEqual({ + kind: 'consume-stale-bot-message', + task, + cursor: 43, + currentStatus: 'merge_ready', + }); + }); + it('runs merge_ready bot-only reviewer follow-ups inline in the message loop', async () => { const chatJid = 'group@test'; const group = makeGroup('codex');