Extract message runtime preflight helpers

src/message-runtime-preflight-messages.test.ts

@@ -0,0 +1,141 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+vi.mock('./db.js', async () => {
+  const actual = await vi.importActual<typeof import('./db.js')>('./db.js');
+  return {
+    ...actual,
+    getMessagesSinceSeq: vi.fn(),
+  };
+});
+
+import { _initTestDatabase, getMessagesSinceSeq } from './db.js';
+import {
+  getFreshHumanPreflightMessages,
+  hasHumanMessageAfterWorkItem,
+} from './message-runtime-preflight-messages.js';
+import type { WorkItem } from './db/work-items.js';
+import type { Channel, NewMessage } 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 makeMessage(overrides: Partial<NewMessage>): NewMessage {
+  return {
+    id: 'message-1',
+    chat_jid: 'chat-1',
+    sender: 'human@test',
+    sender_name: 'human',
+    content: 'hello',
+    timestamp: '2026-04-29T01:00:00.000Z',
+    seq: 1,
+    ...overrides,
+  };
+}
+
+function makeWorkItem(overrides: Partial<WorkItem>): WorkItem {
+  return {
+    id: 1,
+    group_folder: 'room',
+    chat_jid: 'chat-1',
+    agent_type: 'codex',
+    service_id: 'codex-main',
+    delivery_role: 'owner',
+    start_seq: null,
+    end_seq: null,
+    result_payload: 'result',
+    status: 'delivery_retry',
+    last_error: null,
+    delivery_attempts: 1,
+    delivery_message_id: null,
+    created_at: '2026-04-29T01:00:00.000Z',
+    updated_at: '2026-04-29T01:00:00.000Z',
+    delivered_at: null,
+    ...overrides,
+  };
+}
+
+describe('message-runtime-preflight-messages', () => {
+  beforeEach(() => {
+    _initTestDatabase();
+    vi.mocked(getMessagesSinceSeq).mockReset();
+  });
+
+  it('loads messages from the last agent cursor and keeps only fresh human messages', () => {
+    vi.mocked(getMessagesSinceSeq).mockReturnValue([
+      makeMessage({ id: 'human', content: 'new human', seq: 11 }),
+      makeMessage({
+        id: 'self',
+        content: 'self message',
+        seq: 12,
+        is_from_me: true,
+      }),
+      makeMessage({
+        id: 'bot',
+        content: 'bot message',
+        seq: 13,
+        is_bot_message: true,
+      }),
+    ]);
+
+    const messages = getFreshHumanPreflightMessages({
+      chatJid: 'chat-1',
+      channel: makeChannel(),
+      lastAgentTimestamps: { 'chat-1': '10' },
+      assistantName: 'Andy',
+      failureFinalText: 'failure',
+    });
+
+    expect(getMessagesSinceSeq).toHaveBeenCalledWith('chat-1', '10', 'Andy');
+    expect(messages.map((message) => message.id)).toEqual(['human']);
+  });
+
+  it('detects human messages after a work item by sequence', () => {
+    const workItem = makeWorkItem({ start_seq: 10, end_seq: 12 });
+
+    expect(
+      hasHumanMessageAfterWorkItem(workItem, [
+        makeMessage({ id: 'old', seq: 12 }),
+      ]),
+    ).toBe(false);
+    expect(
+      hasHumanMessageAfterWorkItem(workItem, [
+        makeMessage({ id: 'new', seq: 13 }),
+      ]),
+    ).toBe(true);
+  });
+
+  it('falls back to timestamps when sequence cursors are unavailable', () => {
+    const workItem = makeWorkItem({
+      start_seq: null,
+      end_seq: null,
+      updated_at: '2026-04-29T01:00:00.000Z',
+    });
+
+    expect(
+      hasHumanMessageAfterWorkItem(workItem, [
+        makeMessage({
+          id: 'old',
+          seq: undefined,
+          timestamp: '2026-04-29T00:59:59.000Z',
+        }),
+      ]),
+    ).toBe(false);
+    expect(
+      hasHumanMessageAfterWorkItem(workItem, [
+        makeMessage({
+          id: 'new',
+          seq: undefined,
+          timestamp: '2026-04-29T01:00:01.000Z',
+        }),
+      ]),
+    ).toBe(true);
+  });
+});

src/message-runtime-preflight-messages.ts

@@ -0,0 +1,58 @@
+import { getMessagesSinceSeq } from './db.js';
+import {
+  filterLoopingPairedBotMessages,
+  getProcessableMessages,
+} from './message-runtime-rules.js';
+import type { WorkItem } from './db/work-items.js';
+import type { Channel, NewMessage } from './types.js';
+
+export function getFreshHumanPreflightMessages(args: {
+  chatJid: string;
+  channel: Channel;
+  lastAgentTimestamps: Record<string, string>;
+  assistantName: string;
+  failureFinalText: string;
+}): NewMessage[] {
+  const sinceSeqCursor = args.lastAgentTimestamps[args.chatJid] || '0';
+  const preflightRawMessages = getMessagesSinceSeq(
+    args.chatJid,
+    sinceSeqCursor,
+    args.assistantName,
+  );
+  const preflightMessages = filterLoopingPairedBotMessages(
+    args.chatJid,
+    getProcessableMessages(args.chatJid, preflightRawMessages, args.channel),
+    args.failureFinalText,
+  );
+  return preflightMessages.filter(
+    (message) => message.is_from_me !== true && !message.is_bot_message,
+  );
+}
+
+export function hasHumanMessageAfterWorkItem(
+  openWorkItem: WorkItem,
+  freshHumanMessages: NewMessage[],
+): boolean {
+  if (freshHumanMessages.length === 0) {
+    return false;
+  }
+
+  const workItemSeq = openWorkItem.end_seq ?? openWorkItem.start_seq ?? null;
+  const workItemUpdatedAt = Date.parse(openWorkItem.updated_at);
+
+  return freshHumanMessages.some((message) => {
+    if (message.seq != null && workItemSeq != null) {
+      return message.seq > workItemSeq;
+    }
+
+    const messageTimestamp = Date.parse(message.timestamp);
+    if (
+      Number.isFinite(messageTimestamp) &&
+      Number.isFinite(workItemUpdatedAt)
+    ) {
+      return messageTimestamp > workItemUpdatedAt;
+    }
+
+    return true;
+  });
+}

src/message-runtime.ts

@@ -58,7 +58,6 @@ import {
 } from './types.js';
 import { createScopedLogger, logger } from './logger.js';
 import { hasReviewerLease } from './service-routing.js';
-import type { WorkItem } from './db/work-items.js';
 export {
   resolveHandoffCursorKey,
   resolveHandoffRoleOverride,
@@ -74,6 +73,10 @@ import {
   labelPairedSenders,
 } from './message-runtime-turns.js';
 import { resolvePairedRoleChannels } from './message-runtime-role-channels.js';
+import {
+  getFreshHumanPreflightMessages,
+  hasHumanMessageAfterWorkItem,
+} from './message-runtime-preflight-messages.js';
 
 export { isDuplicateOfLastBotFinal };
 
@@ -131,54 +134,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
     deps.queue,
   );
 
-  const getFreshHumanPreflightMessages = (
-    chatJid: string,
-    channel: Channel,
-  ): NewMessage[] => {
-    const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0';
-    const preflightRawMessages = getMessagesSinceSeq(
-      chatJid,
-      sinceSeqCursor,
-      deps.assistantName,
-    );
-    const preflightMessages = filterLoopingPairedBotMessages(
-      chatJid,
-      getProcessableMessages(chatJid, preflightRawMessages, channel),
-      FAILURE_FINAL_TEXT,
-    );
-    return preflightMessages.filter(
-      (message) => message.is_from_me !== true && !message.is_bot_message,
-    );
-  };
-
-  const hasHumanMessageAfterWorkItem = (
-    openWorkItem: WorkItem,
-    freshHumanMessages: NewMessage[],
-  ): boolean => {
-    if (freshHumanMessages.length === 0) {
-      return false;
-    }
-
-    const workItemSeq = openWorkItem.end_seq ?? openWorkItem.start_seq ?? null;
-    const workItemUpdatedAt = Date.parse(openWorkItem.updated_at);
-
-    return freshHumanMessages.some((message) => {
-      if (message.seq != null && workItemSeq != null) {
-        return message.seq > workItemSeq;
-      }
-
-      const messageTimestamp = Date.parse(message.timestamp);
-      if (
-        Number.isFinite(messageTimestamp) &&
-        Number.isFinite(workItemUpdatedAt)
-      ) {
-        return messageTimestamp > workItemUpdatedAt;
-      }
-
-      return true;
-    });
-  };
-
   const scheduleQueuedPairedFollowUp = (args: {
     chatJid: string;
     runId: string;
@@ -409,10 +364,13 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
       : null;
     let openWorkItem = getOpenWorkItemForChat(chatJid, SERVICE_SESSION_SCOPE);
     if (openWorkItem?.delivery_role === 'owner' && pendingTask) {
-      const freshHumanMessages = getFreshHumanPreflightMessages(
+      const freshHumanMessages = getFreshHumanPreflightMessages({
         chatJid,
         channel,
-      );
+        lastAgentTimestamps: deps.getLastAgentTimestamps(),
+        assistantName: deps.assistantName,
+        failureFinalText: FAILURE_FINAL_TEXT,
+      });
       if (
         pendingTask.status === 'merge_ready' &&
         freshHumanMessages.length > 0