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

// 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([])
  })
})