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

Extract message runtime final delivery

src/message-runtime-final-delivery.test.ts

@@ -0,0 +1,216 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+vi.mock('./db.js', () => ({
+  createProducedWorkItem: vi.fn((args) => ({
+    id: 123,
+    chat_jid: args.chat_jid,
+    delivery_role: args.delivery_role,
+    result_payload: args.result_payload,
+    attachments: args.attachments,
+    delivery_attempts: 0,
+  })),
+}));
+
+vi.mock('./logger.js', () => ({
+  logger: {
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+  },
+}));
+
+vi.mock('./message-runtime-delivery.js', () => ({
+  deliverOpenWorkItem: vi.fn(() => Promise.resolve(true)),
+}));
+
+import { createProducedWorkItem } from './db.js';
+import { logger } from './logger.js';
+import { deliverOpenWorkItem } from './message-runtime-delivery.js';
+import { deliverMessageRuntimeFinalText } from './message-runtime-final-delivery.js';
+import type { Channel, RegisteredGroup } from './types.js';
+
+function makeChannel(): Channel {
+  return {
+    name: 'discord-owner',
+    connect: vi.fn(),
+    sendMessage: vi.fn(),
+    isConnected: vi.fn(() => true),
+    ownsJid: vi.fn(() => true),
+    disconnect: vi.fn(),
+  } as unknown as Channel;
+}
+
+function makeGroup(overrides: Partial<RegisteredGroup> = {}): RegisteredGroup {
+  return {
+    name: 'Room',
+    folder: 'room-folder',
+    added_at: '2026-04-29T01:00:00.000Z',
+    ...overrides,
+  };
+}
+
+describe('message-runtime-final-delivery', () => {
+  beforeEach(() => {
+    vi.resetAllMocks();
+    vi.mocked(deliverOpenWorkItem).mockResolvedValue(true);
+    vi.mocked(createProducedWorkItem).mockImplementation(
+      (args: any) =>
+        ({
+          id: 123,
+          chat_jid: args.chat_jid,
+          delivery_role: args.delivery_role,
+          result_payload: args.result_payload,
+          attachments: args.attachments,
+          delivery_attempts: 0,
+        }) as any,
+    );
+  });
+
+  it.each(['reviewer', 'arbiter'] as const)(
+    'skips %s work item delivery after direct terminal IPC delivery',
+    async (deliveryRole) => {
+      const hasDirectTerminalDeliveryForRun = vi.fn(() => true);
+
+      const result = await deliverMessageRuntimeFinalText({
+        text: `${deliveryRole} final`,
+        chatJid: 'chat-1',
+        runId: 'run-1',
+        channel: makeChannel(),
+        group: makeGroup(),
+        startSeq: 10,
+        endSeq: 11,
+        deliveryRole,
+        deliveryServiceId: 'codex-main',
+        hasDirectTerminalDeliveryForRun,
+        isDuplicateOfLastBotFinal: vi.fn(),
+        openContinuation: vi.fn(),
+      });
+
+      expect(result).toBe(true);
+      expect(hasDirectTerminalDeliveryForRun).toHaveBeenCalledWith(
+        'chat-1',
+        'run-1',
+        deliveryRole,
+      );
+      expect(createProducedWorkItem).not.toHaveBeenCalled();
+      expect(deliverOpenWorkItem).not.toHaveBeenCalled();
+      expect(logger.info).toHaveBeenCalledWith(
+        expect.objectContaining({
+          chatJid: 'chat-1',
+          runId: 'run-1',
+          deliveryRole,
+        }),
+        'Skipping final work item delivery because this run already sent a direct terminal IPC message',
+      );
+    },
+  );
+
+  it('creates and delivers a produced work item with default agent type and attachment roots', async () => {
+    const channel = makeChannel();
+    const isDuplicateOfLastBotFinal = vi.fn(() => false);
+    const openContinuation = vi.fn();
+    const attachments = [
+      {
+        path: '/tmp/image.png',
+        name: 'image.png',
+        mime: 'image/png',
+      },
+    ];
+
+    const result = await deliverMessageRuntimeFinalText({
+      text: 'owner final',
+      attachments,
+      chatJid: 'chat-1',
+      runId: 'run-owner',
+      channel,
+      group: makeGroup({ workDir: '/work/room' }),
+      startSeq: 1,
+      endSeq: 2,
+      deliveryRole: 'owner',
+      deliveryServiceId: null,
+      hasDirectTerminalDeliveryForRun: vi.fn(() => true),
+      isDuplicateOfLastBotFinal,
+      openContinuation,
+    });
+
+    expect(result).toBe(true);
+    expect(createProducedWorkItem).toHaveBeenCalledWith({
+      group_folder: 'room-folder',
+      chat_jid: 'chat-1',
+      agent_type: 'claude-code',
+      service_id: undefined,
+      delivery_role: 'owner',
+      start_seq: 1,
+      end_seq: 2,
+      result_payload: 'owner final',
+      attachments,
+    });
+    expect(deliverOpenWorkItem).toHaveBeenCalledWith(
+      expect.objectContaining({
+        channel,
+        log: logger,
+        attachmentBaseDirs: ['/work/room'],
+        replaceMessageId: undefined,
+        isDuplicateOfLastBotFinal,
+        openContinuation,
+      }),
+    );
+  });
+
+  it('preserves forced agent, delivery service, and replacement message metadata', async () => {
+    const channel = makeChannel();
+
+    await deliverMessageRuntimeFinalText({
+      text: 'forced final',
+      chatJid: 'chat-1',
+      runId: 'run-forced',
+      channel,
+      group: makeGroup({ agentType: 'claude-code' }),
+      startSeq: null,
+      endSeq: null,
+      forcedAgentType: 'codex',
+      deliveryRole: 'arbiter',
+      deliveryServiceId: 'codex-main',
+      replaceMessageId: 'discord-message-1',
+      hasDirectTerminalDeliveryForRun: vi.fn(() => false),
+      isDuplicateOfLastBotFinal: vi.fn(),
+      openContinuation: vi.fn(),
+    });
+
+    expect(createProducedWorkItem).toHaveBeenCalledWith(
+      expect.objectContaining({
+        agent_type: 'codex',
+        service_id: 'codex-main',
+        delivery_role: 'arbiter',
+        start_seq: null,
+        end_seq: null,
+      }),
+    );
+    expect(deliverOpenWorkItem).toHaveBeenCalledWith(
+      expect.objectContaining({
+        replaceMessageId: 'discord-message-1',
+        attachmentBaseDirs: undefined,
+      }),
+    );
+  });
+
+  it('returns the deliverOpenWorkItem result', async () => {
+    vi.mocked(deliverOpenWorkItem).mockResolvedValue(false);
+
+    await expect(
+      deliverMessageRuntimeFinalText({
+        text: 'failed final',
+        chatJid: 'chat-1',
+        runId: 'run-failed',
+        channel: makeChannel(),
+        group: makeGroup(),
+        startSeq: 1,
+        endSeq: 2,
+        deliveryRole: null,
+        deliveryServiceId: null,
+        isDuplicateOfLastBotFinal: vi.fn(),
+        openContinuation: vi.fn(),
+      }),
+    ).resolves.toBe(false);
+  });
+});

src/message-runtime-final-delivery.ts

@@ -0,0 +1,73 @@
+import { createProducedWorkItem } from './db.js';
+import { logger } from './logger.js';
+import { deliverOpenWorkItem } from './message-runtime-delivery.js';
+import type {
+  AgentType,
+  Channel,
+  OutboundAttachment,
+  PairedRoomRole,
+  RegisteredGroup,
+} from './types.js';
+
+export async function deliverMessageRuntimeFinalText(args: {
+  text: string;
+  attachments?: OutboundAttachment[];
+  chatJid: string;
+  runId: string;
+  channel: Channel;
+  group: RegisteredGroup;
+  startSeq: number | null;
+  endSeq: number | null;
+  forcedAgentType?: AgentType;
+  deliveryRole: PairedRoomRole | null;
+  deliveryServiceId: string | null;
+  replaceMessageId?: string | null;
+  hasDirectTerminalDeliveryForRun?: (
+    chatJid: string,
+    runId: string,
+    deliveryRole: PairedRoomRole | null,
+  ) => boolean;
+  isDuplicateOfLastBotFinal: (chatJid: string, text: string) => boolean;
+  openContinuation: (chatJid: string) => void;
+}): Promise<boolean> {
+  if (
+    (args.deliveryRole === 'reviewer' || args.deliveryRole === 'arbiter') &&
+    args.hasDirectTerminalDeliveryForRun?.(
+      args.chatJid,
+      args.runId,
+      args.deliveryRole,
+    )
+  ) {
+    logger.info(
+      {
+        chatJid: args.chatJid,
+        runId: args.runId,
+        deliveryRole: args.deliveryRole,
+      },
+      'Skipping final work item delivery because this run already sent a direct terminal IPC message',
+    );
+    return true;
+  }
+
+  const workItem = createProducedWorkItem({
+    group_folder: args.group.folder,
+    chat_jid: args.chatJid,
+    agent_type: args.forcedAgentType ?? args.group.agentType ?? 'claude-code',
+    service_id: args.deliveryServiceId ?? undefined,
+    delivery_role: args.deliveryRole,
+    start_seq: args.startSeq,
+    end_seq: args.endSeq,
+    result_payload: args.text,
+    attachments: args.attachments,
+  });
+
+  return deliverOpenWorkItem({
+    channel: args.channel,
+    item: workItem,
+    log: logger,
+    attachmentBaseDirs: args.group.workDir ? [args.group.workDir] : undefined,
+    replaceMessageId: args.replaceMessageId,
+    isDuplicateOfLastBotFinal: args.isDuplicateOfLastBotFinal,
+    openContinuation: args.openContinuation,
+  });
+}

src/message-runtime.ts

@@ -1,5 +1,4 @@
 import {
-  createProducedWorkItem,
   getOpenWorkItemForChat,
   getMessagesSinceSeq,
   getLatestOpenPairedTaskForChat,
@@ -15,10 +14,7 @@ import { GroupQueue, GroupRunContext } from './group-queue.js';
 import { findChannel, formatMessages } from './router.js';
 import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js';
 import { enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp } from './message-runtime-dispatch.js';
-import {
+import { processOpenWorkItemDelivery } from './message-runtime-delivery.js';
-  deliverOpenWorkItem,
-  processOpenWorkItemDelivery,
-} from './message-runtime-delivery.js';
 import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js';
 import { handleQueuedRunGates } from './message-runtime-gating.js';
 import {
@@ -74,6 +70,7 @@ import {
   hasHumanMessageAfterWorkItem,
 } from './message-runtime-preflight-messages.js';
 import { handleMessageRuntimeAfterDeliverySuccess } from './message-runtime-after-delivery.js';
+import { deliverMessageRuntimeFinalText } from './message-runtime-final-delivery.js';
 
 export { isDuplicateOfLastBotFinal };
 
@@ -191,41 +188,21 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
       deliveryServiceId,
       replaceMessageId,
     }) => {
-      if (
+      return deliverMessageRuntimeFinalText({
-        (deliveryRole === 'reviewer' || deliveryRole === 'arbiter') &&
+        text,
-        deps.queue.hasDirectTerminalDeliveryForRun?.(
-          chatJid,
-          runId,
-          deliveryRole,
-        )
-      ) {
-        logger.info(
-          {
-            chatJid,
-            runId,
-            deliveryRole,
-          },
-          '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,
-        agent_type: forcedAgentType ?? group.agentType ?? 'claude-code',
-        service_id: deliveryServiceId ?? undefined,
-        delivery_role: deliveryRole,
-        start_seq: startSeq,
-        end_seq: endSeq,
-        result_payload: text,
         attachments,
-      });
+        chatJid,
-      return deliverOpenWorkItem({
+        runId,
         channel,
-        item: workItem,
+        group,
-        log: logger,
+        startSeq,
-        attachmentBaseDirs: group.workDir ? [group.workDir] : undefined,
+        endSeq,
+        forcedAgentType,
+        deliveryRole,
+        deliveryServiceId,
         replaceMessageId,
+        hasDirectTerminalDeliveryForRun:
+          deps.queue.hasDirectTerminalDeliveryForRun?.bind(deps.queue),
         isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal,
         openContinuation: (targetChatJid) =>
           continuationTracker.open(targetChatJid),