vllm-omni/docs/design/fullduplex.md

# Unified Full-Duplex Framework

This document is the design record of the full-duplex serving framework:
the engine-resident session model, the typed command/event contract, the
single model plugin, and the serving/API surfaces built on top of them. It is
the owner document for `vllm_omni/engine/duplex/**`,
`vllm_omni/engine/duplex_omni_engine.py`,
`vllm_omni/engine/duplex_orchestrator.py`,
`vllm_omni/entrypoints/duplex_omni.py`, `vllm_omni/entrypoints/duplex/**`,
`vllm_omni/clients/**` and the per-model `duplex/` packages.

`vllm_omni/protocol/realtime/**` (the OpenAI Realtime API) and
`vllm_omni/protocol/duplex/**` (vLLM-Omni's extension of it) are the shared
wire codec duplex is built on. Duplex is its first consumer but not its owner: the codec
is deliberately runtime-agnostic so a second Realtime surface can reuse it
(RFC #6592 P0a).

The public wire contract lives in
[`docs/serving/realtime_duplex_api.md`](../serving/realtime_duplex_api.md);
this page describes how it is produced.

## Purpose

A full-duplex model listens and speaks at the same time: audio streams into
the model while it produces audio, the model decides when to speak, and the
user can interrupt. Serving such a model needs a long-lived session with its
own identity (epoch, turn), ledgers (input, response, playback, history), a
lease (idle TTL, disconnect grace, resume) and a binding to the resumable
stage requests that carry the session's audio through the pipeline.

The framework gives that session exactly one owner, keeps the generic
turn-based stack free of duplex code, and exposes the same session to three
consumers: the OpenAI Realtime WebSocket route, the in-process Python API and
the model plugins.

## Architecture

```text
              Python users                                WebSocket clients
        ┌──────────────────────┐                    ┌────────────────────────┐
        │  InlineDuplexClient  │                    │      DuplexClient      │
        │  (DuplexClientBase)  │                    │   (DuplexClientBase)   │
        └──────────┬───────────┘                    └───────────┬────────────┘
                   │   DuplexSessionHandle                      │ /v1/realtime?duplex=1 (alias /v1/duplex)
                   │                                            ▼
                   │                          ┌──────────────────────────────────────┐
                   │                          │ OmniDuplexSessionHandler (thin)      │
                   │                          │  websocket I/O, command_from_realtime│
                   │                          │  event.to_realtime(), attachment /   │
                   │                          │  resume tokens / replay journal      │
                   │                          └──────────────────┬───────────────────┘
                   ▼                                             ▼   DuplexSessionHandle
        ┌──────────────────────────────────────────────────────────────────────────────┐
        │ DuplexOmni(AsyncOmni)                      [entrypoints/duplex_omni.py, thin] │
        │   open_session / get_session / resume_session / detach_session / close_session│
        │   DuplexSessionHandle: submit(DuplexCommand) / events() -> DuplexEvent       │
        └──────────────────────────────────────┬───────────────────────────────────────┘
                                               │ engine.open/close/resume/touch_session_async (RPC)
                                               │ engine.submit_command_async (one-way)
                                               ▼
        ┌──────────────────────────────────────────────────────────────────────────────┐
        │ DuplexOmniEngine(AsyncOmniEngine) [engine/duplex_omni_engine.py, thin]        │
        │   loads the plugin, validates session_mode, builds queue messages            │
        │  └─ DuplexOrchestrator(Orchestrator)            [engine/duplex_orchestrator.py]│
        │       template seams + DuplexStagePort over the shared stage machinery       │
        │       ├─ DuplexSessionManager      admission, backpressure, lease reaper,     │
        │       │                            open/close/resume/touch, command dispatch  │
        │       └─ DuplexSessionRunner ×N    ONE session state (DuplexEngineSession),   │
        │                                    ordered mailbox, append planning + stage   │
        │                                    submit, output projection, response /     │
        │                                    playback / history, VAD, typed events     │
        │       plugin: DuplexModelPlugin    (PipelineConfig.duplex_plugin, one/model)  │
        └──────────────────────────────────────────────────────────────────────────────┘
```

In each layer the duplex class extends the turn-based one, which extends
the shared base. The dependency runs duplex -> turn-based and never the
reverse, so the turn-based stack is unchanged and the duplex stack reuses
its request machinery instead of duplicating it:

```text
OmniBase                          OmniEngineBase                        OrchestratorBase
 ├─ Omni       (sync, turn)        └─ AsyncOmniEngine  (turn requests)   └─ Orchestrator        (turn-based admission)
 └─ AsyncOmniBase                      └─ DuplexOmniEngine (session msgs)    └─ DuplexOrchestrator  (hosts DuplexSessionManager,
     └─ AsyncOmni  (EngineClient)                                                                     implements DuplexStagePort)
         └─ DuplexOmni

AsyncOmni._create_engine  -> AsyncOmniEngine     AsyncOmniEngine._create_orchestrator  -> Orchestrator
DuplexOmni._create_engine -> DuplexOmniEngine    DuplexOmniEngine._create_orchestrator -> DuplexOrchestrator
```

Each concrete class constructs its collaborator in an explicit factory
method the base calls at the right moment: `OmniBase.__init__` calls
`self._create_engine(...)`; `OmniEngineBase._bootstrap_orchestrator` calls
`self._create_orchestrator(...)` on the orchestrator thread once the stages
are initialized. `DuplexOmniEngine._create_orchestrator` passes the loaded
plugin and the session runtime config to `DuplexOrchestrator` directly.

### Design rules

1. **One session authority.** The whole duplex session (config, epoch/turn,
   ledgers, lease, stage request resources, model state, projection state) is
   one object, `DuplexEngineSession`, owned by one `DuplexSessionRunner` on
   the orchestrator loop. Nothing above the engine keeps session state beyond
   a handle (session id, event queue, capabilities, public session object).
2. **Typed contract.** Commands into a session are `DuplexCommand`
   dataclasses (`engine/duplex/commands.py`); outputs are `DuplexEvent`
   dataclasses (`engine/duplex/events.py`). `command_from_realtime()` and
   `DuplexEvent.to_realtime()` derive the OpenAI Realtime JSON, so the wire
   format is never hand-built, and the websocket handler and the inline client
   share one conversion.
3. **Generic bases and the turn-based classes have zero duplex vocabulary.**
   `OmniBase`, `AsyncOmniBase`, `AsyncOmni`, `OmniEngineBase`,
   `AsyncOmniEngine`, `OrchestratorBase` and `Orchestrator` carry only
   template seams; `tests/engine/test_duplex_import_boundary.py` checks that
   importing the turn-based stack loads no duplex module.
4. **Deployment selects the serving stack.**
   `vllm-omni serve` constructs `DuplexOmni` when the pipeline declares
   `duplex_plugin` and the deploy configuration sets `session_mode: duplex`.
   An explicit `session_mode: turn` selects `AsyncOmni` instead, without
   changing the pipeline's duplex capability. In duplex mode the server exposes `/v1/realtime?duplex=1` (alias
   `/v1/duplex`), `POST /v1/chat/completions`, `/v1/models` and `/health`, and
   every other turn-based route reports "not available". The chat route uses
   the ordinary chat service on the duplex engine, without a session, when
   the plugin declares `DuplexCapabilities.supports_chat_completions`.
   Turn-based deployments use the ordinary API initialization and retain
   model capability checks and endpoint restrictions.
5. **Serving is transport only.** The websocket handler does socket I/O,
   wire-envelope validation, command translation, event rendering and the
   attachment/resume/replay bookkeeping; it holds no session state.
6. **No `typing.Protocol`** in the duplex surfaces: the plugin, data plane,
   session state, PCM buffer, stage port and client transport seams are ABCs.
7. **The wire codec is not duplex, and a wire type is not a mailbox message.**
   The event and command *vocabulary* plus its wire rendering, the audio format
   negotiation, the conversation-item rules and the error envelope are model-
   and runtime-agnostic, and split along the three tiers
   [`docs/serving/realtime_duplex_api.md`](../serving/realtime_duplex_api.md)
   already defines: **Tier 1** (identical to OpenAI) is
   `vllm_omni/protocol/realtime/**`; **Tier 2** (OpenAI names carrying our
   extensions) and **Tier 3** (ours alone) are `vllm_omni/protocol/duplex/**`.
   A Tier 2 class *subclasses* its Tier 1 twin and declares only what it adds,
   so the OpenAI surface stays honestly pure --- `session.created` without
   `resume_token`, `input_audio_buffer.append` without `video_frames`. Our
   error **codes** are Tier 3 (`protocol/duplex/errors.py`); only the envelope
   shape and OpenAI's three `error.type` classes are Tier 1.
   What stays engine-side is everything that is *internal representation*
   rather than contract: `DuplexCommand.payload()` renders the session runner's
   mailbox dictionary, whose channel genuinely differs from the client event
   (`session.update` and the three `conversation.item.*` commands all travel on
   `turn.signal`); `engine/duplex/realtime_commands.py` decides which command a
   decoded event becomes; `engine/duplex/realtime_events.py` holds the
   session's projection state. The codec may not import the engine, the
   entrypoints, `model_executor` or the clients
   (`tests/protocol/realtime/test_protocol_import_boundary.py`), and there is
   exactly one implementation of each codec behaviour
   (`tests/protocol/realtime/test_realtime_codec_single_source.py`).
   The dependency chain is `protocol/realtime` <- `protocol/duplex` <- engine /
   entrypoints / clients, and a duplex consumer uses **only** the middle link:
   `protocol/duplex/**` re-exports the Tier 1 names it does not extend, so a
   helper that later needs a duplex-specific version (`convert_input_audio_with_rate`
   resamples to MiniCPM-o's 16 kHz rather than the client's rate) is overridden
   in one file instead of at every call site
   (`tests/protocol/duplex/test_duplex_protocol_facade.py`).
   `DuplexEvent` is therefore a plain alias of `RealtimeEvent` --- an event has
   no engine-internal half --- while `DuplexCommand` is a real class, because a
   command does.

## Package layout

```text
vllm_omni/
├── entrypoints/
│   ├── omni_base.py                 OmniBase (+ abstract _create_engine)
│   ├── async_omni_base.py           AsyncOmniBase (output pump, _route_engine_message seam)
│   ├── async_omni.py                AsyncOmni (turn-based requests)
│   ├── duplex_omni.py               DuplexOmni, DuplexSessionHandle (thin pipe)
│   ├── duplex/                      SERVING (thin)
│   │   ├── serving.py               OmniDuplexSessionHandler
│   │   ├── realtime_input.py        RealtimeEnvelope (query rules, first message), parse_resume_request
│   │   ├── session_attachment.py    DuplexSessionAttachmentRegistry (resume tokens, replay journal)
│   │   ├── audio_encoding.py        encode_audio, injected into DuplexOmniEngine for the plugin's data plane
│   │   ├── chat_completions.py      DuplexChatCompletionsAdapter (/v1/chat/completions on a session per request)
│   │   └── websocket.py             websocket send/close/receive helpers
│   └── openai/api_server.py         builds DuplexOmni for duplex models; session-backed app state
├── protocol/                        SHARED WIRE CODEC (no engine / no model / no transport)
│   ├── realtime/                    the OpenAI Realtime API itself
│   │   ├── events.py                RealtimeEvent + OpenAI's 30 server events (22 ship as Tier 1;
│   │   │                            8 have a Tier 2 twin in protocol/duplex)
│   │   ├── commands.py              RealtimeCommand + OpenAI's 10 client commands (8 Tier 1,
│   │   │                            2 with a Tier 2 twin) --- wire_type + fields, no payload()
│   │   ├── formats.py               audio format spellings, what is supported, per-session validation
│   │   ├── session.py               session-object readers; RealtimeInputDefaults (append defaults)
│   │   ├── items.py                 conversation item shape, truncation, transcripts, camera frames
│   │   ├── audio_input.py           decode_audio_append -> RealtimeAudioAppend; client speech hints
│   │   ├── audio.py                 PCM / G.711 / WAV decode, resample, re-encode
│   │   ├── errors.py                RealtimeProtocolError (the envelope's exception type)
│   │   └── capabilities.py          RealtimeProtocolCapabilities: what one consumer can serve
│   └── duplex/                      Tier 2 + Tier 3, and the one door a duplex consumer uses
│       ├── __init__.py              re-exports the Tier 1 helper functions (audio, formats,
│       │                            items, session, capabilities) so the engine never
│       │                            imports protocol/realtime directly
│       ├── events.py                the whole 42-event vocabulary: 22 Tier-1 re-exports,
│       │                            8 Tier-2 subclasses (session.created + resume_token, ...),
│       │                            12 Tier-3 events (listen/speak, playback, resume, ...);
│       │                            DuplexEvent = RealtimeEvent
│       ├── commands.py              the whole 17-command vocabulary: 8 Tier-1 re-exports,
│       │                            2 Tier-2 subclasses (append + hints, commit + final),
│       │                            7 Tier-3 commands (barge_in, playback.ack, ...)
│       └── errors.py                our error-code vocabulary (Tier 3) + RealtimeProtocolError
├── engine/
│   ├── omni_engine_base.py          OmniEngineBase (stage processes, orchestrator thread, queues, RPC)
│   ├── async_omni_engine.py         AsyncOmniEngine (turn-based request building)
│   ├── duplex_omni_engine.py        DuplexOmniEngine (session message surface, creates DuplexOrchestrator)
│   ├── orchestrator.py              OrchestratorBase + Orchestrator
│   ├── duplex_orchestrator.py       DuplexOrchestrator (+ DuplexOrchestratorRequestState; implements DuplexStagePort)
│   └── duplex/
│       ├── commands.py              mailbox half: DuplexCommand.payload() + the mailbox `type`,
│       │                            paired with each protocol command; command_from_realtime
│       ├── realtime_commands.py     duplex binding of the codec: decoded event -> DuplexCommand,
│       │                            DUPLEX_REALTIME_CAPABILITIES, duplex_response_format
│       ├── events.py                re-export shim (DuplexEvent = RealtimeEvent) + the runner's
│       │                            epoch-filter sets DOMAIN_TERMINAL_EVENTS / MODEL_OUTPUT_EVENTS
│       ├── delivery.py              DuplexOutputBuffer (bounded output shared with the session handle)
│       ├── realtime_events.py       RealtimeProjectionState: internal event -> typed events
│       ├── messages.py              queue envelopes (Open/Close/Resume/Touch/Command/Result/Event), DuplexSessionError
│       ├── config.py                DuplexSessionConfig, DuplexCapabilities, ResponseCreateOptions
│       ├── contracts.py             DuplexFence (session_id, epoch, turn_id), stage request records, DuplexStagePort
│       ├── plugin.py                DuplexModelPlugin, DuplexModelSessionState, DuplexDataPlane, PcmAppendBuffer ABCs
│       ├── turn_detection.py        server-side VAD turn detector used by the session
│       ├── audio.py                 compatibility re-export, via protocol/duplex
│       ├── vad.py / intermediate.py
│       └── session/                 one engine-resident session and everything that runs it
│           ├── engine_session.py    DuplexEngineSession: ledgers, lease, fence, stage resources, append sequencing
│           ├── runner.py            DuplexSessionRunner (per-session mailbox on the orchestrator loop)
│           ├── manager.py           DuplexSessionManager (admission, backpressure, reaper, dispatch)
│           ├── context.py           DuplexSessionContext / DuplexRunState: what the runner shares with its components
│           ├── emitter.py           SessionEmitter: projection, epoch filter, domain effects of a terminal event
│           ├── model_channel.py     ModelChannel: submit an append, project stage output, continue a turn
│           ├── control.py           SessionControl: server VAD and the events that reconfigure it
│           ├── append_task.py       AppendAttempt: one append in flight and the rollback its failure owes
│           ├── helpers.py           pure reads and payload builders over a session
│           ├── lease.py             DuplexLeaseState (idle TTL, disconnect grace, resume generation)
│           └── overlap_policy.py / commit_policy.py / playback_ledger.py
├── config/stage_config.py           PipelineConfig.duplex_plugin; DuplexSessionRuntimeConfig
├── model_executor/models/minicpmo_4_5/duplex/plugin.py   MiniCPMO45DuplexPlugin (+ data_plane, input, policy, ...)
├── model_executor/models/nemotron_voicechat/duplex/plugin.py   NemotronVoiceChatDuplexPlugin (+ data_plane, input, capabilities)
└── clients/
    ├── duplex.py                    DuplexClientBase (ABC), DuplexClient (websocket), client-side events
    ├── inline_duplex.py             InlineDuplexClient (in-process, over DuplexOmni)
    ├── minicpmo_4_5.py              MiniCPM-o 4.5 session preset
    ├── nemotron_voicechat.py        Nemotron VoiceChat session preset
    └── personaplex.py               PersonaPlex session preset
```

## Session identity

- **The server allocates every session id.** `DuplexOmni.open_session`
  generates `duplex-<uuid4 hex>`; a `session_id` / `id` key inside the
  Realtime session object is ignored (Realtime clients echo the session object
  back). The allocated id reaches the client in `session.created` and is the
  only handle for `session.resume`, `close` and every command.
- **Ids are never reused** within an engine's lifetime, so the id alone
  identifies a session. There is no incarnation counter: stage request ids are
  `duplex-s.<b64 session id>.e.<epoch>.r.<role>`, model-side per-session state
  is keyed by `session_id` alone, and a command, close, touch or resume for an
  id the manager no longer holds is answered with `unknown_session`.
- **Staleness inside a live session is epoch-based.** `DuplexFence
  (session_id, epoch, turn_id)` is an engine-internal identity: every stage
  request is bound to the fence it was submitted under, cancels advance the
  epoch, and tracked tasks re-check the epoch after each `await`.
- **Resume identity** is `(session_id, resume_token, expected_lease_generation)`:
  the token belongs to the attachment registry, the lease generation is the
  engine-side compare-and-swap.

## Session lifecycle

```text
open   DuplexOmni.open_session(config)
         -> engine.open_session_async(session_id, DuplexSessionConfig)      [RPC]
         -> DuplexSessionManager.open: admission (max_sessions, closing sessions keep their slot),
            plugin.validate_client_extra_body / prepare_runtime_config, DuplexEngineSession,
            Stage0 request id reserved (ensure_stage_request), DuplexSessionRunner.start()
         -> first event: SessionCreated (announces the allocated id)
command  handle.submit(DuplexCommand)
         -> engine.submit_command_async -> DuplexSessionCommandMessage on the request queue [one-way]
         -> DuplexSessionManager.dispatch: unknown_session / input_backpressure checks, then runner mailbox
output   DuplexOrchestrator._intercept_stage_output -> runner.on_stage_output -> mailbox -> typed events
         -> session's bounded output buffer -> handle.events(); closure and errors without a live session
            use output_sink (DuplexSessionEventMessage) -> DuplexOmni._route_engine_message
detach   DuplexOmni.detach_session(expected_lease_generation) -> touch(DETACH): engine-owned disconnect
         grace, refused for a lease newer than the one the caller held; expiry -> SessionExpired
resume   DuplexOmni.resume_session(expected_lease_generation) -> lease CAS keyed by the control id (a replay
         answers with the generation it produced); the existing handle is re-entered. A caller cancelled
         mid-RPC observes the outcome afterwards (replaying until the engine answers) and settles a landed
         resume: the generation goes to the resume waiting to activate, else to the connection still
         attached, else the lease is detached again
close    DuplexOmni.close_session -> close RPC; the manager tears the runner down, then the stage cleanup
         (abort submitted requests, release reserved ids), then SessionClosed, then the RPC result.
         SessionClosed is emitted after the cleanup attempt (in a finally), so it is also sent when the
         stage cleanup failed; in that case the admission slot is intentionally retained and the cleanup
         is retried by the reaper, and opening a replacement session can still be refused with
         resource_exhausted. Seeing the event therefore does not by itself guarantee a free slot.
reap     DuplexSessionManager.reaper_loop: idle TTL / disconnect grace expiry, cleanup retries
```

### Concurrency and ordering model of the runner

The runner's session state is owned by the orchestrator asyncio loop and
needs no lock. The output buffer shared with the caller thread uses a lock
for queue updates and the final validity check before delivery.

- **Single writer per session.** `DuplexEngineSession` is mutated only by its
  runner. Inputs reach the runner through one mailbox in this order: client
  commands (from `_request_handler`, in request-queue order), stage outputs
  (`on_stage_output` enqueues an internal item; it does not mutate the session
  inline), timers (silence continuation, bounded auto-response) and internal
  items (deferred overlap promotion, VAD-synthesized commits). The mailbox
  worker processes items one at a time.
- **Slow work is not awaited inline.** An `AppendAudio` handler reserves
  bytes, then schedules a tracked append task on the per-session append tail:
  the task awaits `_offload(decode/resample/VAD)` on a dedicated
  `ThreadPoolExecutor` owned by the manager, then `plugin.plan_append`, then
  `stage_port.submit`. The worker returns to the mailbox immediately, so a
  later `CancelResponse` is not blocked behind an in-flight append.
- **Re-validation after every `await`.** A tracked task captures `(epoch,
  turn_id)` when it starts and re-checks them before the stage submit and
  before committing ledgers; a task that finds the epoch advanced rolls back
  its PCM reservation and exits.
- **Cancel is atomic at the session.** `CancelResponse` / `BargeIn` advance
  the epoch, cancel tracked append tasks, abort the owned stage bindings
  through the stage port and emit the cancelled terminal events before
  yielding. An append task already past its last check is harmless: its stage
  request is bound to the old epoch's request id, which the abort covers, and
  the base never emits session-owned outputs to a client.
- **Ordered event stream.** Because the epoch bump and every emission happen
  on one loop in program order, nothing can be emitted for an old epoch after
  its terminal event; `runner.emit()` still drops a terminal carrying a stale
  epoch and stamps `epoch` on every event for clients that filter
  defensively. In this experimental follow-up, public output goes directly to
  one bounded buffer shared by the engine and `DuplexSessionHandle`. Accepted
  cancellation removes queued `AudioDelta` events for that response through
  the cancelled epoch. Text increments, audio/transcript completion markers,
  other responses and normal completion retain FIFO order. The consumer also
  checks an event already taken from the buffer: WebSocket delivery repeats
  that check under the connection lock, before sequencing and journaling.
  Already sequenced replay entries are unchanged; already delivered audio
  still requires the client to stop its own playback.
- **Backpressure before the mailbox.** `DuplexSessionManager.dispatch`
  checks `max_pending_input_bytes_per_session` and reserves a pending turn for
  `Commit` before the put; a rejected command is answered with
  `ErrorEvent(code="input_backpressure")`.
- **Per-response stage metrics.** `_route_output` computes the per-segment
  `StageMetrics` before calling `_intercept_stage_output`; the runner
  accumulates the snapshot into the active response, so
  `ResponseDone.stage_metrics` and `metadata.vllm_omni.stage_metrics` on the
  wire keep their meaning.

### Experimental output limits

`max_pending_output_bytes_per_session` defaults to 2 MiB and
`max_pending_output_events_per_session` to 512. The byte limit counts compact
Realtime JSON, including base64 audio, not decoded PCM or Python object
overhead. Audio is already base64-encoded by the producer; counting its length
avoids serializing or decoding that large field again. Neither default has
been tuned from a performance study.

Errors and response endings have an additional 64 KiB / eight-event reserve
and retain FIFO order. One final session-closure notification has an independent
slot, so an exhausted reserve cannot prevent closure or break delivery to
other sessions. If that reserve also fills, an error or response-ending
notification may be omitted; the final session-closure notification still
has its independent slot. Oversized close details (over 4 KiB with generated identity)
are replaced by a compact reason, preserving terminal type and identity.
Late writes after closure are ignored; a locally ended iterator is not reopened
by a late terminal, and a failed open closes the abandoned handle. Events other
than errors are not forwarded once their session's buffer has been removed.
One event held by the consumer is outside the queued budget.

The overflow policy fails the active response without committing its partial
text to history and closes only the affected session. If overflow occurs while
a completion batch is being delivered, its not-yet-queued `ResponseDone` is
converted to failure with its identity and output preserved; any history entry
for that response is removed. An ending already accepted by the buffer is not
replaced or followed by a second ending. After overflow, newly
emitted ordinary events are suppressed, including text/audio completion,
content-part and output-item completion markers. Previously queued non-audio
events retain FIFO order; error and failed response-ending notifications may
use the reserve before the independent closure notification. This differs from
accepted cancellation, which removes only matching audio. See the
[public error and closure fields](../serving/realtime_duplex_api.md#output-limits-and-slow-consumers).
Response-only recovery is not implemented. These limits do not bound the runner's raw stage
output mailbox, replay journal, or client playback queues. The shared buffer
requires the current same-process, separate-thread engine arrangement; it is
not a cross-process delivery protocol. Direct handle iteration and WebSocket
delivery share buffer validity, but end-to-end client behavior still needs
independent real-model verification.

## Orchestrator seams

`OrchestratorBase` keeps the generic machinery (request handler skeleton,
abort, collective RPC, output loops, stage error and dead-replica handling,
`_route_output`, stage forwarding, prewarm, cleanup, shutdown) and offers
these seams. `DuplexOrchestrator` extends `Orchestrator` rather than sitting
beside it: a seam it does not override falls through to the turn-based
implementation, and `_dispatch_message` hands anything that is not a session
message to `super()`:

| Seam | `Orchestrator` | `DuplexOrchestrator` |
| --- | --- | --- |
| `_dispatch_message(msg) -> bool` | `add_request`, `streaming_update`, `add_companion_request`, `interaction` | open/command/close/resume/touch messages -> `DuplexSessionManager` |
| `_background_tasks()` | — | `[manager.reaper_loop()]` |
| `_shutdown_extensions()` | — | `await manager.shutdown()` |
| `_on_stage_submitted(stage_id, request_id, replica_id, req_state)` | — | bind the forwarded stage request to the owning session |
| `_intercept_stage_output(...) -> bool` | — | hand every session-owned output (Stage0 decision, Stage1 audio) plus its metric snapshot to `runner.on_stage_output`; `True` = do not forward |
| `_handle_forward_failure(...) -> bool` | — | fail the owning session's response and close the session, keep the loop alive |
| `_cleanup_request_ids(ids, abort=, release_owners=)` | ignores `release_owners` | brackets `super()` with `close_sessions_for_request_ids` / `defer_request_cleanups` / `finalize_closed_sessions` |
| `OrchestratorRequestState.session_owned` | never set | set by `ensure_request`: no synthetic terminal, no client emission, no terminal re-forward, no auto-cleanup on finish |

`DuplexOrchestrator` also implements the `DuplexStagePort` ABC the manager and
runner use (`stage_count`, `sampling_defaults`, `ensure_request`, `submit`,
`cleanup`, `abort_requests`): the resumable Stage0 request is preregistered at
open with `DuplexOrchestratorRequestState(session_owned=True, session_id,
fence)`, submitted with `submit_initial` on the first unit and
`submit_update` afterwards, and the request's `streaming.bridge_states["duplex"]`
carries the session identity (`session_id`, `fence`, `epoch`, `turn_id`,
`model_turn_id`), the public session config and the server-owned runtime
config for the worker-side model hooks.

## Model plugin

`DuplexModelPlugin` (ABC, `engine/duplex/plugin.py`) is the one class a model
provides, selected by `PipelineConfig.duplex_plugin`; "is a duplex model" is
`duplex_plugin is not None`. It merges the engine policy and the session
policy that used to be two separately configured objects:

| Half | Members |
| --- | --- |
| engine policy | `configure_sampling_params(runtime_config, defaults)`, `plan_append(...) -> DuplexAppendPlan` (the resumable Stage0 prompt for one unit), `decide_output(...) -> DuplexOutputDecision \| None` (e.g. the listen decision on a finished Stage0 segment) |
| session policy | `capabilities(max_sessions)`, `validate_client_extra_body`, `prepare_runtime_config(config, model_config)` (server-owned runtime keys, reference audio resolution), `runtime_config_for_update`, `runtime_config_for_function_output`, `runtime_config_after_model_output` (consumption acknowledgement), `create_session_state() -> DuplexModelSessionState`, `data_plane: DuplexDataPlane` (projects raw stage outputs into internal events), `data_plane_context(...)` |

`DuplexOmniEngine._validate_deployment` loads the plugin before any stage
starts; `DuplexSessionManager.__init__` validates it against the stage
sampling defaults once the stage pools exist. Plugin hooks that may block
(`prepare_runtime_config` fetching `ref_audio`) are awaited in `open()` and
offloaded from the loop.

Frame-based plugins use `engine/duplex/intermediate.py::build_duplex_append_prompt`
for the shared request identity, sequencing and config snapshots. Token budgets,
PCM framing and model-specific worker fields remain in the plugins.

See [supported models and deployments](../serving/full_duplex_api.md#enable-full-duplex)
for the current plugin integrations and deployment configurations.

## Serving

`OmniDuplexSessionHandler.handle_realtime_session(websocket)`:

1. accept; build the `RealtimeEnvelope` (autostart / default session from
   query params, `event_id` correlation, error rendering through the typed
   `ErrorEvent`);
2. handshake: first message `session.update` -> `omni.open_session(session)`
   (the id inside the payload is ignored; the allocated id is announced in
   `session.created` together with `resume_token` / `attachment_generation`
   when the model supports resume); `session.resume(session_id, resume_token,
   last_received_server_event_seq)` -> `attachment.authenticate_resume` ->
   `omni.resume_session`;
3. reader loop: JSON -> `RealtimeEnvelope.translate` (`translate_realtime_command`
   with the session's declared audio defaults) -> `handle.submit`;
   envelope-level errors (invalid JSON, oversize frame, unknown type,
   event acks) are answered locally;
4. writer pump (session-scoped, survives reconnects): `async for ev in
   handle.events(): attachment.send_event(ev.to_realtime())`, journaling
   for replay until the journal overflows (`session.resync_required`);
5. disconnect: a resumable session is detached (`attachment.detach` +
   `omni.detach_session`, engine-owned grace); a superseded socket's
   disconnect is a no-op; a non-resumable session is closed. Takeover sends
   `session.replaced` to the old socket and closes it.

## Validation

- CPU suites: `tests/engine/duplex/**` (session, lease, manager, runner,
  plugin, events, commands), `tests/engine/test_duplex_orchestrator.py`,
  `tests/engine/test_duplex_omni_engine.py`,
  `tests/entrypoints/test_duplex_omni.py`,
  `tests/entrypoints/duplex/test_duplex_serving.py`,
  `tests/entrypoints/openai/test_duplex_session_attachment.py`,
  `tests/entrypoints/openai_api/test_duplex_api_server.py` (the duplex
  server: pipeline probe, app state, routes, warmup gate),
  `tests/entrypoints/duplex/test_chat_completions_adapter.py`,
  `tests/clients/**`, `tests/engine/test_duplex_import_boundary.py`,
  `tests/model_executor/models/minicpmo_4_5/duplex/**`,
  `tests/worker/test_native_duplex_hooks.py`.
- GPU: `examples/online_serving/minicpmo/realtime_duplex_demo.py` (one turn
  over the websocket), `examples/online_serving/barge_in_client.py` (cancel
  during playback; `--inline` drives the same scenario through
  `InlineDuplexClient`), `tests/e2e/online_serving/run_minicpmo_realtime_duplex_multi_session.py`
  (admission, resume, takeover, expiry).