# How the Hugging Face Speech-to-Speech Pipeline Handles Concurrent WebSocket Sessions

> Learn how the Hugging Face Speech-to-Speech pipeline efficiently handles concurrent WebSocket sessions using isolated PipelineUnit instances and private queues for zero cross-talk.

- Repository: [Hugging Face/speech-to-speech](https://github.com/huggingface/speech-to-speech)
- Tags: internals
- Published: 2026-08-04

---

**The huggingface/speech-to-speech pipeline manages concurrent WebSocket sessions through a fixed pool of isolated PipelineUnit instances, where each incoming connection atomically claims a dedicated unit with private queues and event loops, ensuring zero cross-talk between clients.**

The `huggingface/speech-to-speech` repository implements a realtime speech-to-speech server that must handle multiple simultaneous WebSocket connections without interference. Understanding how the pipeline handles concurrent WebSocket sessions is essential for anyone deploying this system in production or extending its architecture. The implementation uses a resource-pool pattern with strict isolation guarantees, graceful session lifecycle management, and robust cleanup mechanisms for stuck or crashed handlers.

## Pool Architecture: Building Isolated Pipeline Units

When the application starts in **realtime mode** (`module_kwargs.mode == "realtime"`), the `build_pipeline()` function in [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) constructs a fixed-size pool of independent processing units. This pool size is controlled by the `num_pipelines` configuration parameter.

```python
pool = [_build_realtime_pipeline_unit(... ) for i in range(pool_size)]

```

Each **PipelineUnit** is a self-contained processing chain comprising:

- **Input/output queues** for audio and text events
- **Threading events** for synchronization
- A **RealtimeService** instance for OpenAI-compatible protocol handling
- A full handler pipeline: VAD → STT → Language Model → TTS

This design ensures that every unit operates independently. No shared queues or state exists between units, which makes the system inherently thread-safe for concurrent access.

Reference: [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) lines 7949–7956

## Claiming Units: Atomic Session Assignment

New WebSocket connections arrive at `websocket_router.realtime_endpoint`. The endpoint immediately attempts to reserve a pipeline unit through `_claim_unit(transport)`, which performs an **atomic scan** of the pool for the first idle unit.

```python
unit = _claim_unit(transport)

```

The claim logic checks `unit.session is None` to identify available units. If found, the unit is marked as reserved by attaching a new **SessionState** object. If all units are busy, the server returns an error and closes the WebSocket with code `1008` ("Policy Violation") and the message "All session slots are in use."

```python
if unit is None:
    await send_ws_event(ws, build_error_event(...))
    await ws.close(code=1008, reason="All session slots are in use")
    return

```

This fail-fast approach prevents resource exhaustion and gives clients clear feedback about server capacity.

Reference: [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) lines 7459–7466, 759–778

## Session State: Per-Client Context Isolation

Once claimed, a unit's `SessionState` object (defined in [`src/speech_to_speech/api/openai_realtime/pipeline_unit.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/pipeline_unit.py)) stores all client-specific context:

| Field | Purpose |
|-------|---------|
| `transport` | FastAPI WebSocket or WebRTC transport wrapper |
| `session_id` | UUID generated by `RealtimeService.register()` for OpenAI compatibility |
| `drained` | Boolean flag indicating graceful shutdown progress |
| `created_at` | Timestamp for session lifetime tracking |
| `release_after` / `quarantined_at` | Timestamps for cleanup timeout detection |

Reference: [`src/speech_to_speech/api/openai_realtime/pipeline_unit.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/pipeline_unit.py) lines 13–31

## Event Processing: Private Queues Per Session

While a connection remains active, the endpoint runs a receive loop that forwards all client events to `_dispatch_client_event`. This dispatcher routes payloads to appropriate handlers based on event type (`session.update`, `input_audio_buffer.append`, `response.create`, etc.).

Crucially, **all handlers operate exclusively on the claimed unit's private queues**. The VAD handler consumes from the unit's audio input queue; the STT handler consumes VAD output; the language model consumes STT output; and the TTS handler consumes LM output. No global or shared queues exist that could cause event mixing between sessions.

Reference: [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) lines 501–527

## Send Loop: Background Output Streaming

For every `PipelineUnit`, the server launches a persistent background task via `_send_loop_for(unit)`. This loop continuously polls the unit's output queues and transmits audio chunks or text events back to the client.

The implementation includes a critical safety pattern: the loop **snapshots `unit.session` at each iteration**. This ensures that if a client disconnects mid-iteration, the loop completes processing of any in-flight items using the captured session reference before terminating. Without this pattern, partially processed outputs could be lost or misdirected.

Reference: [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) lines 7222–7236

## Graceful Release: Session Cleanup and Unit Reclamation

When a client disconnects—whether through normal closure, `WebSocketDisconnect` exception, or protocol error—the endpoint's `finally` block triggers `_release_session(unit, session_id)`. This orchestrates a multi-stage cleanup:

1. **Enqueue termination sentinel**: A `SESSION_END` event is pushed to the unit's input queue
2. **Flush pending queues**: Any buffered but unprocessed events are cleared
3. **Start drain task**: `_release_unit_after_drain` spawns asynchronously to wait for handler chain propagation
4. **Await completion**: The drain task blocks until `SESSION_END` reaches the final handler or a timeout expires
5. **Clear session state**: `unit.session` is set to `None`, making the unit claimable again

```python

# Executed in the finally block of the WebSocket endpoint

_release_session(unit, session_id)   # queues SESSION_END and starts drain task

```

This staged approach prevents race conditions where a unit might be reclaimed while handlers still process the previous session's data.

Reference: [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) lines 508–518, 550–557

## Quarantine Handling: Detecting Stuck Units

Production systems must account for handler crashes or infinite loops. If a handler thread fails to drain the `SESSION_END` sentinel, the release task enforces a hard timeout of **180 seconds** (`SESSION_END_QUARANTINE_TIMEOUT_S`). Upon timeout:

- The unit is marked with `quarantined_at` timestamp
- The unit **remains unclaimable** until manual intervention
- Operators can inspect pool state via the `/v1/pool` health endpoint

This quarantine mechanism prevents "zombie" units from silently degrading system capacity and provides observability for debugging stuck pipelines.

Reference: [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) lines 569–586

## Configuration Example: Starting a Server with Concurrent Pipelines

```python
from speech_to_speech.s2s_pipeline import parse_arguments, prepare_all_args, build_pipeline

args = parse_arguments()                     # parses CLI / JSON config

prepare_all_args(*args)                      # normalises arguments

queues_and_events = initialize_queues_and_events()

# Build a pool of 3 concurrent realtime pipelines

pipeline_manager = build_pipeline(
    args.module_kwargs,
    args.socket_receiver_kwargs,
    args.socket_sender_kwargs,
    args.websocket_streamer_kwargs,
    args.vad_handler_kwargs,
    args.whisper_stt_handler_kwargs,
    args.faster_whisper_stt_handler_kwargs,
    args.paraformer_stt_handler_kwargs,
    args.mlx_audio_whisper_stt_handler_kwargs,
    args.parakeet_tdt_stt_handler_kwargs,
    args.language_model_handler_kwargs,
    args.responses_api_language_model_handler_kwargs,
    args.chat_tts_handler_kwargs,
    args.facebook_mms_tts_handler_kwargs,
    args.pocket_tts_handler_kwargs,
    args.kokoro_tts_handler_kwargs,
    args.qwen3_tts_handler_kwargs,
    queues_and_events,
)

pipeline_manager.start()    # launches all pipeline units and their send loops

pipeline_manager.wait()     # blocks until shutdown signal

```

Reference: [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) lines 7890–7904

## Key Files for Concurrent Session Management

| File | Responsibility |
|------|---------------|
| [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) | WebSocket endpoint, unit claim/release, event dispatch, quarantine logic |
| [`src/speech_to_speech/api/openai_realtime/pipeline_unit.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/pipeline_unit.py) | `PipelineUnit` and `SessionState` definitions |
| [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) | Pool construction and pipeline lifecycle management |
| [`src/speech_to_speech/api/openai_realtime/transports.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/transports.py) | Abstract transport layer for WebSocket and WebRTC |
| [`src/speech_to_speech/api/openai_realtime/service.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/service.py) | RealtimeService for OpenAI protocol parsing |

## Summary

- **Pool-based isolation**: The pipeline creates a fixed number of `PipelineUnit` instances at startup, with each unit containing complete independent processing chains
- **Atomic claiming**: `_claim_unit()` provides lock-free reservation of idle units by checking `unit.session is None`
- **Queue isolation**: No shared queues exist between sessions; all handler communication uses per-unit private queues
- **Graceful cleanup**: The `SESSION_END` sentinel and drain task ensure complete pipeline flushing before unit reclamation
- **Failure containment**: The 180-second quarantine timeout prevents crashed handlers from permanently consuming pool capacity

## Frequently Asked Questions

### What limits the number of concurrent WebSocket connections?

The `num_pipelines` configuration parameter sets the pool size. Each connection requires one dedicated `PipelineUnit`, so maximum concurrency equals this value. When the pool is exhausted, new connections receive an immediate error with WebSocket close code `1008`.

### Can one client's audio data leak into another client's session?

No. The architecture guarantees isolation through private queues per `PipelineUnit`. Handlers never access queues from other units, and the claim/release protocol ensures only one session uses a unit at any time.

### What happens if a handler crashes during active processing?

The `SESSION_END` quarantine mechanism detects stuck units. If handlers fail to drain the termination sentinel within 180 seconds, the unit is marked unclaimable with `quarantined_at` timestamp. Operators must restart the server or investigate via the `/v1/pool` endpoint.

### Is the pool size fixed at runtime?

Yes. The `huggingface/speech-to-speech` implementation uses a static pool allocated at startup. Dynamic pool resizing would require significant architectural changes to the claim/release synchronization logic in [`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py).