Merge pull request #165 from phj1081/codex/owner/ejclaw

fix: handle paired preemption and task scope prompts
2026-05-26 03:28:42 +09:00
parent 5a43cc531a c8f99d8349
commit d770c40862
14 changed files with 484 additions and 67 deletions
--- 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<string, string>;
  recentDirectTerminalDeliveries: Map<string, Map<string, string>>;
  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;
--- 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<boolean>((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();
+  });
+});
--- 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,
--- 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();
+  });
 });
--- 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<typeof logger, 'info' | 'warn'>;

 const PAIRED_TASK_EXECUTION_LEASE_HEARTBEAT_MS = 30_000;

+type PairedTaskRecord = NonNullable<ReturnType<typeof getPairedTaskById>>;
+
+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<void>;
+}): Promise<void> {
+  if (args.task?.status !== 'completed' || !args.task.completion_reason) return;
+  const sender = getLastHumanMessageSender(args.chatJid);
+  const mention = sender ? `<@${sender}>` : '';
+  const notifications: Record<string, string> = {
+    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<void>;
  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<string, string> = {
-          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 &&
--- 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<GroupQueue, 'registerProcess' | 'enqueueMessageCheck'> &
-    Partial<Pick<GroupQueue, 'getDirectTerminalDeliveryForRun'>>;
+    Partial<
+      Pick<
+        GroupQueue,
+        'getDirectTerminalDeliveryForRun' | 'getCloseReasonForRun'
+      >
+    >;
  getRoomBindings: () => Record<string, RegisteredGroup>;
  getSessions: () => Record<string, string>;
  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,
  });
--- a/src/message-close-reasons.ts
+++ 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;
+}
--- 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(
--- 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,
--- 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',
--- 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({
--- 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({
--- 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();
+  });
+});
--- 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<boolean>;
  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<void> {
    if (
      this.options.canDeliverFinalText &&