[Core] Make protocol/duplex the single duplex vocabulary; drop the engine copies (follow-up to #7640)

docs/design/fullduplex.md

@@ -46,7 +46,7 @@ the model plugins.
                    │                          ┌──────────────────────────────────────┐
                    │                          │ OmniDuplexSessionHandler (thin)      │
                    │                          │  websocket I/O, command_from_realtime│
-                   │                          │  event.to_realtime(), attachment /   │
+                   │                          │  event.to_wire(), attachment /   │
                    │                          │  resume tokens / replay journal      │
                    │                          └──────────────────┬───────────────────┘
                    ▼                                             ▼   DuplexSessionHandle
@@ -104,9 +104,9 @@ plugin and the session runtime config to `DuplexOrchestrator` directly.
    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
+   dataclasses (`protocol/duplex/commands.py`); outputs are `DuplexEvent`
+   dataclasses (`protocol/duplex/events.py`). `command_from_realtime()` and
+   `DuplexEvent.to_wire()` 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.**
@@ -143,13 +143,17 @@ plugin and the session runtime config to `DuplexOrchestrator` directly.
    `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
+   `protocol/duplex` is the whole duplex vocabulary: it re-exports Tier 1,
+   declares Tier 2 and Tier 3, and decodes a client event
+   (`decode_duplex_command`); `DuplexCommand` / `DuplexEvent` are aliases of
+   `RealtimeCommand` / `RealtimeEvent`. What stays engine-side is everything
+   that is *internal representation* rather than contract:
+   `engine/duplex/mailbox.py` binds the decoder to what the engine can serve
+   (`DUPLEX_REALTIME_CAPABILITIES`, `command_from_realtime`) and renders the
+   session runner's mailbox dictionary (`mailbox_payload`), whose channel
+   genuinely differs from the client event (`session.update` and the three
+   `conversation.item.*` commands all travel on `turn.signal`);
+   `engine/duplex/projection.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
@@ -214,14 +218,12 @@ vllm_omni/
 │   ├── 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
+│       ├── mailbox.py               engine binding of the protocol: DUPLEX_REALTIME_CAPABILITIES,
+│       │                            command_from_realtime, mailbox_payload (typed command -> the
+│       │                            runner's mailbox dictionary and its channel)
+│       ├── projection.py            RealtimeProjectionState: internal event -> typed events,
+│       │                            resolve_* (typed command -> mailbox payloads + 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
@@ -464,12 +466,12 @@ for the current plugin integrations and deployment configurations.
    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`
+3. reader loop: JSON -> `RealtimeEnvelope.translate` (`command_from_realtime`
    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
+   handle.events(): attachment.send_event(ev.to_wire())`, 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

docs/design/module/entrypoints.md

@@ -151,7 +151,7 @@ but never the reverse, and a Tier 1 class MUST NOT carry a vLLM-Omni extension
 field; and each protocol/codec behavior MUST have exactly one
 implementation that every consumer uses. A wire type MUST NOT carry a
 consumer's internal representation: the duplex mailbox rendering
-(`DuplexCommand.payload()`, whose channel differs from the client event for
+(`mailbox_payload()` in `engine/duplex/mailbox.py`, whose channel differs from the client event for
 `session.update` and the `conversation.item.*` commands) stays engine-side.
 A duplex consumer (`engine/duplex/**`, `entrypoints/duplex/**`, the duplex
 clients) MUST import `vllm_omni.protocol.duplex` and MUST NOT import

docs/serving/full_duplex_api.md

@@ -150,7 +150,7 @@ the [Realtime Duplex API guide](realtime_duplex_api.md#capability-negotiation-by
 `vllm_omni.entrypoints.duplex_omni.DuplexOmni` runs the same engine-resident
 sessions in-process: `open_session()` returns a `DuplexSessionHandle` whose
 `submit()` takes typed `DuplexCommand` objects and whose `events()` yields
-typed `DuplexEvent` objects (each with a `to_realtime()` wire rendering).
+typed `DuplexEvent` objects (each with a `to_wire()` wire rendering).
 `vllm_omni.clients.inline_duplex.InlineDuplexClient` exposes that handle
 behind the `DuplexClient` API. See the
 [Realtime Duplex API guide](realtime_duplex_api.md#using-the-python-api).

docs/serving/realtime_duplex_api.md

@@ -320,8 +320,8 @@ resumes and closes them and pipes typed commands in and typed events out
 through a `DuplexSessionHandle`.
 
 ```python
-from vllm_omni.engine.duplex.commands import AppendAudio, Commit
-from vllm_omni.engine.duplex.events import AudioDelta, SessionClosed
+from vllm_omni.protocol.duplex.commands import AppendAudio, Commit
+from vllm_omni.protocol.duplex.events import AudioDelta, ResponseDone, SessionClosed
 from vllm_omni.entrypoints.duplex_omni import DuplexOmni
 
 omni = DuplexOmni(model="openbmb/MiniCPM-o-4_5", trust_remote_code=True)
@@ -347,7 +347,7 @@ convenience wrapper per command (`append_audio`, `commit`, `create_response`,
 `delete_item`, `truncate_item`, `heartbeat`, ...). `events()` is a
 single-consumer async iterator that ends after `session.closed` /
 `session.expired`; rejected commands come back as `ErrorEvent`s on it.
-Every event renders the wire JSON with `to_realtime()`, so anything written
+Every event renders the wire JSON with `to_wire()`, so anything written
 against the WebSocket protocol works unchanged on the typed stream.
 If opening fails, the abandoned handle is closed. Late engine events do not
 reopen a locally closed handle.
@@ -502,7 +502,7 @@ Semantic divergences hidden behind shared names:
 | Turn-taking | `response.speak`, `response.listen`, `overlap.decision`, `overlap_policy`, `barge_in`, `turn.signal`, `input.cancel`, `input.text.append`, `epoch` / `turn_id` fencing, `force_listen` |
 | Playback truth | `playback.ack` / `playback.acknowledged`, `playback_commit_policy` (`ack_only` \| `commit_all_on_done`), `playback{generated_ms, sent_ms, played_ms, committed_ms}`, `history_committed`, `audio_text_marks` |
 | Model negotiation | `capabilities{implementation_level, supports_input_append, supports_barge_in, supports_session_resume, chunk_period_ms, input_modes, …}`, `ref_audio`, `extra_body` (`auto_response`, `force_listen_count`, `duplex_initial_user_text`) |
-| Diagnostics | `duplex.function_call.done`, the full error-code vocabulary (`REALTIME_ERROR_TYPES_BY_CODE` in `vllm_omni/engine/duplex/events.py`) |
+| Diagnostics | `duplex.function_call.done`, the full error-code vocabulary (`REALTIME_ERROR_TYPES_BY_CODE` in `vllm_omni/protocol/duplex/errors.py`) |
 
 #### Consequences for clients
 
@@ -569,8 +569,8 @@ Every event type the Realtime route accepts or emits, with a worked example in t
 client→server and 42 server→client. Aliases (`input_text.append`,
 `push_text`, `signal_turn`, `audio.playback_ack`, `close_session`, `close`)
 share the payload of their canonical event and are not listed
-separately; the alias table is `translate_realtime_command` in
-`vllm_omni/engine/duplex/realtime_commands.py`. The pre-Realtime WAV-append
+separately; the alias table is `decode_duplex_command` in
+`vllm_omni/protocol/duplex/commands.py`. The pre-Realtime WAV-append
 aliases `push_chunk` and `input.audio.append` are **not** accepted on this
 route: send `input_audio_buffer.append` with one of the supported input
 formats (`pcm16`, `pcm_s16le`, `s16le`, `pcm_f32le`, `g711_ulaw`,
@@ -1348,20 +1348,20 @@ flat native shape:
 
 The examples above cover every event type the Realtime route emits or
 accepts (verified by diffing the `type` literals in
-`vllm_omni/engine/duplex/events.py` and `vllm_omni/engine/duplex/commands.py`
+`vllm_omni/protocol/duplex/events.py` and `vllm_omni/protocol/duplex/commands.py`
 against this document). Not shown, because they never reach a client
 unprojected: the session-internal events the runner produces
 (`input.committed`, `input.cancelled`, `audio.cancelled`,
 `response.output_audio.delta`, `response.text.delta`, `function_call.done`)
 — their Realtime projections are the examples above (see the name map at the
 end of this section). Every public event is a typed `DuplexEvent` whose
-`to_realtime()` is the wire object; every client event is parsed into a
+`to_wire()` is the wire object; every client event is parsed into a
 typed `DuplexCommand` by `command_from_realtime()`.
 
 #### Internal to Realtime name map
 
 The session runner emits these session-internal events; the Realtime
-projector (`vllm_omni/engine/duplex/realtime_events.py`) renames and fans
+projector (`vllm_omni/engine/duplex/projection.py`) renames and fans
 them out into typed events before they reach a client.
 
 | Internal event | Realtime projection |

tests/clients/test_duplex_client.py

@@ -46,9 +46,9 @@
     write_pcm16_wav,
 )
 from vllm_omni.clients.inline_duplex import InlineDuplexClient
-from vllm_omni.engine.duplex import commands as duplex_commands
-from vllm_omni.engine.duplex import events as duplex_events
 from vllm_omni.engine.duplex.messages import DuplexSessionError
+from vllm_omni.protocol.duplex import commands as duplex_commands
+from vllm_omni.protocol.duplex import events as duplex_events
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
@@ -125,7 +125,7 @@ def __init__(self, raw: dict[str, object]) -> None:
     def is_terminal(self) -> bool:
         return self.raw.get("type") in {"session.closed", "session.expired"}
 
-    def to_realtime(self) -> dict[str, object]:
+    def to_wire(self) -> dict[str, object]:
         return dict(self.raw)
 
 

tests/engine/duplex/test_delivery.py

@@ -8,8 +8,8 @@
 import pytest
 
 from vllm_omni.engine.duplex.delivery import DuplexOutputBuffer, DuplexOutputOverflowError
-from vllm_omni.engine.duplex.events import AudioDelta, ResponseDone, SessionClosed, TranscriptDelta
-from vllm_omni.engine.duplex.realtime_events import RealtimeProjectionState, project_internal_event
+from vllm_omni.engine.duplex.projection import RealtimeProjectionState, project_internal_event
+from vllm_omni.protocol.duplex.events import AudioDelta, ResponseDone, SessionClosed, TranscriptDelta
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
@@ -25,7 +25,7 @@
 async def test_queued_response_creation_keeps_payload_and_byte_count(event_type, delta_field, content_field):
     state = RealtimeProjectionState(session_id="s")
     events = project_internal_event(state, {"type": "response.created", "response_id": "r", "modalities": ["text"]})
-    payloads = [event.to_realtime() for event in events]
+    payloads = [event.to_wire() for event in events]
     byte_count = sum(
         len(json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode()) for payload in payloads
     )
@@ -40,7 +40,7 @@ async def test_queued_response_creation_keeps_payload_and_byte_count(event_type,
     assert item["content"][0][content_field] == text
     assert output.pending_bytes == byte_count
     delivered = [await output.get() for _ in events]
-    assert [event.to_realtime() for event in delivered] == payloads
+    assert [event.to_wire() for event in delivered] == payloads
     assert output.pending_events == output.pending_bytes == 0
 
 

tests/engine/duplex/test_duplex_audio.py

@@ -6,7 +6,7 @@
 import numpy as np
 import pytest
 
-from vllm_omni.engine.duplex.audio import convert_input_audio_with_rate
+from vllm_omni.protocol.duplex import convert_input_audio_with_rate
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]