nemotron_duplex: live calls at one-shot cost (full duplex 5/6)

configs/nemotron_duplex.yaml

@@ -9,11 +9,19 @@
 model: "nemotron_duplex"
 max_seq_len: 12288  # config: voicechat streaming_max_len / vllm_max_model_len
 
-# One recurrent slot per session (DEFAULT_MAMBA_SLOTS); past it the engine spins
-# on admit and wedges every session. Keep the two equal; a live call past the
-# cap gets a 503, and only one-shot requests queue.
+# Sessions at once; a live call past it gets a 503 (only one-shot requests
+# queue). Each session holds a slot in
+# resources mamba_state, encoder_state, encoder_front and encoder_kv and its
+# pages in nano_kv and talker_kv, all sized for 32 (DEFAULT_SESSIONS); to raise
+# the cap, raise their max_slots (cap + 1 for the recurrent pools, for the
+# sink; cap for encoder_kv) and the KV pools' max_num_pages with it.
+# The server refuses to start with a cap one of them cannot hold.
 max_concurrent_requests: 32
 
+# No per-message deflate on the WebSocket routes: compressing 32 live calls'
+# 80 ms appends cost the event loop ~8% on /generate/ws (H100, 2026-10-17).
+ws_per_message_deflate: false
+
 # A live call's input files over its whole life; past it the call ends with a
 # 400. 160 s of 16 kHz audio is about 10 MB as float32 WAV: 32 leaves room for
 # stereo or a 48 kHz opening clip. A file is decoded at most one frame past

configs/nemotron_duplex_disagg.yaml

@@ -14,11 +14,19 @@
 model: "nemotron_duplex"
 max_seq_len: 12288
 
-# One recurrent slot per session (DEFAULT_MAMBA_SLOTS); past it the engine spins
-# on admit and wedges every session. Keep the two equal; a live call past the
-# cap gets a 503, and only one-shot requests queue.
+# Sessions at once; a live call past it gets a 503 (only one-shot requests
+# queue). Each session holds a slot in
+# resources mamba_state, encoder_state, encoder_front and encoder_kv and its
+# pages in nano_kv and talker_kv, all sized for 32 (DEFAULT_SESSIONS); to raise
+# the cap, raise their max_slots (cap + 1 for the recurrent pools, for the
+# sink; cap for encoder_kv) and the KV pools' max_num_pages with it.
+# The server refuses to start with a cap one of them cannot hold.
 max_concurrent_requests: 32
 
+# No per-message deflate on the WebSocket routes: compressing 32 live calls'
+# 80 ms appends cost the event loop ~8% on /generate/ws (H100, 2026-10-17).
+ws_per_message_deflate: false
+
 # A live call's input files over its whole life; past it the call ends with a
 # 400. 160 s of 16 kHz audio is about 10 MB as float32 WAV: 32 leaves room for
 # stereo or a 48 kHz opening clip. A file is decoded at most one frame past

docs/adding_models.rst

@@ -569,9 +569,12 @@ The spec types are:
        ``SamplingReqConfig`` above.
    * - ``RecurrentStateSpec(config=RecurrentStateConfig(...))``
      - Fixed-size per-request state that carries between steps (an LSTM's ``h`` and
-       ``c``, a cursor, a conv tail): one slot per request, ``blocks`` naming a shape and
-       dtype for each piece, ``num_layers`` copies of each, ``max_slots`` slots with slot 0
-       the sink that padding rows address. ``declare_step`` gives each row a
+       ``c``, a cursor, a conv tail, a Mamba or gated-DeltaNet layer's conv window and SSM
+       state, a streaming encoder's caches): one slot per request and label, ``blocks``
+       naming a shape and dtype for each piece (dtype None: the engine's compute dtype,
+       what its KV caches store), ``num_layers`` copies of each, ``max_slots`` slots
+       (deployment-tunable) with slot 0 the sink that padding rows address. A recurrent
+       kernel reads the step's ``addressing``. ``declare_step`` gives each row a
        ``RecurrentStep`` segment with span >= 1. The forward reads its rows with
        ``pool.gather(name, start=...)`` and writes them back with
        ``pool.scatter_(name, value)``; both follow the step's plan, so there is no
@@ -866,7 +869,9 @@ within itself and matches the offline mask. ``rel_bias`` is then indexed by that
 ``head_dim ** -0.5``). A per-query window takes no sink, and its right context is 0.
 ``retention`` may be a single ``SinkWindow`` when it does not depend on the source length.
 ``BoundedKVConfig(dtype=None)`` stores the cache in the engine's KV dtype, for layers that
-compute in it.
+compute in it. Nemotron VoiceChat's streaming encoder (``ENCODER_KV``,
+``encoder_kv_config`` in ``mstar/model/nemotron_duplex/components/conformer.py``) is the
+per-query example.
 
 Step 3 — Declare the computation graph
 --------------------------------------

docs/clients.rst

@@ -98,6 +98,9 @@ WebSocket ``/generate/ws``
 
 ``/generate/ws`` is the same request over one persistent WebSocket, for control loops
 that cannot afford an HTTP round trip per step (robot policies, streaming world models).
+The server negotiates per-message deflate on its WebSocket routes unless the deployment
+turns it off (``ws_per_message_deflate`` in :doc:`serving`, as the Nemotron VoiceChat
+configs do).
 Each message is one request with the ``/generate`` fields — ``text``, ``files`` as
 ``[{"name": ..., "data": ...}]``, ``input_modalities``, ``output_modalities``,
 ``model_kwargs``, ``request_id`` — sent either as a JSON text frame (``data`` base64) or
@@ -189,6 +192,12 @@ is at ``max_concurrent_requests`` ends at once with ``status`` 503, rather than
 slot and start that far behind its input; a one-shot request waits in the queue. Other
 models ignore ``end_input``, and an append to them is rejected.
 Live input needs the Python frontend: ``--rust-frontend`` serves no ``/generate/ws``.
+The append size that serves best is the model's: Nemotron VoiceChat runs a live step as one
+CUDA graph only for an append of exactly 80 ms (1280 samples at 16 kHz) onto a call whose
+audio so far is a whole number of 80 ms frames. Keep the opening clip and every append a
+multiple of 1280 samples: one that is not (a one-second opening clip, 16000 samples) keeps
+the call off the graph for the rest of its life. Other sizes are served the same, more
+slowly (see :doc:`models`).
 
 .. code-block:: python
 

docs/models.rst

@@ -597,18 +597,31 @@ Nemotron VoiceChat (``nemotron_duplex``) notes
   PCM: each 5-frame chunk is decoded with the previous 3 frames as left context,
   which covers the decoder's receptive field, and a window's last 12 samples,
   the inverse-STFT overlap, are held back and emitted with the next chunk, so
-  the stream equals the full-history decode sample for sample).
+  the stream equals the full-history decode sample for sample; that steady
+  window of 8 frames replays a CUDA graph, other windows decode eagerly).
   The decode loops are stream-terminated: they end when the upstream stream
   closes, not on EOS, which is an ordinary per-frame token in duplex speech.
 - ``nano_llm`` declares a paged KV cache over its four attention layers with
   the attention plan on it (``resources: nano_kv`` / ``nano_attn``; no
   positional encoding, so no position resource), a recurrent-state pool holding
   the 27 Mamba-2 layers' conv window and SSM state, one slot per session
   (``mamba_state``: 32 sessions by default, ~137 MB per slot in fp32, tunable
-  with ``max_slots`` / ``state_dtype``; the shipped configs set
-  ``max_concurrent_requests: 32`` to match, so raise the two together), the Mamba-2 resource planned on it
+  with ``max_slots``), the Mamba-2 resource planned on it
   (``mamba``) and the agent-text sampler (``nano_sampler``). Its decode step is
   captured as a CUDA graph for batch sizes 1 to 32.
+- Every per-session resource is sized for 32 sessions: a slot in each of
+  ``mamba_state``, ``encoder_state`` and ``encoder_front`` (``max_slots``, one
+  more than the sessions, for the sink) and in ``encoder_kv`` (``max_slots``,
+  no sink), and its pages in ``nano_kv`` and
+  ``talker_kv`` (``max_num_pages``; both sized from ``max_session_s``, and the
+  nano's from ``max_prompt_tokens`` too). The shipped configs set
+  ``max_concurrent_requests: 32`` to match; to serve more, raise the cap and
+  all six with it. The server
+  refuses to start with a cap one of them cannot hold, since a session past
+  any of them would wait on admit, and its frames with it, until another ends.
+  The nano, the talker, the encoder and the codec each run at most 32 sessions
+  a step, their widest CUDA graph; more are split over several steps rather
+  than run eagerly.
 - ``eartts_talker`` advances every live session in one backbone pass per frame
   through the engine's paged KV (``talker_kv`` / ``talker_attn`` /
   ``talker_pos``): each session owns two streams, the conditional and the
@@ -634,8 +647,12 @@ Nemotron VoiceChat (``nemotron_duplex``) notes
   fast as it can). Live calls paced in real time over ``/generate/ws``
   (``test/nemotron_duplex/live_ws_request.py``, 80 ms appends, 2026-10-17, one
   H100): input-to-token lag p50 48 ms with one call and about 100-120 ms with
-  8, which keep up; from 24 calls up the server falls behind real time, so lag
-  grows over the call. Served audio is checked for intelligibility by
+  8, which keep up; from 24 calls up the server fell behind real time, so lag
+  grew over the call. Capturing the encoder's live step (below) took its host
+  time at 32 calls from 23.6 to 8.4 ms and cut live lag 2-6x at 16-32 calls
+  (paced runs interleaved against the uncaptured encoder, ``MSTAR_SHM_ARENA=1``);
+  32 calls staggered over 2 s all finish, heard latency p50 2.1-2.3 s (from the
+  user stopping to agent audio playable without gaps). Served audio is checked for intelligibility by
   transcribing it (``test/nemotron_duplex/asr_check.py``, Whisper
   large-v3-turbo): MaskGIT sampling is knife-edge, so attention-backend
   numerics change the waveform but not the words.
@@ -682,6 +699,21 @@ Nemotron VoiceChat (``nemotron_duplex``) notes
   (inferred from those frames, not measured on served text); under the served
   bf16 a different chunking can flip a knife-edge frame such as the agent's
   turn start.
+- The encoder keeps each call's state in a slot per session of three
+  resources: its attention keys and values in a bounded KV (``encoder_kv``),
+  which keeps a call's last 70 frames per layer and runs the attention, each new
+  frame over the 70 before it, with the relative-position term as a band of
+  distances; its conv inputs (``encoder_state``); and its newest audio and mel
+  frames (``encoder_front``), the last two recurrent-state pools. A
+  call that appends one frame of audio (1280 samples) on the frame grid, every
+  live step of an 80 ms client, runs its whole step as a CUDA graph (the grid is
+  the call's start: an opening clip or append that is not a multiple of 1280
+  samples keeps the call off it for good); other calls take the front end
+  eagerly, then the layer stack as a graph when each got at most one frame (a
+  live step, the call's end), or eagerly (a one-shot upload, a large append:
+  a chunk over a second is not batched with other calls, so the engine runs
+  each call of that step on its own). Appends of other sizes are served the same,
+  without the whole-step graph.
 - Checks: ``test/nemotron_duplex/duplex_request.py`` prepares a clip from the
   bundled ``turn_taking.wav`` and sends it; ``parity_compare.py`` runs the
   standalone reference path on the same clip and checks the served text

docs/serving.rst

@@ -273,6 +273,14 @@ A config maps the model's computation-graph nodes to physical GPU ranks. The key
        ``max_call_input_mb`` it counts what the model was handed, not bytes uploaded.
    * - ``model_kwargs``
      - *(optional)* Server-init model parameters (see below).
+   * - ``ws_per_message_deflate``
+     - *(optional, default true)* Whether the server offers per-message deflate on its
+       WebSocket routes (``/generate/ws``, ``/v1/realtime``); it applies to every route.
+       Deflate shrinks 80 ms speech frames to 49-73% of their size as base64 JSON and to
+       55-90% as binary PCM, and digital silence to 1-2%, but compressing every frame
+       costs the event loop: about 8% of it at 32 live Nemotron VoiceChat calls on
+       ``/generate/ws``, whose configs turn it off. Turn it off where the server's
+       event loop, not the network, is the bottleneck.
    * - ``speculation_mixes_walks``
      - *(optional)* Per node, ``true``/``false`` for every walk of the node's combined
        walks, or a mapping of walk -> ``true``/``false``. On for a walk, a speculation

mstar/api_server/entrypoint.py

@@ -2382,6 +2382,16 @@ async def shutdown_event():
 # CLI entry point
 # ------------------------------------------------------------------
 
+def _uvicorn_config(app, host: str, port: int, config: dict) -> uvicorn.Config:
+    # Per-message deflate is negotiated server-wide, for every WebSocket route,
+    # so it is the deployment's to turn off (see ws_per_message_deflate in
+    # docs/serving.rst); on by default, as uvicorn's.
+    return uvicorn.Config(
+        app, host=host, port=port, access_log=False,
+        ws_per_message_deflate=bool(config.get("ws_per_message_deflate", True)),
+    )
+
+
 def _set_preprocess_threads(model, config: dict) -> None:
     """Size torch's intra-op pool for this process's media preprocessing.
 
@@ -2593,9 +2603,7 @@ def main(argv: list[str] | None = None):
             logger.info("Starting mstar API server on %s:%s", args.host, args.port)
             # uvicorn.run() inlined so the server object is reachable. The
             # message thread stops it when the conductor dies.
-            server = uvicorn.Server(
-                uvicorn.Config(app, host=args.host, port=args.port, access_log=False)
-            )
+            server = uvicorn.Server(_uvicorn_config(app, args.host, args.port, config))
 
             def _stop_server():
                 server.should_exit = True

mstar/engine/resources/linear_attn/base.py

@@ -40,7 +40,7 @@ def build(cls, spec: LinearAttnSpec, info: EngineResourceInfo):
                 config=spec.config,
                 geometry=Mamba2Geometry.from_blocks(pool_config.blocks),
                 num_layers=pool_config.num_layers,
-                state_dtype=pool_config.blocks["ssm"].dtype,
+                state_dtype=pool_config.blocks["ssm"].resolved_dtype(info.kv_dtype),
                 has_sink=not pool_config.disable_sink_slot,
                 device=info.device,
             )
@@ -55,7 +55,7 @@ def build(cls, spec: LinearAttnSpec, info: EngineResourceInfo):
                 max_rows_in_combined=info.max_rows_in_combined,
                 geometry=geometry,
                 num_layers=pool_config.num_layers,
-                state_dtype=pool_config.blocks["state"].dtype,
+                state_dtype=pool_config.blocks["state"].resolved_dtype(info.kv_dtype),
                 has_sink=not pool_config.disable_sink_slot,
                 device=info.device,
             )

mstar/engine/resources/recurrent/config.py

@@ -7,7 +7,6 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field, replace
-from math import prod
 from typing import TYPE_CHECKING
 
 import torch
@@ -24,17 +23,23 @@ class RecurrentBlockConfig:
     """One per-slot, per-layer tensor block.
 
     ``shape`` is opaque to the pool. ``shard_dims`` names the axes divided
-    across ranks.
+    across ranks. A ``dtype`` of None is the engine's compute dtype (what its
+    KV caches store), resolved by ``resolved_dtype`` and never written back.
     """
 
     shape: tuple[int, ...]
-    dtype: torch.dtype
+    dtype: torch.dtype | None
     shard_dims: tuple[int, ...] = ()
 
     def __post_init__(self):
         self.shape = tuple(self.shape)
         self._unsharded_shape = self.shape
 
+    def resolved_dtype(self, kv_dtype: torch.dtype | None) -> torch.dtype:
+        """The dtype the pool stores this block in; a resource planned against
+        the pool reads it here rather than from ``dtype``."""
+        return self.dtype or kv_dtype or torch.get_default_dtype()
+
     def shard(self, num_shards: int) -> None:
         from mstar.distributed.utils import divide
 
@@ -43,14 +48,6 @@ def shard(self, num_shards: int) -> None:
             shape[dim] = divide(self._unsharded_shape[dim], num_shards)
         self.shape = tuple(shape)
 
-    @property
-    def numel(self) -> int:
-        return prod(self.shape)
-
-    @property
-    def nbytes(self) -> int:
-        return self.numel * torch.empty((), dtype=self.dtype).element_size()
-
 
 class RecurrentGeometry(ABC):
     @abstractmethod
@@ -91,7 +88,7 @@ def conv_dim(self) -> int:
 
     def to_blocks(
         self,
-        state_dtype: torch.dtype = torch.float32,
+        state_dtype: torch.dtype | None = torch.float32,
         conv_dtype: torch.dtype = torch.bfloat16,
     ) -> dict[str, RecurrentBlockConfig]:
         """Pool blocks for this geometry; head counts are pre-sharding."""
@@ -183,7 +180,7 @@ def conv_dim(self) -> int:
 
     def to_blocks(
         self,
-        state_dtype: torch.dtype = torch.float32,
+        state_dtype: torch.dtype | None = torch.float32,
         conv_dtype: torch.dtype = torch.bfloat16,
     ) -> dict[str, RecurrentBlockConfig]:
         """Pool blocks for this geometry; head counts pre-sharding (heads and
@@ -276,14 +273,6 @@ def shard(self, num_shards: int) -> None:
         for block in self.blocks.values():
             block.shard(num_shards)
 
-    @property
-    def slot_bytes(self) -> int:
-        return self.num_layers * sum(b.nbytes for b in self.blocks.values())
-
-    @property
-    def total_bytes(self) -> int:
-        return self.slot_bytes * self.max_slots
-
 
 @dataclass
 class RecurrentStateSpec(NodeResourceSpec):

mstar/engine/resources/recurrent/pool.py

@@ -76,9 +76,14 @@ def build(
         config = spec.config
         if info.joint_comm_group is not None:
             config.shard(info.joint_comm_group.world_size)
-        return cls(device=info.device, config=config, name=spec.resource_key)
-
-    def __init__(self, device: torch.device, config: RecurrentStateConfig, name: str = "recurrent_state"):
+        # resolved here, not written back: the config is the model's, shared by its specs
+        return cls(device=info.device, config=config, name=spec.resource_key, kv_dtype=info.kv_dtype)
+
+    def __init__(
+        self, device: torch.device, config: RecurrentStateConfig, name: str = "recurrent_state",
+        kv_dtype: torch.dtype | None = None,
+    ):
+        """``kv_dtype`` (the engine's) stands in for a block's ``dtype=None``."""
         self._device = device
         self.config = config
         # the resource key, for errors
@@ -90,7 +95,7 @@ def __init__(self, device: torch.device, config: RecurrentStateConfig, name: str
         self._blocks: dict[str, torch.Tensor] = {
             name: torch.zeros(
                 (config.num_layers, config.max_slots, *block.shape),
-                dtype=block.dtype,
+                dtype=block.resolved_dtype(kv_dtype),
                 device=device,
             )
             for name, block in config.blocks.items()
@@ -127,12 +132,14 @@ def __init__(self, device: torch.device, config: RecurrentStateConfig, name: str
         self._preplanned = False
         self._cached_plan_output: dict[str, RecurrentAddressing] | None = None
 
+        # from the blocks, not the config: a dtype=None block is sized as resolved
+        total_bytes = sum(t.nbytes for t in self._blocks.values())
         logger.info(
             "recurrent state pool: %d usable slots (+%d sink) x %d layers, "
             "%.2f MiB (%.2f KiB/slot), blocks=%s",
             config.usable_slots, config.max_slots - config.usable_slots,
             config.num_layers,
-            config.total_bytes / 2**20, config.slot_bytes / 2**10,
+            total_bytes / 2**20, total_bytes / config.max_slots / 2**10,
             {n: tuple(b.shape) for n, b in config.blocks.items()},
         )