diff --git a/src/group-queue.ts b/src/group-queue.ts index 1e290ad..81b8d39 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -48,6 +48,7 @@ interface GroupState { runPhase: RunPhase; runningTaskId: string | null; currentRunId: string | null; + directTerminalDeliveryRoles: Set; pendingMessages: boolean; pendingTasks: QueuedTask[]; process: ChildProcess | null; @@ -113,6 +114,7 @@ function resetRunState(state: GroupState, groupJid: string): void { state.process = null; state.processName = null; state.ipcDir = null; + state.directTerminalDeliveryRoles.clear(); transitionRunPhase(state, groupJid, 'idle'); } @@ -197,6 +199,7 @@ export class GroupQueue { runPhase: 'idle', runningTaskId: null, currentRunId: null, + directTerminalDeliveryRoles: new Set(), pendingMessages: false, pendingTasks: [], process: null, @@ -421,6 +424,41 @@ export class GroupQueue { } } + noteDirectTerminalDelivery( + groupJid: string, + senderRole?: string | null, + ): void { + const state = this.getGroup(groupJid); + if ( + state.runPhase !== 'running_messages' || + !state.currentRunId || + !senderRole + ) { + return; + } + state.directTerminalDeliveryRoles.add(senderRole); + logger.info( + { + groupJid, + runId: state.currentRunId, + senderRole, + }, + 'Recorded direct terminal delivery for active run', + ); + } + + hasDirectTerminalDeliveryForRun( + groupJid: string, + runId: string, + senderRole?: string | null, + ): boolean { + const state = this.getGroup(groupJid); + if (state.currentRunId !== runId || !senderRole) { + return false; + } + return state.directTerminalDeliveryRoles.has(senderRole); + } + private clearPostCloseTimers(state: GroupState): void { if (state.postCloseTermTimer) { clearTimeout(state.postCloseTermTimer); diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 928cfd7..599de6f 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -1091,6 +1091,190 @@ describe('createMessageRuntime', () => { ); }); + it('skips reviewer final work item delivery when a direct terminal IPC message was already recorded for the run', 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 noteDirectTerminalDelivery = vi.fn(); + let recordedReviewerTerminal = false; + + vi.mocked(config.isClaudeService).mockReturnValue(false); + vi.mocked(config.isReviewService).mockReturnValue(false); + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({ + id: 'task-review-direct-terminal', + 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: 0, + 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', + }); + vi.mocked(db.getLastHumanMessageContent).mockReturnValue('리뷰해줘'); + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: '**DONE**\n\n리뷰 승인', + newSessionId: 'session-review-direct-terminal', + }); + return { + status: 'success', + result: '**DONE**\n\n리뷰 승인', + newSessionId: 'session-review-direct-terminal', + }; + }, + ); + + 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(), + noteDirectTerminalDelivery: vi.fn( + (groupJid: string, senderRole?: string | null) => { + noteDirectTerminalDelivery(groupJid, senderRole); + if (groupJid === chatJid && senderRole === 'reviewer') { + recordedReviewerTerminal = true; + } + }, + ), + hasDirectTerminalDeliveryForRun: vi.fn( + (groupJid: string, runId: string, senderRole?: string | null) => + groupJid === chatJid && + runId === 'run-review-direct-terminal-skip' && + senderRole === 'reviewer' && + recordedReviewerTerminal, + ), + } 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-review-direct-terminal-skip', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(noteDirectTerminalDelivery).toHaveBeenCalledWith( + chatJid, + 'reviewer', + ); + expect(db.createProducedWorkItem).not.toHaveBeenCalled(); + expect(reviewerChannel.sendMessage).not.toHaveBeenCalled(); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid, + runId: 'run-review-direct-terminal-skip', + deliveryRole: 'reviewer', + }), + 'Skipping final work item delivery because this run already sent a direct terminal IPC message', + ); + }); + + it('does not suppress owner finals even if the queue reports a direct terminal delivery for the run', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const noteDirectTerminalDelivery = vi.fn(); + const hasDirectTerminalDeliveryForRun = vi.fn(() => true); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: '**DONE**\n\nowner final', + newSessionId: 'session-owner-terminal', + }); + return { + status: 'success', + result: '**DONE**\n\nowner final', + newSessionId: 'session-owner-terminal', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + noteDirectTerminalDelivery, + hasDirectTerminalDeliveryForRun, + } 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-owner-direct-terminal-ignore', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(noteDirectTerminalDelivery).not.toHaveBeenCalled(); + expect(hasDirectTerminalDeliveryForRun).not.toHaveBeenCalled(); + expect(db.createProducedWorkItem).toHaveBeenCalledWith( + expect.objectContaining({ + result_payload: '**DONE**\n\nowner final', + }), + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '**DONE**\n\nowner final', + ); + }); + it('includes the latest reviewer summary in merge_ready finalize prompts and truncates it', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 947235a..5b827e5 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1,4 +1,5 @@ import { AgentOutput } from './agent-runner.js'; +import { getAgentOutputText } from './agent-output.js'; import { getErrorMessage } from './utils.js'; import { claimServiceHandoff, @@ -140,6 +141,13 @@ function getMissingRoleChannelMessage(role: 'reviewer' | 'arbiter'): string { return `Missing configured ${role} Discord bot channel (${getFixedRoleChannelName(role)}) for role-fixed delivery`; } +function isTerminalStatusMessage(text: string): boolean { + const trimmed = text.trimStart(); + return /^(?:\*\*)?(DONE(?:_WITH_CONCERNS)?|BLOCKED|NEEDS_CONTEXT)\b/.test( + trimmed, + ); +} + export function createMessageRuntime(deps: MessageRuntimeDeps): { processGroupMessages: ( chatJid: string, @@ -496,6 +504,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const isClaudeCodeAgent = (args.forcedAgentType ?? group.agentType ?? 'claude-code') === 'claude-code'; + let directTerminalDeliveryRecorded = false; const turnController = new MessageTurnController({ chatJid, @@ -514,6 +523,25 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { args.deliveryRole ?? args.forcedRole ?? (hasReviewerLease(chatJid) ? 'owner' : null); + if ( + (persistedDeliveryRole === 'reviewer' || + persistedDeliveryRole === 'arbiter') && + deps.queue.hasDirectTerminalDeliveryForRun?.( + chatJid, + runId, + persistedDeliveryRole, + ) + ) { + logger.info( + { + chatJid, + runId, + deliveryRole: persistedDeliveryRole, + }, + 'Skipping final work item delivery because this run already sent a direct terminal IPC message', + ); + return true; + } const workItem = createProducedWorkItem({ group_folder: group.folder, chat_jid: chatJid, @@ -543,7 +571,20 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { prompt, chatJid, runId, - (result) => turnController.handleOutput(result), + async (result) => { + const raw = result ? getAgentOutputText(result) : null; + if ( + !directTerminalDeliveryRecorded && + raw && + isTerminalStatusMessage(raw) && + (args.deliveryRole === 'reviewer' || + args.deliveryRole === 'arbiter') + ) { + deps.queue.noteDirectTerminalDelivery?.(chatJid, args.deliveryRole); + directTerminalDeliveryRecorded = true; + } + await turnController.handleOutput(result); + }, { startSeq, endSeq,