fix(conversation): make realtime sessions cancellable and recoverable

apps/desktop/tests/unit/realtime-session-redteam-r1-30.test.ts

@@ -0,0 +1,529 @@
+// tests/unit/realtime-session-redteam-r1-30.test.ts
+// Realtime 대화 세션: 시작 중 취소 누수, 끊김 후 재연결, open 전 첫 메시지 유실 회귀 테스트.
+// node 환경에서 fake 전송 계층/WebRTC 의존성으로 검증한다.
+
+import { describe, it, expect, vi } from 'vitest'
+import {
+  INITIAL_REALTIME_MODEL,
+  reduceRealtimeServerEvent,
+  type RealtimeModelContext,
+  type RealtimeServerEvent,
+} from '../../src/renderer/services/realtimeConversationModel'
+import {
+  RealtimeAbortError,
+  WebRtcRealtimeTransport,
+  type RealtimeClientEvent,
+  type RealtimeTransport,
+  type WebRtcTransportDeps,
+} from '../../src/renderer/services/realtimeSession'
+import {
+  REALTIME_SESSION_UPDATE,
+  RealtimeSessionController,
+} from '../../src/renderer/services/realtimeSessionController'
+
+// ── helpers ──
+
+interface Deferred<T> {
+  promise: Promise<T>
+  resolve: (value: T) => void
+  reject: (err: Error) => void
+}
+
+function deferred<T>(): Deferred<T> {
+  let resolve!: (value: T) => void
+  let reject!: (err: Error) => void
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res
+    reject = rej
+  })
+  return { promise, resolve, reject }
+}
+
+const flush = async (): Promise<void> => {
+  for (let i = 0; i < 10; i++) await Promise.resolve()
+}
+
+function makeCtx(): RealtimeModelContext {
+  let seq = 0
+  return {
+    nextId: (prefix) => `${prefix}-${++seq}`,
+    now: () => 1000,
+  }
+}
+
+// ── 순수 reducer ──
+
+describe('reduceRealtimeServerEvent', () => {
+  it('maps a full assistant turn to thinking → speaking → message + listening', () => {
+    const ctx = makeCtx()
+    let m = reduceRealtimeServerEvent(INITIAL_REALTIME_MODEL, { type: 'response.created' }, ctx)
+    expect(m.conversationState).toBe('thinking')
+    expect(m.assistantMsgId).toBe('assistant-1')
+
+    m = reduceRealtimeServerEvent(m, { type: 'response.output_audio_transcript.delta', delta: '안녕' }, ctx)
+    m = reduceRealtimeServerEvent(m, { type: 'response.audio_transcript.delta', delta: '하세요' }, ctx)
+    expect(m.conversationState).toBe('speaking')
+    expect(m.streamingText).toBe('안녕하세요')
+    expect(m.streamingMsgId).toBe('assistant-1')
+
+    m = reduceRealtimeServerEvent(m, { type: 'response.done' }, ctx)
+    expect(m.conversationState).toBe('listening')
+    expect(m.streamingText).toBe('')
+    expect(m.streamingMsgId).toBeNull()
+    expect(m.messages).toEqual([
+      { id: 'assistant-1', role: 'assistant', content: '안녕하세요', timestamp: 1000 },
+    ])
+  })
+
+  it('adds trimmed user transcripts and ignores empty ones', () => {
+    const ctx = makeCtx()
+    const ev = (transcript: string): RealtimeServerEvent => ({
+      type: 'conversation.item.input_audio_transcription.completed',
+      transcript,
+    })
+    let m = reduceRealtimeServerEvent(INITIAL_REALTIME_MODEL, ev('  hi  '), ctx)
+    expect(m.messages).toEqual([{ id: 'user-1', role: 'user', content: 'hi', timestamp: 1000 }])
+    const same = reduceRealtimeServerEvent(m, ev('   '), ctx)
+    expect(same).toBe(m)
+    m = reduceRealtimeServerEvent(m, { type: 'error', error: { message: 'boom' } }, ctx)
+    expect(m.error).toBe('boom')
+    expect(reduceRealtimeServerEvent(m, { type: 'error' }, ctx).error).toBe('Realtime error')
+  })
+
+  it('returns the same reference for unknown events and empty deltas', () => {
+    const ctx = makeCtx()
+    expect(reduceRealtimeServerEvent(INITIAL_REALTIME_MODEL, { type: 'session.created' }, ctx)).toBe(
+      INITIAL_REALTIME_MODEL,
+    )
+    expect(
+      reduceRealtimeServerEvent(INITIAL_REALTIME_MODEL, { type: 'response.text.delta', delta: '' }, ctx),
+    ).toBe(INITIAL_REALTIME_MODEL)
+  })
+})
+
+// ── WebRTC transport (fake 의존성) ──
+
+class FakeTrack {
+  stop = vi.fn()
+}
+
+class FakeStream {
+  readonly track = new FakeTrack()
+  getTracks = (): FakeTrack[] => [this.track]
+  getAudioTracks = (): FakeTrack[] => [this.track]
+}
+
+class FakeDataChannel {
+  readyState = 'connecting'
+  onopen: (() => void) | null = null
+  onclose: (() => void) | null = null
+  onmessage: ((e: { data: unknown }) => void) | null = null
+  send = vi.fn()
+  close = vi.fn(() => {
+    this.readyState = 'closed'
+  })
+  open(): void {
+    this.readyState = 'open'
+    this.onopen?.()
+  }
+  receive(data: unknown): void {
+    this.onmessage?.({ data })
+  }
+}
+
+class FakePeerConnection {
+  connectionState = 'new'
+  ontrack: ((e: { streams: unknown[] }) => void) | null = null
+  onconnectionstatechange: (() => void) | null = null
+  readonly dc = new FakeDataChannel()
+  readonly remote = deferred<void>()
+  addTrack = vi.fn()
+  createDataChannel = vi.fn(() => this.dc)
+  createOffer = vi.fn(async () => ({ type: 'offer', sdp: 'offer-sdp' }))
+  setLocalDescription = vi.fn(async () => undefined)
+  setRemoteDescription = vi.fn(() => this.remote.promise)
+  close = vi.fn()
+  setState(state: string): void {
+    this.connectionState = state
+    this.onconnectionstatechange?.()
+  }
+}
+
+interface TransportHarness {
+  transport: WebRtcRealtimeTransport
+  token: Deferred<{ value: string; model: string }>
+  media: Deferred<FakeStream>
+  stream: FakeStream
+  pcs: FakePeerConnection[]
+  sink: { setStream: ReturnType<typeof vi.fn>; dispose: ReturnType<typeof vi.fn> }
+  deps: {
+    getUserMedia: ReturnType<typeof vi.fn>
+    createPeerConnection: ReturnType<typeof vi.fn>
+    postSdp: ReturnType<typeof vi.fn>
+  }
+}
+
+function makeTransport(sdpOk = true): TransportHarness {
+  const token = deferred<{ value: string; model: string }>()
+  const media = deferred<FakeStream>()
+  const stream = new FakeStream()
+  const pcs: FakePeerConnection[] = []
+  const sink = { setStream: vi.fn(), dispose: vi.fn() }
+  const getUserMedia = vi.fn(() => media.promise)
+  const createPeerConnection = vi.fn(() => {
+    const pc = new FakePeerConnection()
+    pcs.push(pc)
+    return pc
+  })
+  const postSdp = vi.fn(async () => ({
+    ok: sdpOk,
+    status: sdpOk ? 201 : 401,
+    text: async () => (sdpOk ? 'answer-sdp' : 'unauthorized'),
+  }))
+  const deps = {
+    getToken: () => token.promise,
+    getUserMedia,
+    createPeerConnection,
+    postSdp,
+    createAudioSink: () => sink,
+    callsUrl: 'https://example.test/calls',
+  } as unknown as WebRtcTransportDeps
+  return {
+    transport: new WebRtcRealtimeTransport(deps),
+    token,
+    media,
+    stream,
+    pcs,
+    sink,
+    deps: { getUserMedia, createPeerConnection, postSdp },
+  }
+}
+
+async function driveToRemoteDescription(h: TransportHarness): Promise<FakePeerConnection> {
+  h.token.resolve({ value: 'ek_1', model: 'gpt-realtime' })
+  await flush()
+  h.media.resolve(h.stream)
+  await flush()
+  const pc = h.pcs[0]
+  expect(pc).toBeDefined()
+  return pc
+}
+
+describe('WebRtcRealtimeTransport', () => {
+  it('resolves connect only after the data channel opens', async () => {
+    const h = makeTransport()
+    const ac = new AbortController()
+    let settled = false
+    const p = h.transport.connect(ac.signal).then(() => {
+      settled = true
+    })
+    const pc = await driveToRemoteDescription(h)
+    expect(h.deps.postSdp).toHaveBeenCalledWith(
+      'https://example.test/calls?model=gpt-realtime',
+      expect.objectContaining({ body: 'offer-sdp' }),
+    )
+    pc.remote.resolve()
+    await flush()
+    expect(settled).toBe(false)
+    expect(h.transport.send({ type: 'x' })).toBe(false)
+
+    pc.dc.open()
+    await p
+    expect(settled).toBe(true)
+    expect(h.transport.send({ type: 'x' })).toBe(true)
+    expect(pc.dc.send).toHaveBeenCalledWith(JSON.stringify({ type: 'x' }))
+  })
+
+  it('abort during the token request never opens the microphone', async () => {
+    const h = makeTransport()
+    const ac = new AbortController()
+    const p = h.transport.connect(ac.signal)
+    ac.abort()
+    h.token.resolve({ value: 'ek', model: 'm' })
+    await expect(p).rejects.toBeInstanceOf(RealtimeAbortError)
+    expect(h.deps.getUserMedia).not.toHaveBeenCalled()
+    expect(h.deps.createPeerConnection).not.toHaveBeenCalled()
+  })
+
+  it('abort during getUserMedia stops the late microphone stream and creates no peer connection', async () => {
+    const h = makeTransport()
+    const ac = new AbortController()
+    const p = h.transport.connect(ac.signal)
+    h.token.resolve({ value: 'ek', model: 'm' })
+    await flush()
+    expect(h.deps.getUserMedia).toHaveBeenCalledTimes(1)
+    ac.abort()
+    h.media.resolve(h.stream)
+    await expect(p).rejects.toBeInstanceOf(RealtimeAbortError)
+    expect(h.stream.track.stop).toHaveBeenCalled()
+    expect(h.deps.createPeerConnection).not.toHaveBeenCalled()
+  })
+
+  it('abort while waiting for the channel releases pc, channel, mic and audio sink', async () => {
+    const h = makeTransport()
+    const ac = new AbortController()
+    const p = h.transport.connect(ac.signal)
+    const pc = await driveToRemoteDescription(h)
+    pc.remote.resolve()
+    await flush()
+    ac.abort()
+    await expect(p).rejects.toBeInstanceOf(RealtimeAbortError)
+    expect(pc.close).toHaveBeenCalled()
+    expect(pc.dc.close).toHaveBeenCalled()
+    expect(h.stream.track.stop).toHaveBeenCalled()
+    expect(h.sink.dispose).toHaveBeenCalled()
+  })
+
+  it('a failed SDP exchange rejects with the status and releases resources', async () => {
+    const h = makeTransport(false)
+    const p = h.transport.connect(new AbortController().signal)
+    const pc = await driveToRemoteDescription(h)
+    await expect(p).rejects.toThrow('SDP exchange failed (401): unauthorized')
+    expect(pc.close).toHaveBeenCalled()
+    expect(h.stream.track.stop).toHaveBeenCalled()
+  })
+
+  it('a peer connection failure before open rejects connect', async () => {
+    const h = makeTransport()
+    const p = h.transport.connect(new AbortController().signal)
+    const pc = await driveToRemoteDescription(h)
+    pc.remote.resolve()
+    await flush()
+    pc.setState('failed')
+    await expect(p).rejects.toThrow('Realtime connection lost')
+    expect(h.stream.track.stop).toHaveBeenCalled()
+  })
+
+  it('after open, a failed connection closes itself and notifies onClosed exactly once', async () => {
+    const h = makeTransport()
+    const onClosed = vi.fn()
+    h.transport.onClosed(onClosed)
+    const p = h.transport.connect(new AbortController().signal)
+    const pc = await driveToRemoteDescription(h)
+    pc.remote.resolve()
+    await flush()
+    pc.dc.open()
+    await p
+
+    pc.setState('disconnected') // 일시적 끊김은 종료로 보지 않는다
+    expect(onClosed).not.toHaveBeenCalled()
+
+    pc.setState('failed')
+    pc.setState('closed')
+    expect(onClosed).toHaveBeenCalledTimes(1)
+    expect(pc.close).toHaveBeenCalled()
+    expect(h.stream.track.stop).toHaveBeenCalled()
+    expect(h.transport.send({ type: 'x' })).toBe(false)
+  })
+
+  it('explicit close does not report onClosed, and forwards only valid JSON events', async () => {
+    const h = makeTransport()
+    const onClosed = vi.fn()
+    const onEvent = vi.fn()
+    h.transport.onClosed(onClosed)
+    h.transport.onEvent(onEvent)
+    const p = h.transport.connect(new AbortController().signal)
+    const pc = await driveToRemoteDescription(h)
+    pc.remote.resolve()
+    await flush()
+    pc.dc.open()
+    await p
+
+    pc.dc.receive('not json')
+    pc.dc.receive(JSON.stringify({ nope: true }))
+    pc.dc.receive(JSON.stringify({ type: 'response.created' }))
+    expect(onEvent).toHaveBeenCalledTimes(1)
+    expect(onEvent).toHaveBeenCalledWith({ type: 'response.created' })
+
+    h.transport.close()
+    expect(onClosed).not.toHaveBeenCalled()
+    expect(pc.close).toHaveBeenCalled()
+  })
+
+  it('is single-use', async () => {
+    const h = makeTransport()
+    h.transport.close()
+    await expect(h.transport.connect(new AbortController().signal)).rejects.toThrow('single-use')
+  })
+})
+
+// ── 컨트롤러 (fake transport) ──
+
+class FakeTransport implements RealtimeTransport {
+  readonly connectCall = deferred<void>()
+  open = false
+  ignoreAbort = false
+  sent: RealtimeClientEvent[] = []
+  close = vi.fn(() => {
+    this.open = false
+  })
+  private eventCb: ((e: RealtimeServerEvent) => void) | null = null
+  private closedCb: (() => void) | null = null
+
+  connect(signal: AbortSignal): Promise<void> {
+    signal.addEventListener('abort', () => {
+      if (!this.ignoreAbort) this.connectCall.reject(new RealtimeAbortError())
+    })
+    return this.connectCall.promise.then(() => {
+      this.open = true
+    })
+  }
+  send(event: RealtimeClientEvent): boolean {
+    if (!this.open) return false
+    this.sent.push(event)
+    return true
+  }
+  onEvent(cb: (e: RealtimeServerEvent) => void): () => void {
+    this.eventCb = cb
+    return () => undefined
+  }
+  onClosed(cb: () => void): () => void {
+    this.closedCb = cb
+    return () => undefined
+  }
+  emit(e: RealtimeServerEvent): void {
+    this.eventCb?.(e)
+  }
+  drop(): void {
+    this.open = false
+    this.closedCb?.()
+  }
+}
+
+function makeController(): { controller: RealtimeSessionController; transports: FakeTransport[] } {
+  const transports: FakeTransport[] = []
+  const controller = new RealtimeSessionController(() => {
+    const t = new FakeTransport()
+    transports.push(t)
+    return t
+  }, makeCtx())
+  return { controller, transports }
+}
+
+describe('RealtimeSessionController', () => {
+  it('stop during connecting cancels the attempt, closes the transport and never goes live', async () => {
+    const { controller, transports } = makeController()
+    const p = controller.start()
+    expect(controller.getSnapshot().connectionState).toBe('connecting')
+    controller.stop()
+    expect(transports[0].close).toHaveBeenCalled()
+    await expect(p).resolves.toBe('cancelled')
+    expect(controller.getSnapshot().connectionState).toBe('idle')
+  })
+
+  it('a connect that completes after stop (abort ignored) is closed and never goes live', async () => {
+    const { controller, transports } = makeController()
+    const p = controller.start()
+    const t = transports[0]
+    t.ignoreAbort = true
+    controller.stop()
+    t.close.mockClear()
+    t.connectCall.resolve()
+    await expect(p).resolves.toBe('cancelled')
+    expect(t.close).toHaveBeenCalled()
+    expect(t.sent).toEqual([])
+    expect(controller.getSnapshot().connectionState).toBe('idle')
+  })
+
+  it('concurrent start calls share one connection attempt', async () => {
+    const { controller, transports } = makeController()
+    const a = controller.start()
+    const b = controller.start()
+    expect(transports).toHaveLength(1)
+    transports[0].connectCall.resolve()
+    await expect(a).resolves.toBe('live')
+    await expect(b).resolves.toBe('live')
+    await expect(controller.start()).resolves.toBe('live')
+    expect(transports).toHaveLength(1)
+  })
+
+  it('sends session.update when live and sendText right after start is delivered', async () => {
+    const { controller, transports } = makeController()
+    const p = controller.start()
+    transports[0].connectCall.resolve()
+    await expect(p).resolves.toBe('live')
+    controller.sendText('hello')
+    const t = transports[0]
+    expect(t.sent[0]).toEqual(REALTIME_SESSION_UPDATE)
+    expect(t.sent.slice(1).map((e) => e.type)).toEqual(['conversation.item.create', 'response.create'])
+    expect(controller.getSnapshot().messages.map((m) => m.content)).toEqual(['hello'])
+    expect(controller.getSnapshot().conversationState).toBe('listening')
+  })
+
+  it('queues text sent while connecting and flushes it after session.update', async () => {
+    const { controller, transports } = makeController()
+    const p = controller.start()
+    controller.sendText('first')
+    expect(controller.getSnapshot().messages.map((m) => m.content)).toEqual(['first'])
+    expect(transports[0].sent).toEqual([])
+    transports[0].connectCall.resolve()
+    await p
+    expect(transports[0].sent.map((e) => e.type)).toEqual([
+      'session.update',
+      'conversation.item.create',
+      'response.create',
+    ])
+  })
+
+  it('after the connection drops, the session reports error and start reconnects with a new transport', async () => {
+    const { controller, transports } = makeController()
+    const p = controller.start()
+    transports[0].connectCall.resolve()
+    await p
+    transports[0].emit({ type: 'response.created' })
+    transports[0].emit({ type: 'response.text.delta', delta: 'partial' })
+
+    transports[0].drop()
+    const snap = controller.getSnapshot()
+    expect(snap.connectionState).toBe('error')
+    expect(snap.conversationState).toBe('idle')
+    expect(snap.messages.map((m) => m.content)).toEqual(['partial'])
+
+    const again = controller.start()
+    expect(transports).toHaveLength(2)
+    transports[1].connectCall.resolve()
+    await expect(again).resolves.toBe('live')
+    // 이전 전송 계층의 늦은 이벤트는 무시된다
+    transports[0].emit({ type: 'response.created' })
+    expect(controller.getSnapshot().conversationState).toBe('listening')
+  })
+
+  it('a failed connect reports failed with the error message and allows retry', async () => {
+    const { controller, transports } = makeController()
+    const p = controller.start()
+    transports[0].connectCall.reject(new Error('token request failed'))
+    await expect(p).resolves.toBe('failed')
+    expect(controller.getSnapshot()).toMatchObject({
+      connectionState: 'error',
+      conversationState: 'idle',
+      error: 'token request failed',
+    })
+    void controller.start()
+    expect(transports).toHaveLength(2)
+    expect(controller.getSnapshot().error).toBeNull()
+  })
+
+  it('notifies subscribers and applies server events only from the current transport', async () => {
+    const { controller, transports } = makeController()
+    const listener = vi.fn()
+    const unsubscribe = controller.subscribe(listener)
+    const p = controller.start()
+    transports[0].connectCall.resolve()
+    await p
+    listener.mockClear()
+    transports[0].emit({ type: 'response.created' })
+    expect(listener).toHaveBeenCalledTimes(1)
+    expect(controller.getSnapshot().conversationState).toBe('thinking')
+    controller.cancelResponse()
+    expect(transports[0].sent.at(-1)).toEqual({ type: 'response.cancel' })
+    expect(controller.getSnapshot().conversationState).toBe('listening')
+    unsubscribe()
+  })
+
+  it('sendText is ignored when no session exists', () => {
+    const { controller, transports } = makeController()
+    controller.sendText('x')
+    expect(transports).toHaveLength(0)
+    expect(controller.getSnapshot().messages).toEqual([])
+  })
+})