Deduplicate direct terminal reviewer deliveries

2026-04-05 06:57:39 +09:00
parent da6e428650
commit e560fe6536
3 changed files with 264 additions and 1 deletions
--- 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<string>;
  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);
--- 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');
--- 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,