diff --git a/src/group-queue-state.ts b/src/group-queue-state.ts index e99b2ff..8719b6f 100644 --- a/src/group-queue-state.ts +++ b/src/group-queue-state.ts @@ -30,6 +30,7 @@ export interface GroupState { runPhase: RunPhase; runningTaskId: string | null; currentRunId: string | null; + lastCloseRequest: { runId: string | null; reason: string | null } | null; directTerminalDeliveries: Map; recentDirectTerminalDeliveries: Map>; pendingMessages: boolean; @@ -52,6 +53,7 @@ export function createGroupState(): GroupState { runPhase: 'idle', runningTaskId: null, currentRunId: null, + lastCloseRequest: null, directTerminalDeliveries: new Map(), recentDirectTerminalDeliveries: new Map(), pendingMessages: false, @@ -139,6 +141,7 @@ export function transitionRunPhase( export function resetRunState(state: GroupState, groupJid: string): void { state.currentRunId = null; state.runningTaskId = null; + state.lastCloseRequest = null; state.startedAt = null; state.process = null; state.processName = null; diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index dfa2806..d2991d1 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -747,3 +747,47 @@ describe('GroupQueue', () => { await vi.advanceTimersByTimeAsync(10); }); }); + +describe('GroupQueue close reason tracking', () => { + let queue: GroupQueue; + + beforeEach(() => { + vi.useFakeTimers(); + vi.clearAllMocks(); + queue = new GroupQueue(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('records the close reason for the active message run until it finishes', async () => { + const ipcDir = '/tmp/ejclaw-test-data/ipc/group-folder'; + let releaseRun!: (value: boolean) => void; + let activeRunId = ''; + const blocker = new Promise((resolve) => { + releaseRun = resolve; + }); + + queue.setProcessMessagesFn( + vi.fn(async (_groupJid: string, context: GroupRunContext) => { + activeRunId = context.runId; + return await blocker; + }), + ); + queue.enqueueMessageCheck('group1@g.us', ipcDir); + await vi.advanceTimersByTimeAsync(10); + + queue.closeStdin('group1@g.us', { reason: 'human-message-detected' }); + + expect(queue.getCloseReasonForRun('group1@g.us', activeRunId)).toBe( + 'human-message-detected', + ); + expect(queue.getCloseReasonForRun('group1@g.us', 'other-run')).toBeNull(); + + releaseRun(true); + await vi.advanceTimersByTimeAsync(10); + + expect(queue.getCloseReasonForRun('group1@g.us', activeRunId)).toBeNull(); + }); +}); diff --git a/src/group-queue.ts b/src/group-queue.ts index f8a4c11..c92d576 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -322,7 +322,10 @@ export class GroupQueue { } return state.directTerminalDeliveries.get(senderRole) ?? null; } - + getCloseReasonForRun(groupJid: string, runId: string): string | null { + const closeRequest = this.getGroup(groupJid).lastCloseRequest; + return closeRequest?.runId === runId ? closeRequest.reason : null; + } hasRecordedDirectTerminalDeliveryForRun( groupJid: string, runId: string, @@ -342,7 +345,6 @@ export class GroupQueue { state.recentDirectTerminalDeliveries.get(runId)?.has(senderRole) ?? false ); } - private clearPostCloseTimers(state: GroupState): void { if (state.postCloseTermTimer) { clearTimeout(state.postCloseTermTimer); @@ -428,29 +430,27 @@ export class GroupQueue { }, POST_CLOSE_SIGKILL_DELAY_MS); } - /** - * Signal the active agent process to wind down by writing a close sentinel. - */ closeStdin( groupJid: string, metadata?: { runId?: string; reason?: string }, ): void { const state = this.getGroup(groupJid); if (state.runPhase === 'idle' || !state.ipcDir) return; + const runId = metadata?.runId ?? state.currentRunId; + state.lastCloseRequest = { runId, reason: metadata?.reason ?? null }; if (state.runPhase === 'running_messages') { transitionRunPhase(state, groupJid, 'closing_messages', { reason: metadata?.reason, - runId: metadata?.runId ?? state.currentRunId, + runId, }); assertRunPhaseInvariants(state, groupJid); } - try { writeCloseSentinel(state.ipcDir); logger.info( { groupJid, - runId: metadata?.runId ?? state.currentRunId, + runId, ipcDir: state.ipcDir, reason: metadata?.reason ?? 'unspecified', }, @@ -460,7 +460,7 @@ export class GroupQueue { logger.warn( { groupJid, - runId: metadata?.runId ?? state.currentRunId, + runId, ipcDir: state.ipcDir, reason: metadata?.reason ?? 'unspecified', err, diff --git a/src/message-agent-executor-paired.test.ts b/src/message-agent-executor-paired.test.ts index c7006e2..f159f01 100644 --- a/src/message-agent-executor-paired.test.ts +++ b/src/message-agent-executor-paired.test.ts @@ -25,6 +25,7 @@ vi.mock('./message-runtime-follow-up.js', () => ({ import type { AgentOutput } from './agent-runner.js'; import * as db from './db.js'; +import * as pairedExecutionContextModule from './paired-execution-context.js'; import { createPairedExecutionLifecycle } from './message-agent-executor-paired.js'; const log = { @@ -109,4 +110,94 @@ describe('createPairedExecutionLifecycle', () => { expect(outputs).toEqual([]); }); + + it('releases an owner turn interrupted by a human message without counting an owner failure', async () => { + const enqueueMessageCheck = vi.fn(); + vi.mocked(db.getPairedTaskById).mockReturnValue({ + id: 'paired-task-human-interrupted', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + title: null, + source_ref: 'HEAD', + plan_notes: null, + round_trip_count: 1, + review_requested_at: '2026-04-09T00:00:00.000Z', + status: 'active', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-04-09T00:00:00.000Z', + updated_at: '2026-04-09T00:00:01.000Z', + }); + + const lifecycle = createPairedExecutionLifecycle({ + pairedExecutionContext: { + task: { + id: 'paired-task-human-interrupted', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + title: null, + source_ref: 'HEAD', + plan_notes: null, + round_trip_count: 1, + review_requested_at: '2026-04-09T00:00:00.000Z', + status: 'active', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-04-09T00:00:00.000Z', + updated_at: '2026-04-09T00:00:00.000Z', + }, + workspace: null, + envOverrides: {}, + }, + pairedTurnIdentity: { + turnId: + 'paired-task-human-interrupted:2026-04-09T00:00:00.000Z:owner-turn', + taskId: 'paired-task-human-interrupted', + taskUpdatedAt: '2026-04-09T00:00:00.000Z', + intentKind: 'owner-turn', + role: 'owner', + }, + completedRole: 'owner', + chatJid: 'group@test', + runId: 'run-human-interrupted', + enqueueMessageCheck, + getCloseReason: () => 'human-message-detected', + log, + }); + + expect( + lifecycle.recordFinalOutputBeforeDelivery( + 'TASK_DONE\n부분 진행 결과를 닫기 전에 내보냅니다.', + ), + ).toBe(false); + lifecycle.updateSummary({ + outputText: '아비터 판단을 내리겠습니다.', + }); + lifecycle.markStatus('succeeded'); + lifecycle.markSawOutput(false); + await lifecycle.asyncFinalize(); + + expect( + pairedExecutionContextModule.completePairedExecutionContext, + ).not.toHaveBeenCalled(); + expect(db.insertPairedTurnOutput).not.toHaveBeenCalled(); + expect(db.releasePairedTaskExecutionLease).toHaveBeenCalledWith({ + taskId: 'paired-task-human-interrupted', + runId: 'run-human-interrupted', + }); + expect(db.failPairedTurn).toHaveBeenCalledWith({ + turnIdentity: expect.objectContaining({ + taskId: 'paired-task-human-interrupted', + role: 'owner', + }), + error: '아비터 판단을 내리겠습니다.', + }); + expect(enqueueMessageCheck).not.toHaveBeenCalled(); + }); }); diff --git a/src/message-agent-executor-paired.ts b/src/message-agent-executor-paired.ts index 0eb9732..9c5748a 100644 --- a/src/message-agent-executor-paired.ts +++ b/src/message-agent-executor-paired.ts @@ -19,12 +19,71 @@ import { resolvePairedFollowUpQueueAction } from './message-agent-executor-rules import { enqueuePairedFollowUpAfterEvent } from './message-runtime-follow-up.js'; import type { PairedTurnIdentity } from './paired-turn-identity.js'; import { resolvePairedTurnRunOwnership } from './paired-turn-run-ownership.js'; +import { isHumanMessageCloseReason } from './message-close-reasons.js'; import type { PairedRoomRole } from './types.js'; type ExecutorLog = Pick; const PAIRED_TASK_EXECUTION_LEASE_HEARTBEAT_MS = 30_000; +type PairedTaskRecord = NonNullable>; + +function releaseInterruptedPairedExecution( + taskId: string, + runId: string, + log: ExecutorLog, +): void { + log.info( + { pairedTaskId: taskId, runId }, + 'Released paired execution lease without counting a failure because a human message interrupted the turn', + ); + try { + releasePairedTaskExecutionLease({ taskId, runId }); + } catch (err) { + log.warn( + { pairedTaskId: taskId, runId, err }, + 'Failed to release paired execution lease after human interruption', + ); + } +} + +function completeStoredExecution( + taskId: string, + role: PairedRoomRole, + status: 'succeeded' | 'failed', + runId: string, + summary: string | null, +): void { + completePairedExecutionContext({ + taskId, + role, + status, + runId, + summary, + }); +} + +async function notifyPairedCompletionIfNeeded(args: { + task: PairedTaskRecord | null | undefined; + chatJid: string; + onOutput?: (output: AgentOutput) => Promise; +}): Promise { + if (args.task?.status !== 'completed' || !args.task.completion_reason) return; + const sender = getLastHumanMessageSender(args.chatJid); + const mention = sender ? `<@${sender}>` : ''; + const notifications: Record = { + escalated: `${mention} ⚠️ 자동 해결 불가 — 확인이 필요합니다.`, + }; + const message = notifications[args.task.completion_reason]; + if (!message) return; + await args.onOutput?.({ + status: 'success', + result: message, + output: { visibility: 'public', text: message }, + phase: 'final', + }); +} + export interface PairedExecutionLifecycle { updateSummary(args: { outputText?: string | null; @@ -47,6 +106,7 @@ export function createPairedExecutionLifecycle(args: { runId: string; enqueueMessageCheck: () => void; getDirectTerminalDeliveryText?: () => string | null; + getCloseReason?: () => string | null; onOutput?: (output: AgentOutput) => Promise; log: ExecutorLog; }): PairedExecutionLifecycle { @@ -58,6 +118,7 @@ export function createPairedExecutionLifecycle(args: { runId, enqueueMessageCheck, getDirectTerminalDeliveryText, + getCloseReason, onOutput, log, } = args; @@ -76,6 +137,8 @@ export function createPairedExecutionLifecycle(args: { pairedExecutionContext?.requiresVisibleVerdict === true; const missingVisibleVerdictSummary = 'Execution completed without a visible terminal verdict.'; + const wasInterruptedByHumanMessage = (): boolean => + isHumanMessageCloseReason(getCloseReason?.() ?? null); const currentRunOwnsActiveAttempt = (reason: string): boolean => { if (!pairedTurnIdentity) { @@ -276,6 +339,7 @@ export function createPairedExecutionLifecycle(args: { }, recordFinalOutputBeforeDelivery(outputText) { + if (wasInterruptedByHumanMessage()) return false; if (!currentRunOwnsActiveAttempt('streamed-final-output')) { return false; } @@ -376,26 +440,34 @@ export function createPairedExecutionLifecycle(args: { const sawOutputForFollowUp = missingVisibleVerdict ? false : pairedSawOutput; + const interruptedByHumanMessage = wasInterruptedByHumanMessage(); if (pairedExecutionContext && !pairedExecutionCompleted) { - if (effectiveStatus === 'succeeded') { - try { - persistPairedTurnOutputIfNeeded(); - } catch (err) { - log.warn( - { pairedTaskId: pairedExecutionContext.task.id, err }, - 'Failed to store paired turn output', - ); + if (interruptedByHumanMessage) { + releaseInterruptedPairedExecution( + pairedExecutionContext.task.id, + runId, + log, + ); + } else { + if (effectiveStatus === 'succeeded') { + try { + persistPairedTurnOutputIfNeeded(); + } catch (err) { + log.warn( + { pairedTaskId: pairedExecutionContext.task.id, err }, + 'Failed to store paired turn output', + ); + } } + completeStoredExecution( + pairedExecutionContext.task.id, + completedRole, + effectiveStatus, + runId, + pairedExecutionSummary, + ); } - - completePairedExecutionContext({ - taskId: pairedExecutionContext.task.id, - role: completedRole, - status: effectiveStatus, - runId, - summary: pairedExecutionSummary, - }); pairedExecutionCompleted = true; } @@ -407,27 +479,15 @@ export function createPairedExecutionLifecycle(args: { if (!pairedExecutionContext) { return; } - - const finishedTask = getPairedTaskById(pairedExecutionContext.task.id); - if ( - finishedTask?.status === 'completed' && - finishedTask.completion_reason - ) { - const sender = getLastHumanMessageSender(chatJid); - const mention = sender ? `<@${sender}>` : ''; - const notifications: Record = { - escalated: `${mention} ⚠️ 자동 해결 불가 — 확인이 필요합니다.`, - }; - const message = notifications[finishedTask.completion_reason]; - if (message) { - await onOutput?.({ - status: 'success', - result: message, - output: { visibility: 'public', text: message }, - phase: 'final', - }); - } + if (interruptedByHumanMessage) { + return; } + const finishedTask = getPairedTaskById(pairedExecutionContext.task.id); + await notifyPairedCompletionIfNeeded({ + task: finishedTask, + chatJid, + onOutput, + }); const queueAction = directTerminalOutput && diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index d230c07..6b85cdb 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -92,7 +92,12 @@ async function enrichArbiterPromptWithMoa(args: { export interface MessageAgentExecutorDeps { assistantName: string; queue: Pick & - Partial>; + Partial< + Pick< + GroupQueue, + 'getDirectTerminalDeliveryForRun' | 'getCloseReasonForRun' + > + >; getRoomBindings: () => Record; getSessions: () => Record; persistSession: (groupFolder: string, sessionId: string) => void; @@ -180,6 +185,8 @@ export async function runAgentForGroup( runId, runtimePairedTurnIdentity?.role ?? activeRole, ) ?? null, + getCloseReason: () => + deps.queue.getCloseReasonForRun?.(chatJid, runId) ?? null, onOutput, log, }); diff --git a/src/message-close-reasons.ts b/src/message-close-reasons.ts new file mode 100644 index 0000000..2c8f18a --- /dev/null +++ b/src/message-close-reasons.ts @@ -0,0 +1,7 @@ +export const HUMAN_MESSAGE_DETECTED_CLOSE_REASON = 'human-message-detected'; + +export function isHumanMessageCloseReason( + reason: string | null | undefined, +): boolean { + return reason === HUMAN_MESSAGE_DETECTED_CLOSE_REASON; +} diff --git a/src/message-runtime-dispatch.ts b/src/message-runtime-dispatch.ts index e6b3538..6543784 100644 --- a/src/message-runtime-dispatch.ts +++ b/src/message-runtime-dispatch.ts @@ -4,6 +4,7 @@ import { buildQueuedTurnDispatch, executeBotOnlyPairedFollowUpAction, } from './message-runtime-flow.js'; +import { HUMAN_MESSAGE_DETECTED_CLOSE_REASON } from './message-close-reasons.js'; import { advanceLastAgentCursor, filterLoopingPairedBotMessages, @@ -305,7 +306,7 @@ export async function processLoopGroupMessages(args: { if (hasExternalHumanMessage(processableGroupMessages)) { const interruptedActiveRun = args.isRunningMessageTurn(chatJid); if (interruptedActiveRun) { - args.closeStdin('human-message-detected'); + args.closeStdin(HUMAN_MESSAGE_DETECTED_CLOSE_REASON); } args.enqueueMessageCheck(); logger.info( diff --git a/src/message-runtime-turns.ts b/src/message-runtime-turns.ts index 97e87c8..78d9d1b 100644 --- a/src/message-runtime-turns.ts +++ b/src/message-runtime-turns.ts @@ -204,6 +204,8 @@ export function createExecuteTurn(deps: CreateExecuteTurnDeps): ExecuteTurnFn { }); return ownership.state !== 'inactive'; }, + getCloseReason: () => + deps.queue.getCloseReasonForRun?.(chatJid, runId) ?? null, deliverFinalText: async (text, options) => { try { return await deps.deliverFinalText({ diff --git a/src/message-turn-controller.test.ts b/src/message-turn-controller.test.ts index a48f153..2de62d1 100644 --- a/src/message-turn-controller.test.ts +++ b/src/message-turn-controller.test.ts @@ -803,3 +803,53 @@ describe('MessageTurnController outbound audit logging', () => { }); }); }); + +describe('MessageTurnController human interruptions', () => { + beforeEach(() => { + vi.resetAllMocks(); + }); + + it('does not replay owner progress as final when a human message interrupted the turn', async () => { + const channel = { ...makeChannel(), name: 'discord' } satisfies Channel; + const deliverFinalText = vi.fn().mockResolvedValue(true); + const controller = new MessageTurnController({ + chatJid: 'dc:test-room', + group: makeGroup(), + runId: 'run-owner-human-interrupted', + channel, + idleTimeout: 1_000, + failureFinalText: '실패', + isClaudeCodeAgent: true, + clearSession: vi.fn(), + requestClose: vi.fn(), + deliverFinalText, + getCloseReason: () => 'human-message-detected', + deliveryRole: 'owner', + pairedTurnIdentity: { + turnId: 'task-1:2026-04-10T14:22:00.000Z:owner-turn', + taskId: 'task-1', + taskUpdatedAt: '2026-04-10T14:22:00.000Z', + intentKind: 'owner-turn', + role: 'owner', + }, + }); + + await controller.start(); + await controller.handleOutput({ + status: 'success', + phase: 'progress', + result: '아비터 판단을 내리겠습니다.', + } as any); + await controller.handleOutput({ + status: 'success', + phase: 'progress', + result: '추가 확인 중입니다.', + } as any); + await flushAsync(); + + await controller.finish('success'); + + expect(channel.sendAndTrack).toHaveBeenCalledTimes(1); + expect(deliverFinalText).not.toHaveBeenCalled(); + }); +}); diff --git a/src/message-turn-controller.ts b/src/message-turn-controller.ts index c7cfe5b..c8d51b0 100644 --- a/src/message-turn-controller.ts +++ b/src/message-turn-controller.ts @@ -9,6 +9,7 @@ import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js'; import { formatElapsedKorean } from './utils.js'; import type { PairedTurnIdentity } from './paired-turn-identity.js'; +import { isHumanMessageCloseReason } from './message-close-reasons.js'; import { normalizeAgentOutputPhase, toVisiblePhase, @@ -45,6 +46,7 @@ interface MessageTurnControllerOptions { }, ) => Promise; canDeliverFinalText?: () => boolean; + getCloseReason?: () => string | null; allowProgressReplayWithoutFinal?: boolean; deliveryRole?: PairedRoomRole | null; deliveryServiceId?: string | null; @@ -350,11 +352,9 @@ export class MessageTurnController { visiblePhase: VisiblePhase; }> { await this.deactivateTyping('turn:finish', { outputStatus }); - if (outputStatus === 'error') { this.hadError = true; } - if ( outputStatus === 'success' && this.visiblePhase === 'progress' && @@ -362,7 +362,9 @@ export class MessageTurnController { this.latestProgressTextForFinal ) { const replayText = this.latestProgressTextForFinal; - if (this.options.allowProgressReplayWithoutFinal !== false) { + if (isHumanMessageCloseReason(this.options.getCloseReason?.() ?? null)) { + this.resetProgressState(); + } else if (this.options.allowProgressReplayWithoutFinal !== false) { this.log.info( 'Sending a separate final message from the last progress output after agent completion', ); @@ -787,13 +789,11 @@ export class MessageTurnController { } await this.publishTerminalText(this.options.failureFinalText); } - private requestAgentClose(reason: string): void { if (this.closeRequested) return; this.closeRequested = true; this.options.requestClose(reason); } - private async sendProgressMessage(text: string): Promise { if ( this.options.canDeliverFinalText &&