feat: bidirectional streaming-input WebSocket frontend

[GuanLuo]

Overview:

First slice (1/N) of the bidirectional streaming-input feature (parent design
"Streaming Request Support"): a WebSocket frontend at /v1/realtime that
accepts NvCreateChatCompletionRequest JSON frames over a single connection
and streams NvCreateChatCompletionStreamResponse chunks back. The endpoint
is wired to a process-scoped mock EchoBidirectionalEngine so the frontend
slice can ship before the engine-side ModelManager refactor that #9174 (2/N)
delivers.

Reusing chat-completion request/response types is intentional for the PoC —
the request can carry audio data and the response can carry text data, which
is what realtime workloads need. A dedicated realtime-API protocol type will
replace the chat overload in a follow-up (#9175).

Details:

Frontend (new surface):

• lib/llm/src/http/service/realtime.rs — /v1/realtime WebSocket handler.
  Splits the socket, pumps Message::Text into a bounded mpsc (channel(64)
  for backpressure), wraps it as a RequestStream, calls
  engine.generate(input), forwards the Annotated response stream as Text
  frames, rejects Binary frames. Inbound writes a non-NORMAL close Message
  into a shared slot (Arc<parking_lot::Mutex<Option<Message>>>) on protocol
  errors (INVALID for malformed JSON, UNSUPPORTED for binary) before
  cancelling the engine via stop_generating(); outbound takes from that
  slot after the response stream ends. The outbound task drives the sink to
  completion (SinkExt::close) under a 5s timeout so a half-broken peer
  can't park the connection indefinitely.
• lib/llm/tests/http_websocket.rs — three integration tests: (1)
  sequential request echo with finish_reason=stop per request, (2)
  explicit server Close frame after client-initiated close, (3)
  binary-frame rejection.
• lib/llm/src/engines.rs — EchoBidirectionalEngine: per-request char-echo
  followed by a terminal FinishReason::Stop, processed sequentially.
• lib/llm/src/types.rs — new pub mod generic::realtime with
  RealtimeBidirectionalEngine = BidirectionalStreamingEngine<NvCreateChatCompletionRequest, Annotated<NvCreateChatCompletionStreamResponse>>.

Runtime-crate fix (originally out of scope):

dynamo_runtime::pipeline::ManyIn<T> = Context<DataStream<T>> was
uninstantiable: DataStream<T> is !Sync (Pin<Box<dyn Stream + Send>>)
while Context<T: Data> requires Send + Sync. The fix:

• Relaxed AsyncEngine<Req>'s Sync bound to Send + 'static. Workspace-
  wide grep confirmed no implementation relied on Req: Sync. The trait-
  object form's Send + Sync properties come from its supertrait bounds,
  not from Req.
• Rebased ManyIn<T> onto the existing EngineStream<T> alias (the same
  alias ManyOut<T> already used). Input and output sides resolve to the
  same trait-object now.
• Added RequestStream<R> as a type alias of ResponseStream<R> — one
  struct backs both directions; the alias is purely documentary at use
  sites.
• Side effect: the previously-dead ClientStreamingEngine and
  BidirectionalStreamingEngine aliases in pipeline.rs are now actually
  instantiable. feat: ModelManager accessor for bidirectional engines (2/N) #9174 (2/N) can use them directly.

Wiring:

• Cargo.toml — + "ws" to axum's workspace feature list.
• lib/llm/Cargo.toml — tokio-tungstenite = "0.26" as dev-dep (matches
  axum's bundled tungstenite version, so only one version in Cargo.lock).
• lib/llm/src/http/service.rs — pub mod realtime;.
• lib/llm/src/http/service/service_v2.rs — realtime_router() merged
  into the inference router (unconditional registration; the engine slot
  is opt-in via OnceLock).

Where should the reviewer start?

Reading order, smallest mental model first:

1. lib/llm/tests/http_websocket.rs — what does this PR enable? The three
   tests show wire shape, client-close cleanup, and binary-frame rejection.
2. lib/runtime/src/engine.rs + lib/runtime/src/pipeline.rs — the
   runtime-crate fix. Read RequestStream<R> and ResponseStream<R> side
   by side to see how minimally they differ.
3. lib/llm/src/types.rs — RealtimeBidirectionalEngine typedef. Note
   placement under generic::realtime rather than openai::chat_completions:
   the bidirectional path isn't a chat-completions concept; it just reuses
   the schema for now.
4. lib/llm/src/engines.rs — EchoBidirectionalEngine. Pins the engine
   contract the handler has to satisfy.
5. lib/llm/src/http/service/realtime.rs — the bridge. Now you know the
   I/O types and engine signature; the handler reads as plumbing.

To verify locally:

cargo test -p dynamo-llm --test http_websocket
cargo test -p dynamo-llm --lib engines::tests
cargo check --workspace

Related Issues: (use one of the action keywords Closes / Fixes / Resolves / Relates to)

• Closes feat: bidirectional streaming-input WebSocket frontend (1/N) #9173 — first slice (1/N) of bidirectional streaming-input.
• Relates to feat: ModelManager accessor for bidirectional engines (2/N) #9174 (2/N) — ModelManager accessor for bidirectional
  engines; retires the OnceLock registry introduced here.
• Relates to feat: dedicated realtime-API protocol types for bidirectional streaming #9175 — dedicated realtime-API protocol type; retires the
  chat-completion overload in generic::realtime.

Internal Linear tickets DIS-1858, DIS-1859, and DIS-1928 mirror the above
GitHub issues (cross-linked per .ai/linear-ticket-refs.md).

[GuanLuo]

@CodeRabbit review

[coderabbitai]

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Walkthrough

This PR implements WebSocket ASR (Automatic Speech Recognition) support by relaxing trait bounds to enable bidirectional streaming engines, adding an echo-based bidirectional chat completion engine, exposing a /v1/asr WebSocket endpoint that handles JSON-over-text streaming, integrating it into the HTTP service, and providing integration tests. It updates the axum dependency to include the ws feature and adds tokio-tungstenite for development.

Changes

WebSocket ASR Bidirectional Streaming

Layer / File(s)	Summary
Runtime Trait Foundation lib/runtime/src/engine.rs, lib/runtime/src/pipeline.rs, Cargo.toml	Relaxed AsyncEngine trait Req bound from Send + Sync + 'static to Send + 'static to permit non-Sync request types; added RequestStream<R> alias; redefined ManyIn<T> from Context<DataStream<T>> to EngineStream<T> for proper streaming input surface; upgraded workspace axum to include ws feature.
Type System Definition lib/llm/src/types.rs, lib/llm/src/http/service.rs	Defined ASRBidirectionalEngine as a BidirectionalStreamingEngine over NvCreateChatCompletionRequest and annotated stream responses; publicly exposed asr submodule in HTTP service.
Bidirectional Engine Logic lib/llm/src/engines.rs	Implemented EchoBidirectionalEngine streaming one character at a time per incoming request, emitting per-character deltas and terminal FinishReason::Stop choices, respecting cancellation; included unit tests validating per-request echo ordering and stop events.
WebSocket Service Handler lib/llm/src/http/service/asr.rs	Implemented /v1/asr WebSocket endpoint requiring pre-installed EchoBidirectionalEngine; handles bidirectional JSON-text streaming with concurrent outbound serialization and inbound frame parsing; rejects Binary frames and malformed JSON with connection closure; includes engine installer functions and close-frame helpers.
Service Integration lib/llm/src/http/service/service_v2.rs	Registered asr_router unconditionally, merging ASR routes into inference_router and appending endpoint docs to OpenAPI output.
Dev Dependencies & Integration Tests lib/llm/Cargo.toml, lib/llm/tests/http_websocket.rs	Added tokio-tungstenite dev-dependency; added three end-to-end WebSocket tests: character-echo validation across multiple requests, server close-frame emission after client shutdown, and rejection of unsupported Binary frame type.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'feat: bidirectional streaming-input WebSocket frontend' accurately summarizes the main change: adding a WebSocket frontend for bidirectional streaming input, which is the core feature delivered in this PR slice.
Docstring Coverage	✅ Passed	Docstring coverage is 92.59% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description check	✅ Passed	The pull request description comprehensively covers all required sections: overview explains the feature and its multi-part nature, details describe frontend changes and runtime fixes with code locations, reviewer guidance is provided with suggested reading order, and related issues are listed with issue numbers.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

Warning

Review ran into problems

🔥 Problems

Git: Failed to clone repository. Please run the @coderabbitai full review command to re-trigger a full review. If the issue persists, set path_filters to include or exclude specific files.

Tip

💬 Introducing Slack Agent: The best way for teams to turn conversations into code.

Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.

• Generate code and open pull requests
• Plan features and break down work
• Investigate incidents and troubleshoot customer tickets together
• Automate recurring tasks and respond to alerts with triggers
• Summarize progress and report instantly

Built for teams:

• Shared memory across your entire org—no repeating context
• Per-thread sandboxes to safely plan and execute work
• Governance built-in—scoped access, auditability, and budget controls

One agent for your entire SDLC. Right inside Slack.

👉 Get started

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 7

🧹 Nitpick comments (2)

lib/llm/src/engines.rs (1)

163-177: 💤 Low value

Fallback text quietly conflates "no messages" and "non-text content".

messages.iter().next_back() returns None for an empty messages array, and the closure also returns None for any non-User variant or non-Text content. All three of those cases collapse onto the same <chunk N: non-text content> placeholder, which is then character-echoed back and followed by FinishReason::Stop. For a mock that exists to demo the bidirectional plumbing this is fine, but the type-level contract on the public struct doc (lines 130-133) only mentions "missing/unsupported"; consider either widening the doc or splitting the placeholder ("empty request" vs "non-text content") so debuggers can tell the cases apart from a captured stream.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/llm/src/engines.rs` around lines 163 - 177, The fallback currently
conflates an empty messages array and non-text content into the same placeholder
when building summary (the let summary = ... using
req.inner.messages.iter().next_back()), so change the logic to first detect an
empty messages list (req.inner.messages.is_empty()) and return a distinct
placeholder like "<chunk {chunk_index}: empty request>" for that case, otherwise
inspect the last message (the existing next_back() / match on
ChatCompletionRequestMessage::User and Text) and keep the current "<chunk
{chunk_index}: non-text content>" only for non-User or non-Text variants; update
the summary construction in engines.rs (the summary binding that uses
req.inner.messages and chunk_index) accordingly so callers and logs can
distinguish empty vs unsupported content.

lib/llm/Cargo.toml (1)

201-201: 💤 Low value

Update tokio-tungstenite to a more recent version to avoid duplicate transitive dependencies.

tokio-tungstenite = "0.21" is significantly behind the latest stable release (v0.29.0). Since axum 0.8's ws feature internally uses tokio-tungstenite ^0.28.0, which depends on tungstenite ^0.29.0, pinning to 0.21 will pull an older tungstenite version into Cargo.lock alongside the one axum brings transitively, creating duplicate versions. As a dev-dependency this has low runtime impact, but updating to ≥0.28 keeps the lock file lean.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/llm/Cargo.toml` at line 201, Replace the old tokio-tungstenite pin in
lib/llm/Cargo.toml (currently tokio-tungstenite = "0.21") with a modern
compatible version (>= 0.28, e.g., tokio-tungstenite = "0.28" or the latest
0.29.x) so it aligns with axum's transitives; after updating the version entry
run cargo update (or cargo update -p tokio-tungstenite) to refresh Cargo.lock
and eliminate duplicate tungstenite transitive versions.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@lib/llm/src/http/service/asr.rs`:
- Around line 44-45: Replace the internal Linear ticket IDs found in comments in
lib/llm/src/http/service/asr.rs (e.g., the comment containing "DIS-1859" near
the ModelManager / `model_name` note and the other occurrence "DIS-1928") with
the corresponding GitHub issue references; locate the comment lines that mention
DIS-1859 and DIS-1928 and update them to the proper GitHub issue numbers/links
(e.g., "GH-<number>" or full "https://github.com/.../issues/<number>") while
preserving the existing comment text and context around ModelManager and the
placeholder note.
- Around line 97-100: Replace the unbounded_channel usage with a bounded channel
(capacity 64) to apply backpressure: create a bounded channel for
NvCreateChatCompletionRequest (keeping req_tx and req_rx names), swap
UnboundedReceiverStream::new(req_rx) for the appropriate
ReceiverStream::new(req_rx) wrapper (or equivalent stream wrapper for the
bounded receiver), and keep constructing RequestStream::new(...) with the boxed
pinned receiver stream and Context::new(()).context(); ensure imports/types are
adjusted to the bounded channel/receiver stream types.

In `@lib/llm/src/http/service/service_v2.rs`:
- Around line 552-558: Update the comment that mentions the internal Linear ID
DIS-1858 by replacing it with the appropriate GitHub issue reference (use prose
form like `#NNNN` or identifier form like GH-NNNN); edit the comment near the
asr_router usage (around symbols asr_router, asr_docs, asr_route,
inference_router) so it no longer contains "DIS-1858" and instead uses the
approved GitHub issue token (or remove the ticket reference entirely if none
applies).

In `@lib/llm/src/types.rs`:
- Around line 141-144: Replace the internal Linear ticket reference "DIS-1928"
in the comment block that starts with "[gluo TODO] DIS-1928: reuses
chat-completion request/response..." with a GitHub issue identifier and prose
form (use GH-NNNN for identifiers and `#NNNN` in prose); update the comment to
read e.g. "[gluo TODO] GH-NNNN: reuses chat-completion..." and change any inline
prose "DIS-1928" to "#NNNN" so the comment no longer contains internal Linear
IDs.
- Around line 152-153: The intra-doc link to RequestStream is broken because
RequestStream isn't exported from crate::types; update the doc comment that
currently points to `crate::types::RequestStream` to reference the actual
re-export location (e.g. `dynamo_runtime::pipeline::RequestStream` or the
original `dynamo_runtime::engine::RequestStream`) so rustdoc can resolve it;
modify the doc string for the `/v1/asr` WebSocket endpoint to use the correct
path (`dynamo_runtime::pipeline::RequestStream`) or the fully qualified original
(`dynamo_runtime::engine::RequestStream`) as appropriate.

In `@lib/llm/tests/http_websocket.rs`:
- Around line 27-28: In the comment containing "OnceLock" and "ModelManager"
replace the internal Linear ticket ID "DIS-1859" with the repository's GitHub
issue format (e.g., "#1859") so the line reads something like "will replace ...
with proper ModelManager-keyed lookups; this helper goes away then (`#1859`)";
ensure only the ticket identifier is changed and the rest of the comment stays
intact.
- Around line 29-35: The test helper ensure_echo_engine_installed mutates the
process env every call (unsafe std::env::set_var) which can race when tests run
in parallel; wrap the env mutation in a one-time initializer (e.g., use
std::sync::Once or OnceCell and call_once) so the unsafe set_var is executed
exactly once before calling asr::install_echo_engine(), and remove/limit the
unsafe block to that single-call site; also update the comment referencing
DIS-1859 to point to the correct GitHub issue number.

---

Nitpick comments:
In `@lib/llm/Cargo.toml`:
- Line 201: Replace the old tokio-tungstenite pin in lib/llm/Cargo.toml
(currently tokio-tungstenite = "0.21") with a modern compatible version (>=
0.28, e.g., tokio-tungstenite = "0.28" or the latest 0.29.x) so it aligns with
axum's transitives; after updating the version entry run cargo update (or cargo
update -p tokio-tungstenite) to refresh Cargo.lock and eliminate duplicate
tungstenite transitive versions.

In `@lib/llm/src/engines.rs`:
- Around line 163-177: The fallback currently conflates an empty messages array
and non-text content into the same placeholder when building summary (the let
summary = ... using req.inner.messages.iter().next_back()), so change the logic
to first detect an empty messages list (req.inner.messages.is_empty()) and
return a distinct placeholder like "<chunk {chunk_index}: empty request>" for
that case, otherwise inspect the last message (the existing next_back() / match
on ChatCompletionRequestMessage::User and Text) and keep the current "<chunk
{chunk_index}: non-text content>" only for non-User or non-Text variants; update
the summary construction in engines.rs (the summary binding that uses
req.inner.messages and chunk_index) accordingly so callers and logs can
distinguish empty vs unsupported content.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 4fb77fdf-d135-499e-be2f-0589aa0377a0

📥 Commits

Reviewing files that changed from the base of the PR and between 8db5ad2 and 5ef98fb.

⛔ Files ignored due to path filters (4)

• Cargo.lock is excluded by !**/*.lock
• lib/bindings/kvbm/Cargo.lock is excluded by !**/*.lock
• lib/bindings/python/Cargo.lock is excluded by !**/*.lock
• lib/runtime/examples/Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (10)

• Cargo.toml
• lib/llm/Cargo.toml
• lib/llm/src/engines.rs
• lib/llm/src/http/service.rs
• lib/llm/src/http/service/asr.rs
• lib/llm/src/http/service/service_v2.rs
• lib/llm/src/types.rs
• lib/llm/tests/http_websocket.rs
• lib/runtime/src/engine.rs
• lib/runtime/src/pipeline.rs

[zhongdaor-nv]

had a peek with claude and the most concerns has been addressed. And the rest concern will be fixed in 2/N.

LGTM.

[furionw]

Thank you.

[grahamking]

@GuanLuo Can you refresh the description / plan? There is no asr.rs.