diff --git a/src/message-runtime-preflight-messages.test.ts b/src/message-runtime-preflight-messages.test.ts new file mode 100644 index 0000000..da6b586 --- /dev/null +++ b/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('./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 { + 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 { + 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); + }); +}); diff --git a/src/message-runtime-preflight-messages.ts b/src/message-runtime-preflight-messages.ts new file mode 100644 index 0000000..88f1a46 --- /dev/null +++ b/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; + 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; + }); +} diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 0b4de29..a7d29f5 100644 --- a/src/message-runtime.ts +++ b/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