From d8b34db92ccba2e4597b38fd3cf6a262d9cb7cab Mon Sep 17 00:00:00 2001 From: ejclaw Date: Fri, 10 Apr 2026 23:41:41 +0900 Subject: [PATCH] fix: persist direct terminal verdicts for paired reviewers --- src/group-queue.test.ts | 43 ++++++++ src/group-queue.ts | 28 ++++-- src/index.ts | 2 +- src/message-agent-executor-paired.ts | 63 ++++++++++-- src/message-agent-executor.test.ts | 143 +++++++++++++++++++++++++++ src/message-agent-executor.ts | 31 +++++- 6 files changed, 292 insertions(+), 18 deletions(-) diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index fc72e6f..4c45386 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -124,6 +124,49 @@ describe('GroupQueue', () => { await vi.advanceTimersByTimeAsync(10); }); + it('stores direct terminal text for the active run and clears it when the run ends', async () => { + let releaseRun!: (value: boolean) => void; + let runId: string | undefined; + const blocker = new Promise((resolve) => { + releaseRun = resolve; + }); + + queue.setProcessMessagesFn( + async (_groupJid: string, context: GroupRunContext) => { + runId = context.runId; + return await blocker; + }, + ); + + queue.enqueueMessageCheck('group1@g.us'); + await vi.advanceTimersByTimeAsync(10); + + expect(runId).toEqual(expect.any(String)); + + queue.noteDirectTerminalDelivery( + 'group1@g.us', + 'reviewer', + 'DONE_WITH_CONCERNS\n리뷰 완료', + ); + + expect( + queue.hasDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'), + ).toBe(true); + expect( + queue.getDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'), + ).toBe('DONE_WITH_CONCERNS\n리뷰 완료'); + + releaseRun(true); + await vi.advanceTimersByTimeAsync(10); + + expect( + queue.hasDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'), + ).toBe(false); + expect( + queue.getDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'), + ).toBeNull(); + }); + it('force-terminates a lingering process after output was delivered', async () => { class StubbornProcess extends EventEmitter { exitCode: number | null = null; diff --git a/src/group-queue.ts b/src/group-queue.ts index 81b8d39..6a441e0 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -48,7 +48,7 @@ interface GroupState { runPhase: RunPhase; runningTaskId: string | null; currentRunId: string | null; - directTerminalDeliveryRoles: Set; + directTerminalDeliveries: Map; pendingMessages: boolean; pendingTasks: QueuedTask[]; process: ChildProcess | null; @@ -114,7 +114,7 @@ function resetRunState(state: GroupState, groupJid: string): void { state.process = null; state.processName = null; state.ipcDir = null; - state.directTerminalDeliveryRoles.clear(); + state.directTerminalDeliveries.clear(); transitionRunPhase(state, groupJid, 'idle'); } @@ -199,7 +199,7 @@ export class GroupQueue { runPhase: 'idle', runningTaskId: null, currentRunId: null, - directTerminalDeliveryRoles: new Set(), + directTerminalDeliveries: new Map(), pendingMessages: false, pendingTasks: [], process: null, @@ -427,21 +427,25 @@ export class GroupQueue { noteDirectTerminalDelivery( groupJid: string, senderRole?: string | null, + text?: string | null, ): void { const state = this.getGroup(groupJid); if ( state.runPhase !== 'running_messages' || !state.currentRunId || - !senderRole + !senderRole || + !text || + text.length === 0 ) { return; } - state.directTerminalDeliveryRoles.add(senderRole); + state.directTerminalDeliveries.set(senderRole, text); logger.info( { groupJid, runId: state.currentRunId, senderRole, + textLength: text.length, }, 'Recorded direct terminal delivery for active run', ); @@ -456,7 +460,19 @@ export class GroupQueue { if (state.currentRunId !== runId || !senderRole) { return false; } - return state.directTerminalDeliveryRoles.has(senderRole); + return state.directTerminalDeliveries.has(senderRole); + } + + getDirectTerminalDeliveryForRun( + groupJid: string, + runId: string, + senderRole?: string | null, + ): string | null { + const state = this.getGroup(groupJid); + if (state.currentRunId !== runId || !senderRole) { + return null; + } + return state.directTerminalDeliveries.get(senderRole) ?? null; } private clearPostCloseTimers(state: GroupState): void { diff --git a/src/index.ts b/src/index.ts index 41878f1..28e3fc4 100644 --- a/src/index.ts +++ b/src/index.ts @@ -391,7 +391,7 @@ async function main(): Promise { (senderRole === 'reviewer' || senderRole === 'arbiter') && isTerminalStatusMessage(text) ) { - queue.noteDirectTerminalDelivery(jid, senderRole); + queue.noteDirectTerminalDelivery(jid, senderRole, text); } }, nudgeScheduler: nudgeSchedulerLoop, diff --git a/src/message-agent-executor-paired.ts b/src/message-agent-executor-paired.ts index 09c94cd..40cdcbd 100644 --- a/src/message-agent-executor-paired.ts +++ b/src/message-agent-executor-paired.ts @@ -44,6 +44,7 @@ export function createPairedExecutionLifecycle(args: { chatJid: string; runId: string; enqueueMessageCheck: () => void; + getDirectTerminalDeliveryText?: () => string | null; onOutput?: (output: AgentOutput) => Promise; log: ExecutorLog; }): PairedExecutionLifecycle { @@ -54,6 +55,7 @@ export function createPairedExecutionLifecycle(args: { chatJid, runId, enqueueMessageCheck, + getDirectTerminalDeliveryText, onOutput, log, } = args; @@ -61,6 +63,7 @@ export function createPairedExecutionLifecycle(args: { let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed'; let pairedExecutionSummary: string | null = null; let pairedFinalOutput: string | null = null; + let pairedSummaryLocked = false; let pairedExecutionCompleted = false; let pairedExecutionDelegated = false; let pairedSawOutput = false; @@ -180,8 +183,48 @@ export function createPairedExecutionLifecycle(args: { pairedExecutionCompleted = true; }; + const lockVisibleVerdict = (outputText: string) => { + if (outputText.length === 0) { + return; + } + if (!pairedFinalOutput || pairedFinalOutput.length === 0) { + pairedFinalOutput = outputText; + } + if (!pairedSummaryLocked) { + pairedExecutionSummary = outputText.slice(0, 500); + pairedSummaryLocked = true; + } + pairedSawOutput = true; + }; + + const adoptDirectTerminalDeliveryIfNeeded = () => { + const outputText = getDirectTerminalDeliveryText?.(); + if (!outputText || outputText.length === 0) { + return null; + } + if (!pairedFinalOutput || pairedFinalOutput.length === 0) { + lockVisibleVerdict(outputText); + log.info( + { + pairedTaskId: pairedExecutionContext?.task.id ?? null, + role: completedRole, + runId, + }, + 'Adopted direct terminal delivery as paired final output', + ); + } else if (!pairedSummaryLocked) { + pairedExecutionSummary = pairedFinalOutput.slice(0, 500); + pairedSummaryLocked = true; + } + return outputText; + }; + return { updateSummary({ outputText, errorText }) { + if (pairedSummaryLocked) { + return; + } + if (outputText && outputText.length > 0) { pairedExecutionSummary = outputText.slice(0, 500); return; @@ -193,7 +236,7 @@ export function createPairedExecutionLifecycle(args: { }, recordFinalOutputBeforeDelivery(outputText) { - pairedFinalOutput = outputText; + lockVisibleVerdict(outputText); completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded(); persistPairedTurnOutputIfNeeded(); }, @@ -258,6 +301,8 @@ export function createPairedExecutionLifecycle(args: { return; } + const directTerminalOutput = adoptDirectTerminalDeliveryIfNeeded(); + const missingVisibleVerdict = requiresVisibleVerdict && (!pairedFinalOutput || pairedFinalOutput.length === 0); @@ -337,12 +382,16 @@ export function createPairedExecutionLifecycle(args: { } } - const queueAction = resolvePairedFollowUpQueueAction({ - completedRole, - executionStatus: effectiveStatus, - sawOutput: sawOutputForFollowUp, - taskStatus: finishedTask?.status ?? null, - }); + const queueAction = + directTerminalOutput && + (completedRole === 'reviewer' || completedRole === 'arbiter') + ? 'none' + : resolvePairedFollowUpQueueAction({ + completedRole, + executionStatus: effectiveStatus, + sawOutput: sawOutputForFollowUp, + taskStatus: finishedTask?.status ?? null, + }); if (queueAction !== 'pending' || !finishedTask) { return; } diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 890a519..318febf 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -1083,6 +1083,149 @@ describe('runAgentForGroup room memory', () => { expect(db.insertPairedTurnOutput).not.toHaveBeenCalled(); }); + it('adopts a direct terminal reviewer delivery as the paired final output and avoids requeue', async () => { + const group = { + ...makeGroup(), + folder: 'test-group', + workDir: '/repo', + agentType: 'codex' as const, + }; + const deps = { + ...makeDeps(), + queue: { + registerProcess: vi.fn(), + enqueueMessageCheck: vi.fn(), + getDirectTerminalDeliveryForRun: vi.fn( + (groupJid: string, runId: string, senderRole?: string | null) => + groupJid === 'group@test' && + runId === 'run-review-direct-terminal' && + senderRole === 'reviewer' + ? 'DONE_WITH_CONCERNS\n핵심 concern이 있습니다.' + : null, + ), + }, + }; + + vi.mocked(serviceRouting.getEffectiveChannelLease).mockReturnValue({ + chat_jid: 'group@test', + owner_agent_type: 'claude-code', + reviewer_agent_type: 'codex', + arbiter_agent_type: null, + owner_service_id: 'claude', + reviewer_service_id: 'codex-review', + arbiter_service_id: null, + activated_at: null, + reason: null, + explicit: false, + }); + vi.mocked( + pairedExecutionContext.preparePairedExecutionContext, + ).mockReturnValue({ + task: { + id: 'paired-task-review-direct-terminal', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + review_requested_at: '2026-04-10T00:00:00.000Z', + plan_notes: null, + round_trip_count: 1, + status: 'in_review', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-04-10T00:00:00.000Z', + updated_at: '2026-04-10T00:00:00.000Z', + }, + claimedTaskUpdatedAt: '2026-04-10T00:00:00.000Z', + workspace: null, + envOverrides: {}, + requiresVisibleVerdict: true, + }); + vi.mocked(db.getPairedTaskById).mockReturnValue({ + id: 'paired-task-review-direct-terminal', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-review', + title: null, + source_ref: 'HEAD', + plan_notes: null, + round_trip_count: 1, + review_requested_at: '2026-04-10T00:00:00.000Z', + status: 'review_ready', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-04-10T00:00:00.000Z', + updated_at: '2026-04-10T00:00:00.000Z', + }); + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '리뷰 완료했습니다. 핵심 concern을 정리 중입니다.', + output: { + visibility: 'public', + text: '리뷰 완료했습니다. 핵심 concern을 정리 중입니다.', + }, + } as any); + return { + status: 'success', + result: null, + }; + }, + ); + + const result = await runAgentForGroup(deps, { + group, + prompt: 'please review', + chatJid: 'group@test', + runId: 'run-review-direct-terminal', + forcedRole: 'reviewer', + pairedTurnIdentity: { + turnId: + 'paired-task-review-direct-terminal:2026-04-10T00:00:00.000Z:reviewer-turn', + taskId: 'paired-task-review-direct-terminal', + taskUpdatedAt: '2026-04-10T00:00:00.000Z', + intentKind: 'reviewer-turn', + role: 'reviewer', + }, + onOutput: async () => {}, + }); + + expect(result).toBe('success'); + expect( + pairedExecutionContext.completePairedExecutionContext, + ).toHaveBeenCalledWith( + expect.objectContaining({ + taskId: 'paired-task-review-direct-terminal', + role: 'reviewer', + status: 'succeeded', + summary: 'DONE_WITH_CONCERNS\n핵심 concern이 있습니다.', + }), + ); + expect(db.insertPairedTurnOutput).toHaveBeenCalledWith( + 'paired-task-review-direct-terminal', + 1, + 'reviewer', + 'DONE_WITH_CONCERNS\n핵심 concern이 있습니다.', + ); + expect(db.completePairedTurn).toHaveBeenCalledWith({ + turnId: + 'paired-task-review-direct-terminal:2026-04-10T00:00:00.000Z:reviewer-turn', + taskId: 'paired-task-review-direct-terminal', + taskUpdatedAt: '2026-04-10T00:00:00.000Z', + intentKind: 'reviewer-turn', + role: 'reviewer', + }); + expect(db.failPairedTurn).not.toHaveBeenCalled(); + expect(deps.queue.enqueueMessageCheck).not.toHaveBeenCalled(); + }); + it('passes paired workspace env overrides into the runner when execution metadata exists', async () => { const group = { ...makeGroup(), diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index b2f064b..92a1fb2 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -68,7 +68,8 @@ import type { AgentType, PairedRoomRole, RegisteredGroup } from './types.js'; export interface MessageAgentExecutorDeps { assistantName: string; - queue: Pick; + queue: Pick & + Partial>; getRoomBindings: () => Record; getSessions: () => Record; persistSession: (groupFolder: string, sessionId: string) => void; @@ -402,9 +403,21 @@ export async function runAgentForGroup( chatJid, runId, enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), + getDirectTerminalDeliveryText: () => + deps.queue.getDirectTerminalDeliveryForRun?.( + chatJid, + runId, + runtimePairedTurnIdentity?.role ?? activeRole, + ) ?? null, onOutput, log, }); + const hasDirectTerminalDelivery = (): boolean => + !!deps.queue.getDirectTerminalDeliveryForRun?.( + chatJid, + runId, + runtimePairedTurnIdentity?.role ?? activeRole, + ); const maybeHandoffToCodex = ( reason: AgentTriggerReason, @@ -917,7 +930,11 @@ export async function runAgentForGroup( }); } - if (!attempt.sawOutput && output.status !== 'error') { + if ( + !attempt.sawOutput && + !hasDirectTerminalDelivery() && + output.status !== 'error' + ) { const claudeRetryResult = await retryClaudeAttemptIfNeeded(attempt); if (claudeRetryResult) { return claudeRetryResult; @@ -973,7 +990,11 @@ export async function runAgentForGroup( // success-null-result with no visible output — agent returned nothing useful. // But if output was already delivered to Discord (sawOutput), treat as success. - if (attempt.sawSuccessNullResultWithoutOutput && !attempt.sawOutput) { + if ( + attempt.sawSuccessNullResultWithoutOutput && + !attempt.sawOutput && + !hasDirectTerminalDelivery() + ) { log.error( 'Agent returned success with null result and no visible output', ); @@ -981,7 +1002,9 @@ export async function runAgentForGroup( } pairedExecutionLifecycle.markStatus('succeeded'); - pairedExecutionLifecycle.markSawOutput(attempt.sawOutput); + pairedExecutionLifecycle.markSawOutput( + attempt.sawOutput || hasDirectTerminalDelivery(), + ); return 'success'; };