From 1974f8a6dc072008a4cc56eabcf16fd9985ba2ff Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Mon, 23 Mar 2026 04:13:53 +0900 Subject: [PATCH] refactor: remove active follow-up piping --- src/group-queue.test.ts | 119 +----------------- src/group-queue.ts | 114 +---------------- src/message-runtime.test.ts | 236 +----------------------------------- src/message-runtime.ts | 157 +++--------------------- 4 files changed, 23 insertions(+), 603 deletions(-) diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 5911ab2..6fffbd0 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -8,7 +8,7 @@ vi.mock('./config.js', () => ({ MAX_CONCURRENT_AGENTS: 2, })); -// Mock fs operations used by sendMessage/closeStdin +// Mock fs operations used by closeStdin vi.mock('fs', async () => { const actual = await vi.importActual('fs'); return { @@ -409,7 +409,7 @@ describe('GroupQueue', () => { await vi.advanceTimersByTimeAsync(10); }); - it('keeps a post-final follow-up queued instead of preempting the idle run', async () => { + it('preempts an idle run so the next message is handled in a fresh run', async () => { const fs = await import('fs'); let resolveProcess: () => void; @@ -425,7 +425,6 @@ describe('GroupQueue', () => { await vi.advanceTimersByTimeAsync(10); queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group'); - queue.setFollowUpPipeAllowed('group1@g.us', () => false); queue.notifyIdle('group1@g.us'); const writeFileSync = vi.mocked(fs.default.writeFileSync); @@ -436,119 +435,7 @@ describe('GroupQueue', () => { const closeWrites = writeFileSync.mock.calls.filter( (call) => typeof call[0] === 'string' && call[0].endsWith('_close'), ); - expect(closeWrites).toHaveLength(0); - - resolveProcess!(); - await vi.advanceTimersByTimeAsync(10); - - expect(processMessages).toHaveBeenCalledTimes(2); - }); - - it('sendMessage resets idleWaiting so a subsequent task enqueue does not preempt', async () => { - const fs = await import('fs'); - let resolveProcess: () => void; - - const processMessages = vi.fn(async () => { - await new Promise((resolve) => { - resolveProcess = resolve; - }); - return true; - }); - - queue.setProcessMessagesFn(processMessages); - queue.enqueueMessageCheck('group1@g.us'); - await vi.advanceTimersByTimeAsync(10); - queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group'); - - // Agent becomes idle - queue.notifyIdle('group1@g.us'); - - // A new user message arrives — resets idleWaiting - queue.sendMessage('group1@g.us', 'hello'); - - // Task enqueued after message reset — should NOT preempt (agent is working) - const writeFileSync = vi.mocked(fs.default.writeFileSync); - writeFileSync.mockClear(); - - const taskFn = vi.fn(async () => {}); - queue.enqueueTask('group1@g.us', 'task-1', taskFn); - - const closeWrites = writeFileSync.mock.calls.filter( - (call) => typeof call[0] === 'string' && call[0].endsWith('_close'), - ); - expect(closeWrites).toHaveLength(0); - - resolveProcess!(); - await vi.advanceTimersByTimeAsync(10); - }); - - it('sendMessage touches active run activity after piping follow-up', async () => { - let resolveProcess: () => void; - const touch = vi.fn(); - - const processMessages = vi.fn(async () => { - await new Promise((resolve) => { - resolveProcess = resolve; - }); - return true; - }); - - queue.setProcessMessagesFn(processMessages); - queue.enqueueMessageCheck('group1@g.us'); - await vi.advanceTimersByTimeAsync(10); - queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group'); - queue.setActivityTouch('group1@g.us', touch); - - expect(queue.sendMessage('group1@g.us', 'hello')).toBe(true); - expect(touch).toHaveBeenCalledTimes(1); - - resolveProcess!(); - await vi.advanceTimersByTimeAsync(10); - }); - - it('sendMessage returns false for task agents so user messages queue up', async () => { - let resolveTask: () => void; - - const taskFn = vi.fn(async () => { - await new Promise((resolve) => { - resolveTask = resolve; - }); - }); - - // Start a task (sets isTaskProcess = true) - queue.enqueueTask('group1@g.us', 'task-1', taskFn); - await vi.advanceTimersByTimeAsync(10); - queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group'); - - // sendMessage should return false — user messages must not go to task agents - const result = queue.sendMessage('group1@g.us', 'hello'); - expect(result).toBe(false); - - resolveTask!(); - await vi.advanceTimersByTimeAsync(10); - }); - - it('does not pipe follow-up messages to an agent after closeStdin', async () => { - let resolveProcess: () => void; - - const processMessages = vi.fn(async () => { - await new Promise((resolve) => { - resolveProcess = resolve; - }); - return true; - }); - - queue.setProcessMessagesFn(processMessages); - queue.enqueueMessageCheck('group1@g.us', 'test-group'); - await vi.advanceTimersByTimeAsync(10); - - queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group'); - queue.notifyIdle('group1@g.us'); - queue.closeStdin('group1@g.us'); - - expect(queue.sendMessage('group1@g.us', 'hello after clear')).toBe(false); - - queue.enqueueMessageCheck('group1@g.us', 'test-group'); + expect(closeWrites).toHaveLength(1); resolveProcess!(); await vi.advanceTimersByTimeAsync(10); diff --git a/src/group-queue.ts b/src/group-queue.ts index c8f0fe2..d82c9e7 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -11,12 +11,6 @@ interface QueuedTask { fn: () => Promise; } -export interface GroupActivityTouchMeta { - source: 'follow-up'; - textLength: number; - filename: string; -} - export interface GroupRunContext { runId: string; reason: 'messages' | 'drain'; @@ -41,8 +35,6 @@ interface GroupState { retryTimer: ReturnType | null; retryScheduledAt: number | null; startedAt: number | null; - activityTouch: ((meta?: GroupActivityTouchMeta) => void) | null; - followUpPipeAllowed: (() => boolean) | null; } export interface GroupStatus { @@ -81,8 +73,6 @@ export class GroupQueue { retryTimer: null, retryScheduledAt: null, startedAt: null, - activityTouch: null, - followUpPipeAllowed: null, }; this.groups.set(groupJid, state); } @@ -95,32 +85,8 @@ export class GroupQueue { this.processMessagesFn = fn; } - setActivityTouch( - groupJid: string, - touch: ((meta?: GroupActivityTouchMeta) => void) | null, - ): void { - const state = this.getGroup(groupJid); - state.activityTouch = touch; - } - - setFollowUpPipeAllowed( - groupJid: string, - allowed: (() => boolean) | null, - ): void { - const state = this.getGroup(groupJid); - state.followUpPipeAllowed = allowed; - } - private canPreemptIdleActiveRun(state: GroupState): boolean { - if (!state.idleWaiting) { - return false; - } - - if (!state.followUpPipeAllowed) { - return true; - } - - return state.followUpPipeAllowed(); + return state.idleWaiting; } private createRunId(): string { @@ -269,80 +235,6 @@ export class GroupQueue { } } - /** - * Send a follow-up message to the active agent process via IPC file. - * Returns true if the message was written, false if no active agent process. - */ - sendMessage(groupJid: string, text: string): boolean { - const state = this.getGroup(groupJid); - if (!state.active || !state.groupFolder || state.isTaskProcess) { - logger.debug( - { - groupJid, - runId: state.currentRunId, - active: state.active, - closingStdin: state.closingStdin, - groupFolder: state.groupFolder, - isTaskProcess: state.isTaskProcess, - }, - 'Cannot pipe follow-up message to active agent', - ); - return false; - } - if (state.closingStdin) { - logger.info( - { groupJid, runId: state.currentRunId, groupFolder: state.groupFolder }, - 'Skipping follow-up IPC because active agent is closing', - ); - return false; - } - if (state.followUpPipeAllowed && !state.followUpPipeAllowed()) { - logger.info( - { groupJid, runId: state.currentRunId, groupFolder: state.groupFolder }, - 'Skipping follow-up IPC because active agent requires a fresh logical run', - ); - return false; - } - state.idleWaiting = false; // Agent is about to receive work, no longer idle - - const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input'); - try { - fs.mkdirSync(inputDir, { recursive: true }); - const filename = `${Date.now()}-${Math.random().toString(36).slice(2, 6)}.json`; - const filepath = path.join(inputDir, filename); - const tempPath = `${filepath}.tmp`; - fs.writeFileSync(tempPath, JSON.stringify({ type: 'message', text })); - fs.renameSync(tempPath, filepath); - state.activityTouch?.({ - source: 'follow-up', - textLength: text.length, - filename, - }); - logger.info( - { - groupJid, - runId: state.currentRunId, - groupFolder: state.groupFolder, - textLength: text.length, - filename, - }, - 'Queued follow-up message for active agent', - ); - return true; - } catch (err) { - logger.warn( - { - groupJid, - runId: state.currentRunId, - groupFolder: state.groupFolder, - err, - }, - 'Failed to queue follow-up message for active agent', - ); - return false; - } - } - /** * Signal the active agent process to wind down by writing a close sentinel. */ @@ -446,8 +338,6 @@ export class GroupQueue { state.processName = null; state.groupFolder = null; state.currentRunId = null; - state.activityTouch = null; - state.followUpPipeAllowed = null; this.activeCount--; this.drainGroup(groupJid); } @@ -482,8 +372,6 @@ export class GroupQueue { state.process = null; state.processName = null; state.groupFolder = null; - state.activityTouch = null; - state.followUpPipeAllowed = null; this.activeCount--; this.drainGroup(groupJid); } diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 69fd571..cfa7378 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -335,7 +335,7 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(clearSession).not.toHaveBeenCalled(); - expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-2'); + expect(notifyIdle).not.toHaveBeenCalled(); expect(closeStdin).toHaveBeenCalledWith(chatJid, { runId: 'run-2', reason: 'output-delivered-close', @@ -432,8 +432,7 @@ describe('createMessageRuntime', () => { chatJid, 'CI 상태 확인 중입니다.', ); - expect(notifyIdle).toHaveBeenCalledTimes(1); - expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-progress'); + expect(notifyIdle).not.toHaveBeenCalled(); expect(persistSession).toHaveBeenCalledWith( group.folder, 'session-progress', @@ -889,8 +888,7 @@ describe('createMessageRuntime', () => { chatJid, '테스트가 끝났습니다.', ); - expect(notifyIdle).toHaveBeenCalledTimes(1); - expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-final'); + expect(notifyIdle).not.toHaveBeenCalled(); } finally { vi.useRealTimers(); } @@ -956,129 +954,6 @@ describe('createMessageRuntime', () => { expect(channel.sendAndTrack).not.toHaveBeenCalled(); }); - it('does not emit a synthetic failure final when follow-up activity happens during a silent run', async () => { - const chatJid = 'group@test'; - const group = makeGroup('codex'); - const channel = makeChannel(chatJid); - const lastAgentTimestamps: Record = {}; - const saveState = vi.fn(); - let activityTouch: - | ((meta?: { - source: 'follow-up'; - textLength: number; - filename: string; - }) => void) - | null = null; - let followUpPersisted = false; - - vi.mocked(db.getMessagesSince).mockImplementation( - (_chatJid, sinceSeqCursor) => { - if (String(sinceSeqCursor || '0') === '0') { - return followUpPersisted - ? [ - { - id: 'msg-1', - chat_jid: chatJid, - sender: 'user@test', - sender_name: 'User', - content: 'hello', - timestamp: '2026-03-19T00:00:00.000Z', - seq: 1, - }, - { - id: 'msg-2', - chat_jid: chatJid, - sender: 'codex@test', - sender_name: 'Codex', - content: 'follow-up', - timestamp: '2026-03-19T00:00:05.000Z', - seq: 2, - }, - ] - : [ - { - id: 'msg-1', - chat_jid: chatJid, - sender: 'user@test', - sender_name: 'User', - content: 'hello', - timestamp: '2026-03-19T00:00:00.000Z', - seq: 1, - }, - ]; - } - - if (String(sinceSeqCursor || '0') === '1' && followUpPersisted) { - return [ - { - id: 'msg-2', - chat_jid: chatJid, - sender: 'codex@test', - sender_name: 'Codex', - content: 'follow-up', - timestamp: '2026-03-19T00:00:05.000Z', - seq: 2, - }, - ]; - } - - return []; - }, - ); - - vi.mocked(agentRunner.runAgentProcess).mockImplementationOnce(async () => { - lastAgentTimestamps[chatJid] = '2'; - followUpPersisted = true; - activityTouch?.({ - source: 'follow-up', - textLength: 9, - filename: 'follow-up.json', - }); - return { - status: 'success', - result: null, - newSessionId: 'session-silent-follow-up', - }; - }); - - 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(), - setActivityTouch: vi.fn( - (_jid, touch) => (activityTouch = touch as typeof activityTouch), - ), - setFollowUpPipeAllowed: vi.fn(), - } as any, - getRegisteredGroups: () => ({ [chatJid]: group }), - getSessions: () => ({}), - getLastTimestamp: () => '', - setLastTimestamp: vi.fn(), - getLastAgentTimestamps: () => lastAgentTimestamps, - saveState, - persistSession: vi.fn(), - clearSession: vi.fn(), - }); - - const result = await runtime.processGroupMessages(chatJid, { - runId: 'run-silent-follow-up', - reason: 'messages', - }); - - expect(result).toBe(true); - expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(1); - expect(saveState).toHaveBeenCalled(); - expect(lastAgentTimestamps[chatJid]).toBe('2'); - expect(channel.sendMessage).not.toHaveBeenCalled(); - }); - it('resets tracked progress after a final output that becomes empty after formatting', async () => { vi.useFakeTimers(); const chatJid = 'group@test'; @@ -1301,111 +1176,6 @@ describe('createMessageRuntime', () => { } }); - it('ignores follow-up activity after a final was already observed in the same run', async () => { - vi.useFakeTimers(); - const chatJid = 'group@test'; - const group = makeGroup('codex'); - const channel = makeChannel(chatJid); - const closeStdin = vi.fn(); - const enqueueMessageCheck = vi.fn(); - const lastAgentTimestamps: Record = { [chatJid]: '1' }; - let activityTouch: - | ((meta?: { - source: 'follow-up'; - textLength: number; - filename: string; - }) => void) - | null = null; - - 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', - seq: 2, - }, - ]); - - vi.mocked(agentRunner.runAgentProcess).mockImplementation( - async (_group, _input, _onProcess, onOutput) => { - await onOutput?.({ - status: 'success', - phase: 'final', - result: '첫 번째 턴 최종 답변', - newSessionId: 'session-final-boundary', - }); - activityTouch?.({ - source: 'follow-up', - textLength: 32, - filename: 'late-follow-up.json', - }); - lastAgentTimestamps[chatJid] = '3'; - await onOutput?.({ - status: 'success', - result: null, - newSessionId: 'session-final-boundary', - }); - return { - status: 'success', - result: null, - newSessionId: 'session-final-boundary', - }; - }, - ); - - const runtime = createMessageRuntime({ - assistantName: 'Andy', - idleTimeout: 1_000, - pollInterval: 1_000, - timezone: 'UTC', - triggerPattern: /^@Andy\b/i, - channels: [channel], - queue: { - registerProcess: vi.fn(), - closeStdin, - notifyIdle: vi.fn(), - enqueueMessageCheck, - setActivityTouch: vi.fn( - (_jid, touch) => (activityTouch = touch as typeof activityTouch), - ), - } as any, - getRegisteredGroups: () => ({ [chatJid]: group }), - getSessions: () => ({}), - getLastTimestamp: () => '', - setLastTimestamp: vi.fn(), - getLastAgentTimestamps: () => lastAgentTimestamps, - saveState: vi.fn(), - persistSession: vi.fn(), - clearSession: vi.fn(), - }); - - try { - const result = await runtime.processGroupMessages(chatJid, { - runId: 'run-final-boundary', - reason: 'messages', - }); - - expect(result).toBe(true); - expect(closeStdin).toHaveBeenCalledWith(chatJid, { - runId: 'run-final-boundary', - reason: 'output-delivered-close', - }); - expect(channel.sendMessage).toHaveBeenCalledWith( - chatJid, - '첫 번째 턴 최종 답변', - ); - expect(channel.sendMessage).toHaveBeenCalledTimes(1); - expect(channel.sendAndTrack).not.toHaveBeenCalled(); - expect(enqueueMessageCheck).not.toHaveBeenCalled(); - expect(lastAgentTimestamps[chatJid]).toBe('3'); - } finally { - vi.useRealTimers(); - } - }); - it('retries editing progress message instead of creating a duplicate when edit fails', async () => { vi.useFakeTimers(); const chatJid = 'group@test'; diff --git a/src/message-runtime.ts b/src/message-runtime.ts index bf72acd..54d7f72 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -664,11 +664,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ); type VisiblePhase = 'silent' | 'progress' | 'final'; - type PendingFollowUp = { - queuedAt: number; - textLength: number; - filename: string; - } | null; let idleTimer: ReturnType | null = null; let hadError = false; let producedDeliverySucceeded = true; @@ -681,8 +676,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { let progressEditFailCount = 0; let latestProgressTextForFinal: string | null = null; let poisonedSessionDetected = false; - let canPipeFollowUps = true; - let pendingFollowUp: PendingFollowUp = null; + let closeRequested = false; const hasVisibleOutput = () => visiblePhase !== 'silent'; const terminalObserved = () => visiblePhase === 'final'; @@ -703,10 +697,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { progressEditFailCount = 0; }; - const clearPendingFollowUp = () => { - pendingFollowUp = null; - }; - const renderProgressMessage = (text: string) => { const elapsedSeconds = progressStartedAt === null @@ -830,6 +820,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { await deliverFinalText(FAILURE_FINAL_TEXT); }; + const requestAgentClose = (reason: string) => { + if (closeRequested) return; + closeRequested = true; + deps.queue.closeStdin(chatJid, { runId, reason }); + }; + const sendProgressMessage = async (text: string) => { if (!text || (text === latestProgressText && progressMessageId)) { return; @@ -914,84 +910,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { idleTimer = null; return; } - idleTimer = setTimeout( - () => { - const closeReason = - !hasVisibleOutput() && pendingFollowUp - ? 'follow-up-no-output-preemption' - : 'idle-timeout'; - if (!hasVisibleOutput() && pendingFollowUp) { - logger.warn( - { - group: group.name, - chatJid, - runId, - followUpQueuedAt: new Date( - pendingFollowUp.queuedAt, - ).toISOString(), - followUpQueuedTextLength: pendingFollowUp.textLength, - followUpQueuedFilename: pendingFollowUp.filename, - followUpWaitMs: Date.now() - pendingFollowUp.queuedAt, - }, - 'Idle timeout reached while a queued follow-up still had no visible output', - ); - } else { - logger.debug( - { group: group.name, chatJid, runId, closeReason }, - 'Idle timeout, closing agent stdin', - ); - } - deps.queue.closeStdin(chatJid, { runId, reason: closeReason }); - }, - deps.idleTimeout, - ); + idleTimer = setTimeout(() => { + logger.debug( + { group: group.name, chatJid, runId }, + 'Idle timeout, closing agent stdin', + ); + requestAgentClose('idle-timeout'); + }, deps.idleTimeout); }; - const noteFollowUpQueued = (meta?: { - source: 'follow-up'; - textLength: number; - filename: string; - }) => { - if (meta?.source !== 'follow-up' || terminalObserved()) return; - pendingFollowUp = { - queuedAt: Date.now(), - textLength: meta.textLength, - filename: meta.filename, - }; - logger.info( - { - group: group.name, - chatJid, - runId, - followUpQueuedTextLength: pendingFollowUp.textLength, - followUpQueuedFilename: pendingFollowUp.filename, - }, - 'Registered follow-up activity for active agent', - ); - resetIdleTimer(); - }; - - const noteVisibleOutputObserved = (phase?: string) => { - if (!pendingFollowUp) return; - logger.info( - { - group: group.name, - chatJid, - runId, - followUpQueuedAt: new Date(pendingFollowUp.queuedAt).toISOString(), - followUpQueuedTextLength: pendingFollowUp.textLength, - followUpQueuedFilename: pendingFollowUp.filename, - followUpWaitMs: Date.now() - pendingFollowUp.queuedAt, - resultPhase: phase, - }, - 'Agent produced visible output after a queued follow-up', - ); - clearPendingFollowUp(); - }; - - deps.queue.setActivityTouch?.(chatJid, noteFollowUpQueued); - deps.queue.setFollowUpPipeAllowed?.(chatJid, () => canPipeFollowUps); - resetIdleTimer(); await channel.setTyping?.(chatJid, true); @@ -1059,8 +986,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (result.phase === 'progress') { if (text) { - canPipeFollowUps = false; - noteVisibleOutputObserved(result.phase); await sendProgressMessage(text); } if (!poisonedSessionDetected) { @@ -1073,8 +998,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } if (text) { - canPipeFollowUps = false; - noteVisibleOutputObserved(result.phase); await finalizeProgressMessage(); await deliverFinalText(text); } else if (raw) { @@ -1097,18 +1020,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } await channel.setTyping?.(chatJid, false); - if (!poisonedSessionDetected) { - resetIdleTimer(); - } - if (result.status === 'success' && !poisonedSessionDetected) { - deps.queue.notifyIdle(chatJid, runId); - if (hasVisibleOutput()) { - deps.queue.closeStdin(chatJid, { - runId, - reason: 'output-delivered-close', - }); - } + requestAgentClose('output-delivered-close'); } if (result.status === 'error') { @@ -1152,9 +1065,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { clearProgressTicker(); if (idleTimer) clearTimeout(idleTimer); - deps.queue.setActivityTouch?.(chatJid, null); - deps.queue.setFollowUpPipeAllowed?.(chatJid, null); - clearPendingFollowUp(); if (!producedDeliverySucceeded) { logger.warn( @@ -1277,42 +1187,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (!hasTrigger) continue; } - const allPending = getMessagesSinceSeq( - chatJid, - deps.getLastAgentTimestamps()[chatJid] || '', - deps.assistantName, - ); - const processablePending = getProcessableMessages( - chatJid, - allPending, - channel, - ); - const messagesToSend = - processablePending.length > 0 - ? processablePending - : processableGroupMessages; - const formatted = formatMessages(messagesToSend, deps.timezone); - - if (deps.queue.sendMessage(chatJid, formatted)) { - logger.debug( - { chatJid, count: messagesToSend.length }, - 'Piped messages to active agent', - ); - const endSeq = messagesToSend[messagesToSend.length - 1].seq; - if (endSeq != null) { - advanceLastAgentCursor(chatJid, endSeq); - } - channel - .setTyping?.(chatJid, true) - ?.catch((err) => - logger.warn( - { chatJid, err }, - 'Failed to set typing indicator', - ), - ); - } else { - deps.queue.enqueueMessageCheck(chatJid, group.folder); - } + deps.queue.enqueueMessageCheck(chatJid, group.folder); } } } catch (err) {