diff --git a/cli/src/api/apiSession.test.ts b/cli/src/api/apiSession.test.ts index 5b4b5a44..deffd7c2 100644 --- a/cli/src/api/apiSession.test.ts +++ b/cli/src/api/apiSession.test.ts @@ -8,31 +8,40 @@ const socketHarness = vi.hoisted(() => ({ connectImmediately: boolean emitted: Array<{ event: string; args: unknown[] }> listeners: Map void>> + trigger: (event: string, ...args: any[]) => void triggerConnect: () => void triggerConnectError: () => void }> })) +const axiosHarness = vi.hoisted(() => ({ + get: vi.fn() +})) + vi.mock('socket.io-client', () => ({ io: () => { - const state = { + const state: (typeof socketHarness.sockets)[number] = { connected: false, connectCalls: 0, connectImmediately: true, emitted: [] as Array<{ event: string; args: unknown[] }>, listeners: new Map void>>(), + trigger: () => {}, triggerConnect: () => {}, triggerConnectError: () => {} } + state.trigger = (event: string, ...args: any[]) => { + for (const listener of state.listeners.get(event) ?? []) { + listener(...args) + } + } const triggerConnect = () => { state.connected = true - for (const listener of state.listeners.get('connect') ?? []) listener() + state.trigger('connect') } state.triggerConnect = triggerConnect state.triggerConnectError = () => { - for (const listener of state.listeners.get('connect_error') ?? []) { - listener(new Error('connect failed')) - } + state.trigger('connect_error', new Error('connect failed')) } const socket = { get connected() { @@ -73,6 +82,18 @@ vi.mock('socket.io-client', () => ({ } })) +vi.mock('axios', () => ({ + default: { + get: axiosHarness.get, + isAxiosError: (error: unknown) => ( + typeof error === 'object' + && error !== null + && 'isAxiosError' in error + && error.isAxiosError === true + ) + } +})) + import { ApiSessionClient, isExternalUserMessage, IncomingMessageFilter } from './apiSession' function createSession(overrides: Partial = {}): Session { @@ -111,6 +132,37 @@ function deferred() { return { promise, resolve, reject } } +function triggerIncomingUserMessage( + socket: (typeof socketHarness.sockets)[number], + message: { + id?: string + seq: number + text: string + sentFrom: 'cli' | 'webapp' | 'telegram-bot' + } +): void { + socket.trigger('update', { + body: { + t: 'new-message', + message: { + id: message.id, + seq: message.seq, + localId: null, + content: { + role: 'user', + content: { + type: 'text', + text: message.text + }, + meta: { + sentFrom: message.sentFrom + } + } + } + } + }) +} + describe('ApiSessionClient lazy materialization', () => { it('does not connect or materialize without a real user message', async () => { socketHarness.sockets.length = 0 @@ -322,6 +374,135 @@ describe('ApiSessionClient lazy materialization', () => { }) }) +describe('ApiSessionClient incoming user messages', () => { + it('ignores CLI-originated transcript messages while advancing the incoming cursor', () => { + socketHarness.sockets.length = 0 + const client = new ApiSessionClient('token', createSession({ namespace: 'default' })) + const socket = socketHarness.sockets[0] + if (!socket) throw new Error('expected socket') + const onUserMessage = vi.fn() + client.onUserMessage(onUserMessage) + + triggerIncomingUserMessage(socket, { + id: 'historical-cli-message', + seq: 10, + text: 'historical prompt from the local transcript', + sentFrom: 'cli' + }) + triggerIncomingUserMessage(socket, { + seq: 10, + text: 'legacy duplicate at the filtered cursor', + sentFrom: 'webapp' + }) + triggerIncomingUserMessage(socket, { + id: 'live-web-message', + seq: 11, + text: 'new prompt from the phone', + sentFrom: 'webapp' + }) + + expect(onUserMessage).toHaveBeenCalledTimes(1) + expect(onUserMessage).toHaveBeenCalledWith( + expect.objectContaining({ + content: expect.objectContaining({ text: 'new prompt from the phone' }) + }), + undefined + ) + client.close() + }) + + it('delivers only remote prompts from a mixed reconnect backfill', async () => { + socketHarness.sockets.length = 0 + axiosHarness.get.mockReset() + axiosHarness.get.mockResolvedValue({ + data: { + messages: [ + { + id: 'backfilled-cli-message', + seq: 2, + createdAt: 2, + localId: null, + content: { + role: 'user', + content: { type: 'text', text: 'historical local prompt' }, + meta: { sentFrom: 'cli' } + } + }, + { + id: 'backfilled-web-message', + seq: 3, + createdAt: 3, + localId: null, + content: { + role: 'user', + content: { type: 'text', text: 'remote prompt after reconnect' }, + meta: { sentFrom: 'webapp' } + } + } + ] + } + }) + const client = new ApiSessionClient('token', createSession({ namespace: 'default' })) + const socket = socketHarness.sockets[0] + if (!socket) throw new Error('expected socket') + const receivedTexts: string[] = [] + client.onUserMessage((message) => { + receivedTexts.push(message.content.text) + }) + triggerIncomingUserMessage(socket, { + id: 'initial-web-message', + seq: 1, + text: 'initial remote prompt', + sentFrom: 'webapp' + }) + + socket.connected = false + socket.trigger('disconnect', 'transport close') + socket.triggerConnect() + + await vi.waitFor(() => expect(axiosHarness.get).toHaveBeenCalledOnce()) + await vi.waitFor(() => expect(receivedTexts).toEqual([ + 'initial remote prompt', + 'remote prompt after reconnect' + ])) + expect(axiosHarness.get).toHaveBeenCalledWith( + expect.stringContaining('/cli/sessions/'), + expect.objectContaining({ + params: { afterSeq: 1, limit: 200 } + }) + ) + client.close() + }) + + it.each(['webapp', 'telegram-bot'] as const)( + 'delivers %s-originated user messages', + (sentFrom) => { + socketHarness.sockets.length = 0 + const client = new ApiSessionClient('token', createSession({ namespace: 'default' })) + const socket = socketHarness.sockets[0] + if (!socket) throw new Error('expected socket') + const onUserMessage = vi.fn() + client.onUserMessage(onUserMessage) + + triggerIncomingUserMessage(socket, { + id: `${sentFrom}-message`, + seq: 1, + text: `prompt from ${sentFrom}`, + sentFrom + }) + + expect(onUserMessage).toHaveBeenCalledOnce() + expect(onUserMessage).toHaveBeenCalledWith( + expect.objectContaining({ + meta: { sentFrom } + }), + undefined + ) + client.close() + } + ) +}) + describe('isExternalUserMessage', () => { const baseUserMsg = { type: 'user' as const, diff --git a/cli/src/api/apiSession.ts b/cli/src/api/apiSession.ts index bf79df8d..46ca8a44 100644 --- a/cli/src/api/apiSession.ts +++ b/cli/src/api/apiSession.ts @@ -627,6 +627,12 @@ export class ApiSessionClient extends EventEmitter { const userResult = UserMessageSchema.safeParse(message.content) if (userResult.success) { + // User messages mirrored from a local agent transcript are history, + // not new remote input. Keep them in the incoming filter above so + // reconnect backfill still advances and deduplicates correctly. + if (userResult.data.meta?.sentFrom === 'cli') { + return + } this.enqueueUserMessage(userResult.data, message.localId ?? undefined) return }