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; 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

              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:

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 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

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

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. 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 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).