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

Route ordinary outbound through canonical delivery
2026-04-29 00:18:54 +09:00
parent 32605e2fda 4daa1382f0
commit 9655501bdf
5 changed files with 180 additions and 88 deletions
--- a/src/index.test.ts
+++ b/src/index.test.ts
@@ -3,7 +3,6 @@ import { describe, expect, it, vi } from 'vitest';
 import {
  editFormattedTrackedChannelMessage,
  isTerminalStatusMessage,
-  sendFormattedChannelMessage,
  sendFormattedTrackedChannelMessage,
 } from './index.js';
 import { composeDashboardContent } from './dashboard-render.js';
@@ -81,12 +80,6 @@ describe('index scheduler messaging helpers', () => {
      'msg-123',
      '<internal>only hidden</internal>',
    );
-    await sendFormattedChannelMessage(
-      [channel],
-      'dc:test',
-      '<internal>only hidden</internal>',
-    );
-
    expect(trackedResult).toBeNull();
    expect(sendAndTrack).not.toHaveBeenCalled();
    expect(editMessage).not.toHaveBeenCalled();
--- a/src/index.ts
+++ b/src/index.ts
@@ -30,10 +30,12 @@ import { composeDashboardContent } from './dashboard-render.js';
 import { GroupQueue } from './group-queue.js';
 import { resolveGroupIpcPath } from './group-folder.js';
 import { startIpcWatcher } from './ipc.js';
-import { deliverIpcOutboundMessage } from './ipc-outbound-delivery.js';
+import {
+  deliverCanonicalOutboundMessage,
+  deliverIpcOutboundMessage,
+} from './ipc-outbound-delivery.js';
 import {
  findChannel,
-  findChannelByName,
  formatOutbound,
  normalizeMessageForDedupe,
 } from './router.js';
@@ -90,20 +92,6 @@ import { FAILOVER_MIN_DURATION_MS } from './config.js';

 // Token rotation is initialized lazily on first use or at startup below

-export async function sendFormattedChannelMessage(
-  channels: Channel[],
-  jid: string,
-  rawText: string,
-): Promise<void> {
-  const channel = findChannel(channels, jid);
-  if (!channel) {
-    logger.warn({ jid }, 'No channel owns JID, cannot send message');
-    return;
-  }
-  const text = formatOutbound(rawText);
-  if (text) await channel.sendMessage(jid, text);
-}
-
 export async function sendFormattedTrackedChannelMessage(
  channels: Channel[],
  jid: string,
@@ -156,6 +144,23 @@ const runtime = createMessageRuntime({
  clearSession: runtimeState.clearSession,
 });

+async function deliverFormattedCanonicalMessage(
+  jid: string,
+  rawText: string,
+  deliveryRole?: 'owner' | 'reviewer' | 'arbiter',
+): Promise<void> {
+  const text = formatOutbound(rawText);
+  if (!text) return;
+  await deliverCanonicalOutboundMessage(
+    { jid, text, deliveryRole },
+    {
+      channels,
+      roomBindings: runtimeState.getRoomBindings,
+      log: logger,
+    },
+  );
+}
+
 /**
 * Get available groups list for the agent.
 * Returns groups ordered by most recent activity.
@@ -192,8 +197,7 @@ async function announceRestartRecovery(
      return explicitContext;
    }

-    await sendFormattedChannelMessage(
-      channels,
+    await deliverFormattedCanonicalMessage(
      explicitContext.chatJid,
      buildRestartAnnouncement(explicitContext),
    );
@@ -207,8 +211,7 @@ async function announceRestartRecovery(
      if (hasRecentRestartAnnouncement(interrupted.chatJid, dedupeSince)) {
        continue;
      }
-      await sendFormattedChannelMessage(
-        channels,
+      await deliverFormattedCanonicalMessage(
        interrupted.chatJid,
        buildInterruptedRestartAnnouncement(interrupted),
      );
@@ -230,8 +233,7 @@ async function announceRestartRecovery(
    return null;
  }

-  await sendFormattedChannelMessage(
-    channels,
+  await deliverFormattedCanonicalMessage(
    inferred.chatJid,
    inferred.lines.join('\n'),
  );
@@ -360,13 +362,6 @@ async function main(): Promise<void> {
  }

  // Start subsystems (independently of connection handler)
-  // Paired-room scheduler output goes through the reviewer bot slot.
-  const reviewerChannelName = 'discord-review';
-  const reviewerChannelForCron = findChannelByName(
-    channels,
-    reviewerChannelName,
-  );
-
  startSchedulerLoop({
    roomBindings: runtimeState.getRoomBindings,
    getSessions: runtimeState.getSessions,
@@ -374,13 +369,9 @@ async function main(): Promise<void> {
    onProcess: (groupJid, proc, processName, ipcDir) =>
      queue.registerProcess(groupJid, proc, processName, ipcDir),
    sendMessage: (jid, rawText) =>
-      sendFormattedChannelMessage(channels, jid, rawText),
-    sendMessageViaReviewerBot: reviewerChannelForCron
-      ? async (jid, rawText) => {
-          const text = formatOutbound(rawText);
-          if (text) await reviewerChannelForCron.sendMessage(jid, text);
-        }
-      : undefined,
+      deliverFormattedCanonicalMessage(jid, rawText),
+    sendMessageViaReviewerBot: (jid, rawText) =>
+      deliverFormattedCanonicalMessage(jid, rawText, 'reviewer'),
    sendTrackedMessage: (jid, rawText) =>
      sendFormattedTrackedChannelMessage(channels, jid, rawText),
    editTrackedMessage: (jid, messageId, rawText) =>
--- a/src/ipc-outbound-delivery.test.ts
+++ b/src/ipc-outbound-delivery.test.ts
@@ -1,6 +1,9 @@
 import { describe, expect, it, vi } from 'vitest';

-import { deliverIpcOutboundMessage } from './ipc-outbound-delivery.js';
+import {
+  deliverCanonicalOutboundMessage,
+  deliverIpcOutboundMessage,
+} from './ipc-outbound-delivery.js';
 import type { WorkItem } from './db/work-items.js';
 import type { Channel, RegisteredGroup } from './types.js';

@@ -56,6 +59,47 @@ function makeWorkItem(input: {
 }

 describe('deliverIpcOutboundMessage', () => {
+  it('funnels ordinary outbound through canonical work item delivery', async () => {
+    const channel = makeChannel();
+    const createdItem = makeWorkItem({
+      chat_jid: 'dc:room',
+      group_folder: 'room-folder',
+      result_payload: '일반 안내',
+    });
+    const createWorkItem = vi.fn(() => createdItem);
+    const deliverWorkItem = vi.fn(async () => true);
+
+    const result = await deliverCanonicalOutboundMessage(
+      { jid: 'dc:room', text: '일반 안내' },
+      {
+        channels: [channel],
+        roomBindings: () => ({ 'dc:room': makeGroup() }),
+        createWorkItem,
+        deliverWorkItem,
+      },
+    );
+
+    expect(result).toBe('delivered');
+    expect(createWorkItem).toHaveBeenCalledWith({
+      group_folder: 'room-folder',
+      chat_jid: 'dc:room',
+      agent_type: 'codex',
+      delivery_role: null,
+      start_seq: null,
+      end_seq: null,
+      result_payload: '일반 안내',
+      attachments: undefined,
+    });
+    expect(deliverWorkItem).toHaveBeenCalledWith(
+      expect.objectContaining({
+        channel,
+        item: createdItem,
+        attachmentBaseDirs: ['/repo'],
+      }),
+    );
+    expect(channel.sendMessage).not.toHaveBeenCalled();
+  });
+
  it('funnels IPC outbound through canonical work item delivery', async () => {
    const ownerChannel = makeChannel();
    const reviewerChannel = makeChannel('discord-review', false);
@@ -155,7 +199,7 @@ describe('deliverIpcOutboundMessage', () => {
          deliverWorkItem,
        },
      ),
-    ).rejects.toThrow('No registered room binding for IPC outbound JID');
+    ).rejects.toThrow('No registered room binding for outbound JID');

    expect(createWorkItem).not.toHaveBeenCalled();
    expect(deliverWorkItem).not.toHaveBeenCalled();
--- a/src/ipc-outbound-delivery.ts
+++ b/src/ipc-outbound-delivery.ts
@@ -43,6 +43,23 @@ export interface IpcOutboundDeliveryDeps {
  deliverWorkItem?: typeof deliverOpenWorkItem;
 }

+export interface CanonicalOutboundDeliveryArgs {
+  jid: string;
+  text: string;
+  deliveryRole?: PairedRoomRole;
+  attachments?: OutboundAttachment[];
+}
+
+export interface CanonicalOutboundDeliveryDeps {
+  channels: Channel[];
+  roomBindings: () => Record<string, RegisteredGroup>;
+  log?: IpcDeliveryLog;
+  createWorkItem?: typeof createProducedWorkItem;
+  deliverWorkItem?: typeof deliverOpenWorkItem;
+}
+
+export type CanonicalOutboundDeliveryResult = 'delivered' | 'queued_retry';
+
 export type IpcOutboundDeliveryResult =
  | 'delivered'
  | 'queued_retry'
@@ -65,6 +82,61 @@ function isTerminalStatusMessage(text: string): boolean {
  return parseVisibleVerdict(text) !== 'continue';
 }

+export async function deliverCanonicalOutboundMessage(
+  args: CanonicalOutboundDeliveryArgs,
+  deps: CanonicalOutboundDeliveryDeps,
+): Promise<CanonicalOutboundDeliveryResult> {
+  const log = deps.log ?? logger;
+  const group = deps.roomBindings()[args.jid];
+  if (!group) {
+    throw new Error(`No registered room binding for outbound JID: ${args.jid}`);
+  }
+
+  const route = resolveChannelForDeliveryRole(
+    deps.channels,
+    args.jid,
+    args.deliveryRole,
+  );
+  if (!route.channel) throw new Error(`No channel for JID: ${args.jid}`);
+
+  log.info(
+    {
+      transition: 'outbound:route',
+      chatJid: args.jid,
+      deliveryRole: args.deliveryRole ?? null,
+      requestedRoleChannel: route.requestedRoleChannelName,
+      selectedChannel: route.selectedChannelName,
+      usedRoleChannel: route.usedRoleChannel,
+      fallbackUsed: route.fallbackUsed,
+    },
+    'Routed outbound message to canonical work item delivery',
+  );
+
+  const createWorkItem = deps.createWorkItem ?? createProducedWorkItem;
+  const workItem = createWorkItem({
+    group_folder: group.folder,
+    chat_jid: args.jid,
+    agent_type: group.agentType ?? 'claude-code',
+    delivery_role: args.deliveryRole ?? null,
+    start_seq: null,
+    end_seq: null,
+    result_payload: args.text,
+    attachments: args.attachments,
+  });
+
+  const deliverWorkItem = deps.deliverWorkItem ?? deliverOpenWorkItem;
+  const delivered = await deliverWorkItem({
+    channel: route.channel,
+    item: workItem as WorkItem,
+    log,
+    attachmentBaseDirs: group.workDir ? [group.workDir] : undefined,
+    isDuplicateOfLastBotFinal: () => false,
+    openContinuation: () => {},
+  });
+
+  return delivered ? 'delivered' : 'queued_retry';
+}
+
 export async function deliverIpcOutboundMessage(
  args: IpcOutboundDeliveryArgs,
  deps: IpcOutboundDeliveryDeps,
@@ -92,58 +164,34 @@ export async function deliverIpcOutboundMessage(
    return 'skipped_recorded_terminal';
  }

-  const group = deps.roomBindings()[args.jid];
-  if (!group) {
-    throw new Error(
-      `No registered room binding for IPC outbound JID: ${args.jid}`,
-    );
-  }
-
-  const route = resolveChannelForDeliveryRole(
-    deps.channels,
-    args.jid,
-    deliveryRole,
-  );
-  if (!route.channel) throw new Error(`No channel for JID: ${args.jid}`);
-
  log.info(
    {
      transition: 'ipc:route',
      chatJid: args.jid,
      runId: args.runId ?? null,
      senderRole: deliveryRole ?? null,
-      requestedRoleChannel: route.requestedRoleChannelName,
-      selectedChannel: route.selectedChannelName,
-      usedRoleChannel: route.usedRoleChannel,
-      fallbackUsed: route.fallbackUsed,
    },
    'IPC relay routed message to canonical work item delivery',
  );

-  const createWorkItem = deps.createWorkItem ?? createProducedWorkItem;
-  const workItem = createWorkItem({
-    group_folder: group.folder,
-    chat_jid: args.jid,
-    agent_type: group.agentType ?? 'claude-code',
-    delivery_role: deliveryRole ?? null,
-    start_seq: null,
-    end_seq: null,
-    result_payload: args.text,
-    attachments: args.attachments,
-  });
+  const result = await deliverCanonicalOutboundMessage(
+    {
+      jid: args.jid,
+      text: args.text,
+      deliveryRole,
+      attachments: args.attachments,
+    },
+    {
+      channels: deps.channels,
+      roomBindings: deps.roomBindings,
+      log,
+      createWorkItem: deps.createWorkItem,
+      deliverWorkItem: deps.deliverWorkItem,
+    },
+  );

-  const deliverWorkItem = deps.deliverWorkItem ?? deliverOpenWorkItem;
-  const delivered = await deliverWorkItem({
-    channel: route.channel,
-    item: workItem as WorkItem,
-    log,
-    attachmentBaseDirs: group.workDir ? [group.workDir] : undefined,
-    isDuplicateOfLastBotFinal: () => false,
-    openContinuation: () => {},
-  });
-
-  if (!delivered) {
-    return 'queued_retry';
+  if (result !== 'delivered') {
+    return result;
  }

  if (
--- a/src/message-runtime.ts
+++ b/src/message-runtime.ts
@@ -20,6 +20,7 @@ import {
  deliverOpenWorkItem,
  processOpenWorkItemDelivery,
 } from './message-runtime-delivery.js';
+import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js';
 import { handleQueuedRunGates } from './message-runtime-gating.js';
 import {
  enqueuePendingHandoffs as enqueueClaimedServiceHandoffs,
@@ -93,6 +94,21 @@ export interface MessageRuntimeDeps {
  clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void;
 }

+async function deliverSessionCommandMessage(
+  deps: Pick<MessageRuntimeDeps, 'channels' | 'getRoomBindings'>,
+  chatJid: string,
+  text: string,
+): Promise<void> {
+  await deliverCanonicalOutboundMessage(
+    { jid: chatJid, text },
+    {
+      channels: deps.channels,
+      roomBindings: deps.getRoomBindings,
+      log: logger,
+    },
+  );
+}
+
 export function createMessageRuntime(deps: MessageRuntimeDeps): {
  processGroupMessages: (
    chatJid: string,
@@ -110,7 +126,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
    chatJid: string,
    messages: NewMessage[],
  ): NewMessage[] => labelPairedSenders(deps.channels, chatJid, messages);
-
  const enqueueScopedGroupMessageCheck = buildScopedMessageCheckEnqueuer(
    deps.queue,
  );
@@ -587,7 +602,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
        timezone: deps.timezone,
        hasImplicitContinuationWindow: continuationTracker.has,
        sessionCommandDeps: {
-          sendMessage: (text) => channel.sendMessage(chatJid, text),
+          sendMessage: (text) =>
+            deliverSessionCommandMessage(deps, chatJid, text),
          setTyping: (typing) =>
            channel.setTyping?.(chatJid, typing) ?? Promise.resolve(),
          runAgent: (prompt, onOutput) =>