From df04ef25ebd944d190f18e90a67e573d1a3a92f5 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Mon, 13 Apr 2026 10:51:17 +0900 Subject: [PATCH] runtime: resume review after CI watcher completion --- src/message-runtime.test.ts | 107 ++++++++++++++++++++++++++ src/message-runtime.ts | 18 +++++ src/paired-execution-context-owner.ts | 20 ++--- src/paired-execution-context.test.ts | 36 +++++++++ 4 files changed, 172 insertions(+), 9 deletions(-) diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 591fad3..9fc06d2 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -137,6 +137,7 @@ vi.mock('./db.js', () => { getOpenWorkItemForChat: vi.fn((chatJid: string) => getOpenWorkItem(chatJid), ), + hasActiveCiWatcherForChat: vi.fn(() => false), getLatestOpenPairedTaskForChat: vi.fn(() => undefined), getPairedTaskById: vi.fn(() => undefined), getPairedTurnOutputs: vi.fn(() => []), @@ -1878,6 +1879,112 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead ); }); + it('defers reviewer enqueue after owner delivery when a CI watcher is still active', 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-watcher-deferred', + 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: 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.hasActiveCiWatcherForChat).mockReturnValue(true); + vi.mocked(db.getLatestOpenPairedTaskForChat).mockImplementation( + () => pairedTask, + ); + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'human-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\nowner complete', + newSessionId: 'session-owner-delivery-watcher-deferred', + }); + return { + status: 'success', + result: 'DONE\nowner complete', + newSessionId: 'session-owner-delivery-watcher-deferred', + }; + }, + ); + + 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, + getRoomBindings: () => ({ [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-owner-delivery-watcher-deferred', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(enqueueMessageCheck).not.toHaveBeenCalled(); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid, + runId: 'run-owner-delivery-watcher-deferred', + completedRole: 'owner', + taskId: 'task-owner-delivery-watcher-deferred', + taskStatus: 'review_ready', + }), + 'Deferred paired follow-up after successful owner delivery because CI watcher is still active', + ); + }); + it('skips a stale reviewer follow-up after owner delivery when the latest persisted turn already belongs to the reviewer', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index b45f6e1..eba3a1c 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1,5 +1,6 @@ import { createProducedWorkItem, + hasActiveCiWatcherForChat, getOpenWorkItemForChat, getMessagesSinceSeq, getLatestOpenPairedTaskForChat, @@ -217,6 +218,23 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return; } const pendingTaskAfterDelivery = getLatestOpenPairedTaskForChat(chatJid); + if ( + deliveryRole === 'owner' && + pendingTaskAfterDelivery?.status === 'review_ready' && + hasActiveCiWatcherForChat(chatJid) + ) { + logger.info( + { + chatJid, + runId, + completedRole: deliveryRole, + taskId: pendingTaskAfterDelivery.id, + taskStatus: pendingTaskAfterDelivery.status, + }, + 'Deferred paired follow-up after successful owner delivery because CI watcher is still active', + ); + return; + } const followUpResult = enqueuePairedFollowUpAfterEvent({ chatJid, runId, diff --git a/src/paired-execution-context-owner.ts b/src/paired-execution-context-owner.ts index d132320..1c2f9dd 100644 --- a/src/paired-execution-context-owner.ts +++ b/src/paired-execution-context-owner.ts @@ -142,15 +142,6 @@ function maybeAutoTriggerReviewerAfterOwnerCompletion(args: { logMessage: string; }): void { const { task, taskId, now, logMessage } = args; - - if (hasActiveCiWatcherForChat(task.chat_jid)) { - logger.info( - { taskId, chatJid: task.chat_jid }, - 'Active CI watcher found, deferring auto-review until watcher completes', - ); - return; - } - if (task.round_trip_count >= PAIRED_MAX_ROUND_TRIPS) { logger.info( { @@ -177,6 +168,17 @@ function maybeAutoTriggerReviewerAfterOwnerCompletion(args: { round_trip_count: task.round_trip_count + 1, }, }); + if (hasActiveCiWatcherForChat(task.chat_jid)) { + logger.info( + { + taskId, + chatJid: task.chat_jid, + roundTrip: task.round_trip_count + 1, + }, + 'Active CI watcher found, marked task review_ready and deferred reviewer enqueue until watcher completes', + ); + return; + } logger.info({ taskId, roundTrip: task.round_trip_count + 1 }, logMessage); } } diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts index 7c91f35..6caef09 100644 --- a/src/paired-execution-context.test.ts +++ b/src/paired-execution-context.test.ts @@ -757,6 +757,42 @@ describe('paired execution context', () => { ); }); + it('marks review_ready but defers reviewer enqueue when an active CI watcher exists', () => { + vi.mocked(db.getPairedTaskById).mockReturnValue( + buildPairedTask({ + status: 'active', + round_trip_count: 0, + }), + ); + vi.mocked(db.hasActiveCiWatcherForChat).mockReturnValue(true); + vi.mocked(pairedWorkspaceManager.markPairedTaskReviewReady).mockReturnValue( + { + ownerWorkspace: buildWorkspace('owner', '/tmp/paired/task-1/owner'), + reviewerWorkspace: buildWorkspace( + 'reviewer', + '/tmp/paired/task-1/reviewer', + ), + }, + ); + + completePairedExecutionContext({ + taskId: 'task-1', + role: 'owner', + status: 'succeeded', + summary: 'DONE', + }); + + expect( + pairedWorkspaceManager.markPairedTaskReviewReady, + ).toHaveBeenCalledWith('task-1'); + expect(db.updatePairedTask).toHaveBeenCalledWith( + 'task-1', + expect.objectContaining({ + round_trip_count: 1, + }), + ); + }); + it('requests arbiter instead of re-reviewing when repeated DONE finalize loops exceed the threshold', () => { vi.spyOn(config, 'isArbiterEnabled').mockReturnValue(true);