# Session Lifecycle and PipelineUnit Pool Management in Speech-to-Speech Realtime API

> Understand the session lifecycle in the Speech-to-Speech realtime API. Learn how the PipelineUnit pool manages concurrent connections with claim, drain, and release to ensure isolation.

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

---

**The Speech-to-Speech realtime API maintains a pool of isolated PipelineUnit objects where each unit handles exactly one websocket session, ensuring complete isolation between concurrent connections through a claim-drain-release lifecycle.**

The Hugging Face `speech-to-speech` repository implements a robust realtime architecture that manages multiple concurrent speech-to-speech conversations using a fixed pool of pipeline units. Understanding how the **session lifecycle** works and how the **PipelineUnit pool** handles concurrent connections is essential for operating the server at scale and debugging connection issues.

## Session Creation: Claiming a PipelineUnit

When a websocket client connects, the FastAPI route in [`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) calls **`_claim_unit`** (lines 282-296) to acquire an available pipeline unit. The function iterates through the pool and selects the first unit whose `session` attribute is `None`.

```python
for unit in pool:
    if unit.session is None:
        unit.session = SessionState(websocket=ws)  # Creates fresh SessionState

        return unit

```

Upon successful claim, the router creates a new **`SessionState`** instance (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), lines 12-33) which encapsulates:

- **`websocket`** — The actual WebSocket object for this client connection
- **`session_id`** — Unique identifier generated by `RealtimeService.register()`
- **`pending_output_item`** — Temporary holder for the audio item currently being transmitted
- **`drained`** — An `asyncio.Event` that signals when the `SESSION_END` marker has cleared the entire handler chain
- **`released_at`** — Timestamp captured at disconnect to help identify "stuck" units

After claiming the unit, the route registers the session with the service (`session_id = unit.service.register()`) and initiates a dedicated send loop (`_send_loop_for(unit)`) that forwards processed audio chunks to the client.

## Normal Operation and Thread Isolation

While the connection remains active, the unit's send loop continuously polls `unit.text_output_queue` (or `unit.output_queue`) and streams data to the client via the stored websocket. Each generated chunk updates `unit.session.pending_output_item` to track in-flight data.

All heavy processing—including Voice Activity Detection (VAD), Speech-to-Text (STT), Language Model (LM) inference, and Text-to-Speech (TTS) synthesis—runs within separate threads managed by `ThreadManager`. However, every thread reads from and writes to the **same per-unit queues**, guaranteeing that concurrent clients never interfere with one another's data streams.

## Session Termination: The Release Process

When a websocket disconnects, the route handler executes a graceful shutdown sequence to prevent data leakage between sessions. First, it aborts ongoing work and flushes pending data (see [`websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/websocket_router.py), lines 166-185):

```python
unit.cancel_scope.cancel()          # Abort any ongoing generation

_flush_queue(unit.input_queue)      # Drop pending requests

_flush_queue(unit.text_prompt_queue)
_flush_queue(unit.output_queue, preserve=...)
_flush_queue(unit.text_output_queue, preserve=...)
unit.response_playing.clear()
unit.cancel_scope.reset()
unit.should_listen.set()

```

After cancellation, the router invokes **`_release_unit_after_drain`** (lines 227-255) to asynchronously wait for the `SESSION_END` marker to propagate through the entire handler chain. This function:

1. Awaits `unit.session.drained.wait()` until the session-end marker clears all queues
2. Calls `unit.service.unregister(session_id)` to remove usage metrics
3. Clears `unit.session` back to `None`, making the unit available for new claims
4. Logs the release event

Only after `unit.session` is set to `None` can another client claim that specific `PipelineUnit`. This ensures no residual audio or processing state from the previous session contaminates new connections.

## Handling Multiple Concurrent Connections

The pool itself is a simple Python list created at server startup in [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) (lines 448-475). The pool size determines the absolute maximum number of simultaneous websocket connections the server can handle.

The pool operates under two conditions:

- **Available units exist** — `_claim_unit` immediately assigns the first free unit to the incoming connection
- **Pool exhaustion** — If all units have active sessions, `_claim_unit` returns `None`, triggering the router to send a stateless error event (HTTP 503-style response) indicating "no pipeline unit available," forcing the client to retry or back off

Because each `PipelineUnit` maintains its own queues, `CancelScope`, and `SessionState`, **concurrent connections remain completely isolated** regardless of how many units are active simultaneously.

## Observability and Monitoring

The `/v1/pool` endpoint (implemented in [`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)) exposes the current state of every unit in the pool, including:

- `session_id` (or `null` if the unit is idle)
- `released_at` timestamp (if the unit is waiting for `SESSION_END` to drain)

This allows operators to identify "stuck" units that have not completed their shutdown sequence or detect imbalances in pool utilization.

## Summary

- The **PipelineUnit pool** is a fixed-size list of isolated processing units created at server startup in [`s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/s2s_pipeline.py).
- Each websocket connection claims one unit via `_claim_unit`, which attaches a `SessionState` object containing the session ID, websocket reference, and drainage event.
- All processing threads for a given session read from and write to per-unit queues, ensuring **complete isolation between concurrent connections**.
- On disconnect, the system cancels in-flight work, flushes queues, and waits for the `SESSION_END` marker to drain before clearing `unit.session` back to `None`.
- If the pool is exhausted, new connections receive a 503-style error and must retry.
- The `/v1/pool` endpoint provides visibility into unit status and helps diagnose stuck sessions.

## Frequently Asked Questions

### How does the system prevent audio from one session leaking into another?

Each `PipelineUnit` maintains private queues (`input_queue`, `output_queue`, `text_output_queue`) and a dedicated `CancelScope`. When a client disconnects, the router flushes all unit-specific queues and waits for the `drained` event before marking the unit as available. Since units are never shared between active sessions, audio data cannot cross between clients.

### What happens when the PipelineUnit pool is full?

When all units have non-null `session` attributes, `_claim_unit` returns `None`. The websocket router then sends a stateless error event indicating that no pipeline units are available, effectively applying back-pressure. Clients must implement retry logic with exponential backoff to handle this condition gracefully.

### How can I monitor which sessions are currently active or stuck?

Query the `/v1/pool` endpoint, which returns the status of every unit including the `session_id` (if active) and `released_at` timestamp (if the unit is in the process of draining). Units with a `released_at` value but no active session indicate potential shutdown issues that may require investigation.

### Why does the release process wait for the SESSION_END marker to drain?

The `SESSION_END` marker acts as a pipeline barrier that ensures all in-flight audio processing (VAD → STT → LM → TTS) has completed and all output queues are empty before the unit is recycled. This prevents truncated audio or partial transcripts from persisting into the next session that claims the unit.