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-flow.ts b/src/message-runtime-flow.ts index 35480af..c794a46 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -137,6 +137,7 @@ export function buildPendingPairedTurn(args: { turnOutputs, recentHumanMessages, lastHumanMessage, + taskCreatedAt: task.created_at, }), channel: resolveChannel(taskStatus), cursor, @@ -188,6 +189,7 @@ export function buildPendingPairedTurn(args: { turnOutputs, recentHumanMessages, lastHumanMessage, + taskCreatedAt: task.created_at, }), channel: resolveChannel(taskStatus), cursor, diff --git a/src/message-runtime-prompts.test.ts b/src/message-runtime-prompts.test.ts index 1d04561..5a2335e 100644 --- a/src/message-runtime-prompts.test.ts +++ b/src/message-runtime-prompts.test.ts @@ -12,14 +12,17 @@ import type { NewMessage, PairedTask, PairedTurnOutput } from './types.js'; const CARRY_FORWARD_MARKER = '[Carried forward context from the previous task: latest owner final]'; -function makeHumanMessage(content: string): NewMessage { +function makeHumanMessage( + content: string, + timestamp: string = '2026-04-20T01:00:00.000Z', +): NewMessage { return { id: `msg-${content}`, chat_jid: 'group@test', sender: 'user@test', sender_name: 'User', content, - timestamp: '2026-04-20T01:00:00.000Z', + timestamp, is_bot_message: false, is_from_me: false, }; @@ -134,6 +137,32 @@ describe('message-runtime-prompts output-only context', () => { expect(prompt).not.toContain('과거 요청을 다시 기준으로 보면 안 됨'); }); + it('includes current task user scope in reviewer pending prompts without pulling older human messages', () => { + const prompt = buildReviewerPendingPrompt({ + chatJid: 'group@test', + timezone: 'UTC', + turnOutputs: [ + makeTurnOutput('TASK_DONE\n현재 owner 결과', 'owner', { + turn_number: 1, + created_at: '2026-04-20T02:10:00.000Z', + }), + ], + recentHumanMessages: [ + makeHumanMessage('이전 작업의 gap 요청', '2026-04-20T01:50:00.000Z'), + makeHumanMessage( + '뒤쪽 트레일 주변 디졸브 넣어줘', + '2026-04-20T02:00:01.000Z', + ), + ], + lastHumanMessage: '뒤쪽 트레일 주변 디졸브 넣어줘', + taskCreatedAt: '2026-04-20T02:00:00.000Z', + }); + + expect(prompt).toContain('뒤쪽 트레일 주변 디졸브 넣어줘'); + expect(prompt).toContain('현재 owner 결과'); + expect(prompt).not.toContain('이전 작업의 gap 요청'); + }); + it('prepends a carry-forward warning to owner pending prompts', () => { const prompt = buildOwnerPendingPrompt({ chatJid: 'group@test', @@ -175,6 +204,38 @@ describe('message-runtime-prompts output-only context', () => { expect(prompt).not.toContain('이전 작업의 사용자 메시지'); }); + it('includes current task user scope in owner pending prompts without pulling older human messages', () => { + const prompt = buildOwnerPendingPrompt({ + chatJid: 'group@test', + timezone: 'UTC', + turnOutputs: [ + makeTurnOutput('DONE_WITH_CONCERNS\n현재 reviewer 피드백', 'reviewer', { + id: 2, + turn_number: 2, + created_at: '2026-04-20T02:11:00.000Z', + }), + ], + recentHumanMessages: [ + makeHumanMessage( + '이전 작업의 사용자 메시지', + '2026-04-20T01:50:00.000Z', + ), + makeHumanMessage( + '현재 task에서 추가한 조건', + '2026-04-20T02:00:01.000Z', + ), + ], + lastHumanMessage: '현재 task에서 추가한 조건', + taskCreatedAt: '2026-04-20T02:00:00.000Z', + }); + + expect(prompt).toContain('현재 task에서 추가한 조건'); + expect(prompt).toContain('현재 reviewer 피드백'); + expect(prompt).not.toContain('이전 작업의 사용자 메시지'); + }); +}); + +describe('message-runtime-prompts arbiter output context', () => { it('keeps arbiter prompts output-only when turn outputs exist', () => { const prompt = buildArbiterPromptForTask({ task: makeTask({ status: 'arbiter_requested' }), @@ -189,8 +250,12 @@ describe('message-runtime-prompts output-only context', () => { turn_number: 2, }), ], - recentMessages: [makeHumanMessage('예전 유저 지시')], - labeledRecentMessages: [makeHumanMessage('예전 유저 지시')], + recentMessages: [ + makeHumanMessage('예전 유저 지시', '2026-04-20T00:50:00.000Z'), + ], + labeledRecentMessages: [ + makeHumanMessage('예전 유저 지시', '2026-04-20T00:50:00.000Z'), + ], }); expect(prompt).toContain('owner 주장'); @@ -215,8 +280,12 @@ describe('message-runtime-prompts output-only context', () => { }, ); }), - recentMessages: [makeHumanMessage('예전 유저 지시')], - labeledRecentMessages: [makeHumanMessage('예전 유저 지시')], + recentMessages: [ + makeHumanMessage('예전 유저 지시', '2026-04-20T00:50:00.000Z'), + ], + labeledRecentMessages: [ + makeHumanMessage('예전 유저 지시', '2026-04-20T00:50:00.000Z'), + ], }); expect(prompt).not.toContain('arbiter-context-output-01'); @@ -226,6 +295,46 @@ describe('message-runtime-prompts output-only context', () => { expect(prompt).not.toContain('예전 유저 지시'); }); + it('includes current task user scope in arbiter prompts while keeping the output cap', () => { + const prompt = buildArbiterPromptForTask({ + task: makeTask({ + status: 'arbiter_requested', + created_at: '2026-04-20T02:00:00.000Z', + }), + chatJid: 'group@test', + timezone: 'UTC', + turnOutputs: Array.from({ length: 8 }, (_, index) => { + const turnNumber = index + 1; + return makeTurnOutput( + `arbiter-context-output-${String(turnNumber).padStart(2, '0')}`, + turnNumber % 2 === 0 ? 'reviewer' : 'owner', + { + id: turnNumber, + turn_number: turnNumber, + created_at: `2026-04-20T02:${String(turnNumber).padStart(2, '0')}:00.000Z`, + }, + ); + }), + recentMessages: [ + makeHumanMessage('이전 작업의 유저 지시', '2026-04-20T01:50:00.000Z'), + makeHumanMessage( + '현재 task의 트레일 디졸브 요청', + '2026-04-20T02:00:01.000Z', + ), + ], + labeledRecentMessages: [], + }); + + expect(prompt).toContain('현재 task의 트레일 디졸브 요청'); + expect(prompt).not.toContain('이전 작업의 유저 지시'); + expect(prompt).not.toContain('arbiter-context-output-01'); + expect(prompt).not.toContain('arbiter-context-output-02'); + expect(prompt).toContain('arbiter-context-output-03'); + expect(prompt).toContain('arbiter-context-output-08'); + }); +}); + +describe('message-runtime-prompts prompt hygiene', () => { it('preserves turn output order instead of timestamp interleaving', () => { const prompt = buildReviewerPendingPrompt({ chatJid: 'group@test', diff --git a/src/message-runtime-prompts.ts b/src/message-runtime-prompts.ts index 58f0d5a..abe836e 100644 --- a/src/message-runtime-prompts.ts +++ b/src/message-runtime-prompts.ts @@ -9,6 +9,7 @@ const CARRIED_FORWARD_OWNER_FINAL_GUIDANCE = `System note: If you see a message beginning with "${CARRIED_FORWARD_OWNER_FINAL_MARKER}", treat it as background only. Do not repeat, continue, or answer that carried-forward final directly. Respond only to the latest human request and the current task.`; const ARBITER_TURN_OUTPUT_CONTEXT_LIMIT = 6; +const TASK_USER_CONTEXT_START_SKEW_MS = 5_000; function turnOutputsToMessages( outputs: PairedTurnOutput[], @@ -47,6 +48,23 @@ function latestTurnOutputs( return outputs.slice(-limit); } +function currentTaskHumanMessages( + messages: NewMessage[], + taskCreatedAt: string | null | undefined, +): NewMessage[] { + if (!taskCreatedAt) return []; + const taskStartMs = Date.parse(taskCreatedAt); + if (!Number.isFinite(taskStartMs)) return []; + return messages.filter((message) => { + if (message.is_bot_message) return false; + const messageMs = Date.parse(message.timestamp); + return ( + Number.isFinite(messageMs) && + messageMs >= taskStartMs - TASK_USER_CONTEXT_START_SKEW_MS + ); + }); +} + function hasCarriedForwardOwnerFinal(outputs: PairedTurnOutput[]): boolean { return outputs.some((output) => output.output_text.startsWith(CARRIED_FORWARD_OWNER_FINAL_MARKER), @@ -95,9 +113,13 @@ function turnOutputsOnlyPrompt( chatJid: string, timezone: string, turnOutputs: PairedTurnOutput[], + taskHumanMessages: NewMessage[] = [], ): string { return prependCarriedForwardGuidance( - formatMessages(turnOutputsToMessages(turnOutputs, chatJid), timezone), + formatMessages( + [...taskHumanMessages, ...turnOutputsToMessages(turnOutputs, chatJid)], + timezone, + ), turnOutputs, ); } @@ -108,9 +130,15 @@ export function buildReviewerPendingPrompt(args: { turnOutputs: PairedTurnOutput[]; recentHumanMessages: NewMessage[]; lastHumanMessage: string | null | undefined; + taskCreatedAt?: string | null; }): string { if (args.turnOutputs.length > 0) { - return turnOutputsOnlyPrompt(args.chatJid, args.timezone, args.turnOutputs); + return turnOutputsOnlyPrompt( + args.chatJid, + args.timezone, + args.turnOutputs, + currentTaskHumanMessages(args.recentHumanMessages, args.taskCreatedAt), + ); } if (!args.lastHumanMessage) { @@ -126,9 +154,15 @@ export function buildOwnerPendingPrompt(args: { turnOutputs: PairedTurnOutput[]; recentHumanMessages: NewMessage[]; lastHumanMessage: string | null | undefined; + taskCreatedAt?: string | null; }): string { if (args.turnOutputs.length > 0) { - return turnOutputsOnlyPrompt(args.chatJid, args.timezone, args.turnOutputs); + return turnOutputsOnlyPrompt( + args.chatJid, + args.timezone, + args.turnOutputs, + currentTaskHumanMessages(args.recentHumanMessages, args.taskCreatedAt), + ); } if (!args.lastHumanMessage) { @@ -146,15 +180,22 @@ export function buildArbiterPromptForTask(args: { recentMessages: NewMessage[]; labeledRecentMessages: NewMessage[]; }): string { + const taskHumanMessages = currentTaskHumanMessages( + args.recentMessages, + args.task.created_at, + ); const messages = args.turnOutputs.length > 0 - ? turnOutputsToMessages( - latestTurnOutputs( - args.turnOutputs, - ARBITER_TURN_OUTPUT_CONTEXT_LIMIT, + ? [ + ...taskHumanMessages, + ...turnOutputsToMessages( + latestTurnOutputs( + args.turnOutputs, + ARBITER_TURN_OUTPUT_CONTEXT_LIMIT, + ), + args.chatJid, ), - args.chatJid, - ) + ] : args.labeledRecentMessages; return buildArbiterContextPrompt({ 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 &&