Audio Streaming Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                              MEMQL SERVER                                │
│                                                                          │
│  /memql/ws  (gRPC tunneled over WS)         /memql/audio (legacy WS)     │
│   - All gRPC messages incl. AiTranscribe*    - In-space STT + TTS chunks │
│   - Queries / mutations / subscriptions      - Single-user per stream    │
│   - Streaming transcription                                              │
│                                                                          │
│  HTTP (browser-required exceptions only):                                │
│    /auth/*        OAuth callbacks (HTTP-required)                        │
│    /healthz       Health probe                                           │
│    /spaces/{id}/attachments  multipart upload                            │
│    /polyphon/room-token, /polyphon/status  Polyphon multi-agent voice    │
│                                                                          │
│  All paths share the same identity-service-validated context.            │
└─────────────────────────────────────────────────────────────────────────┘

{
  "type": "start",
  "streamId": "550e8400-e29b-41d4-a716-446655440000",
  "spaceId": "space-123",
  "participantId": "participant-456",
  "format": "pcm16",
  "sampleRate": 16000,
  "channels": 1,
  "languageHint": "en"
}

{
  "type": "chunk",
  "streamId": "550e8400-e29b-41d4-a716-446655440000",
  "audio": "SGVsbG8gV29ybGQ=",
  "sequence": 1
}

{
  "type": "end",
  "streamId": "550e8400-e29b-41d4-a716-446655440000",
  "cancelled": false
}

{
  "type": "started",
  "streamId": "550e8400-e29b-41d4-a716-446655440000"
}

{
  "type": "transcription",
  "streamId": "550e8400-e29b-41d4-a716-446655440000",
  "text": "Hello, how are you?",
  "isFinal": true,
  "confidence": 0.95,
  "words": [
    { "word": "Hello", "start": 0, "end": 320, "confidence": 0.98 },
    { "word": "how", "start": 350, "end": 480, "confidence": 0.94 },
    { "word": "are", "start": 500, "end": 580, "confidence": 0.96 },
    { "word": "you", "start": 600, "end": 750, "confidence": 0.93 }
  ],
  "utteranceId": "utt-voice-1702156800000000000",
  "durationMs": 1500
}

{
  "type": "error",
  "streamId": "550e8400-e29b-41d4-a716-446655440000",
  "error": {
    "code": "STREAM_NOT_FOUND",
    "message": "No active stream with this ID"
  }
}

function float32ToPcm16(float32Array) {
  const pcm16 = new Int16Array(float32Array.length);
  for (let i = 0; i < float32Array.length; i++) {
    const s = Math.max(-1, Math.min(1, float32Array[i]));
    pcm16[i] = s < 0 ? s * 0x8000 : s * 0x7FFF;
  }
  return pcm16;
}

1. User presses mic button
2. Client opens /memql/audio WebSocket (if not already open)
3. Client sends "start" message with spaceId, participantId
4. Client captures audio via getUserMedia + AudioWorklet
5. Client converts Float32 to PCM16, base64 encodes, sends "chunk" messages
6. Server forwards chunks to STT provider (Deepgram Nova-3 when configured; OpenAI Realtime / OpenAI Whisper otherwise)
7. Server receives interim transcriptions, sends to client (isFinal: false)
8. User releases mic button
9. Client sends "end" message
10. Server finalizes STT stream, gets complete transcription
11. Server inserts v1:cognition:utterance with:
    - utteranceType: "speech"
    - source.inputMethod: "stt"
    - source.sttProvider: configured provider name
    - timestamps.words: word-level timing
12. Server sends final transcription event (isFinal: true) with utteranceId
13. Event bus emits graph.node.created.v1:cognition:utterance
14. All participants receive utterance via /memql/ws subscription
15. Chat UI displays the voice message

{
  "concept": "v1:cognition:utterance",
  "id": "utt-voice-1702156800000000000",
  "payload": {
    "spaceId": "space-123",
    "participantId": "participant-456",
    "utteranceType": "speech",
    "text": "Hello, how are you?",
    "duration": 1500,
    "timestamps": {
      "words": [
        { "word": "Hello", "start": 0, "end": 320 },
        { "word": "how", "start": 350, "end": 480 },
        { "word": "are", "start": 500, "end": 580 },
        { "word": "you", "start": 600, "end": 750 }
      ]
    },
    "source": {
      "inputMethod": "stt",
      "sttProvider": "openai-whisper"
    }
  }
}

server/audiows/
├── handler.go      # WebSocket handler, session management
└── messages.go     # Message type definitions
 
integrations/stt/
├── stt.go              # Provider interface, common types
├── openai_whisper.go   # OpenAI Whisper (batch)
├── openai_realtime.go  # OpenAI Realtime (streaming)
└── deepgram.go         # Deepgram Nova-3 (streaming)

// StreamingProvider provides real-time streaming transcription
type StreamingProvider interface {
    // StartStream begins a new streaming session
    StartStream(ctx context.Context, config StreamConfig) (StreamingSession, error)
 
    // Name returns the provider name
    Name() string
}
 
// StreamingSession represents an active transcription session
type StreamingSession interface {
    // SendAudio sends audio data to the STT service
    SendAudio(audio []byte) error
 
    // Receive returns a channel for transcription events
    Receive() <-chan TranscriptionResult
 
    // Finalize closes the stream and returns final transcription
    Finalize(ctx context.Context) (*FinalTranscription, error)
 
    // Close terminates without waiting for final result
    Close() error
}

{
  "type": "synthesize",
  "requestId": "req-550e8400-e29b-41d4-a716-446655440000",
  "text": "Hello, how are you today?",
  "voice": "nova",
  "format": "wav",
  "sampleRate": 24000
}

{
  "type": "tts_started",
  "requestId": "req-550e8400-e29b-41d4-a716-446655440000",
  "format": "wav",
  "sampleRate": 24000,
  "spaceId": "space-123",
  "participantId": "ai-participant-456",
  "text": "Hello, how can I help you?"
}

{
  "type": "tts_chunk",
  "requestId": "req-550e8400-e29b-41d4-a716-446655440000",
  "audio": "UklGRiQAAABXQVZFZm10IBAAAA...",
  "format": "wav",
  "sampleRate": 24000,
  "sequence": 0,
  "done": false,
  "spaceId": "space-123",
  "participantId": "ai-participant-456",
  "text": "Hello, how can I help you?"
}

{
  "type": "tts_ended",
  "requestId": "req-550e8400-e29b-41d4-a716-446655440000",
  "spaceId": "space-123",
  "participantId": "ai-participant-456"
}

// Each chunk is a complete WAV file - decode and play immediately!
for await (const chunk of ttsStream) {
  const wavBuffer = base64ToArrayBuffer(chunk.audio);
  const audioBuffer = await audioContext.decodeAudioData(wavBuffer);
  // Queue immediately for playback - starts playing within ~200ms
  queueAudioForPlayback(audioBuffer);
}

1. User clicks "Read Aloud" on a message
2. Client sends "synthesize" message with text and requestId
3. Server sends "tts_started" message with format info
4. Server calls OpenAI TTS API (from engine provider registry)
5. Server streams "tts_chunk" messages (WAV audio)
6. Client uses native decodeAudioData() for playback
7. Server sends "tts_ended" on completion

client -> server                        server -> client
─────────────────────────────────       ─────────────────────────────────
AiTranscribeStreamStart {               AiTranscribeStreamDelta {
  request_id, sample_rate, ...           request_id, text, is_final
}                                       }     (zero or more interim deltas)
AiTranscribeStreamChunk {              AiTranscribeStreamComplete {
  request_id, audio  (PCM16 bytes)       request_id, transcript, words
}                                       }
... more chunks ...
AiTranscribeStreamEnd { request_id }