feat: bidirectional request-plane dispatch + runtime-side ingress

[GuanLuo]

PR description (Rust-only 3/N)

Title (under 70 chars):
feat: bidirectional request-plane dispatch + runtime-side ingress

Base branch: gluo/dis-1859-multimodal-streaming-input-2n-add-bidirectionalstreaming (PR #9308). Retarget to main once #9308 squash-merges.

---

Summary

• Wire end-to-end bidirectional dispatch on the call-home TCP transport: PushRouter::generate(ManyIn<T>) now reaches a real worker through AddressedPushRouter::generate_bidirectional + a new Ingress<ManyIn<T>, ManyOut<U>>::PushWorkHandler impl, completing the runtime side of the streaming-input feature.
• Fills the bail point introduced by feat(llm/realtime): wire /v1/realtime through ModelManager + bidirectional PushRouter #9308 (2/N): bidirectional_dispatch resolves the sticky-selected worker, dispatches via the request plane, awaits both call-home stream halves, and pumps frames in both directions.
• Header-only initial envelope for RequestType::ManyIn; all request frames (including the first) flow on the request-stream socket. Wire-format choice deliberately diverges from the unary [control_msg, request] shape.

Where to start reading

1. lib/runtime/src/pipeline/network.rs — RequestControlMessage::request_stream_connection_info field + the consolidated pump_response_stream helper on Ingress.
2. lib/runtime/src/pipeline/network/tcp/{client,server}.rs — symmetric create_request_stream / process_request_stream pair filling the previously-stubbed request-streaming half of the call-home transport.
3. lib/runtime/src/pipeline/network/egress/addressed_router.rs — generate_bidirectional. Reuses finalize_response_stream with the unary path.
4. lib/runtime/src/pipeline/network/egress/push_router.rs — de-bailed bidirectional AsyncEngine impl; bidirectional_dispatch shares resolve_transport / check_workers_available / wrap_with_fault_detection with the unary path.
5. lib/runtime/src/pipeline/network/ingress/push_handler.rs — PushWorkHandler for Ingress<ManyIn<T>, ManyOut<U>>; shares pump_response_stream with the unary impl.
6. lib/runtime/tests/bidirectional_e2e.rs — full-stack echo round-trip in its own integration-test binary.

Wire-format choice

• RequestType::SingleIn: envelope is HeaderAndData (control + request body) — unchanged.
• RequestType::ManyIn: envelope is HeaderOnly (control only); all request frames flow on the request-stream socket.

RequestControlMessage.request_stream_connection_info: Option<ConnectionInfo> is None for unary, Some(_) for bidirectional. Wire-compatible: older readers ignore the field via #[serde(default, skip_serializing_if = "Option::is_none")]. Each ingress impl strictly accepts only its own envelope shape and rejects the other.

Sync-bound relaxation

This PR also relaxes Req: PipelineIO + Sync to Req: PipelineIO in four places — Frontend::generate, the SegmentSource / ServiceFrontend macro, ServiceBackend::Sink, SegmentSink::Sink — and drops Sync from Ingress::new / add_metrics / link / for_pipeline / for_engine. The bounds were vestigial: AsyncEngine only requires Req: Send + 'static and the bodies don't actually need Sync. Needed so Ingress<ManyIn<T>, ManyOut<U>> (whose ManyIn<T> = Pin<Box<dyn AsyncEngineStream<T>>> is Send but not Sync) can be constructed via the existing Ingress::for_engine path. Pure relaxation — no caller becomes stricter.

Worker selection: reserve-before-observe

The bidirectional PushRouter::generate(ManyIn<T>) reserves a sticky worker BEFORE polling the first frame. Router-mode strategies (RoundRobin, Random, PowerOfTwoChoices, LeastLoaded, DeviceAwareWeighted) don't consume frame content, so waiting for the first frame added no information. Reserve-up-front lets connection setup (register both halves, build envelope, send) run concurrently with the client producing its first frame. A client that connects but never sends releases the slot via the normal response-stream-drop / cancel_both cleanup path.

Test plan

• cargo test --lib -p dynamo-runtime — 150 passed.
• cargo test --test bidirectional_e2e — 1 passed (full-stack echo round-trip).
• cargo clippy --no-deps --all-targets -- -D warnings — clean.
• cargo fmt --check — clean.
• Reviewer manual run of cargo test --test bidirectional_e2e -- --nocapture to confirm the wire path on their machine.

Follow-ups

• Python bridge (PyAsyncRequestStream + PythonBidirectionalEngine + serve_realtime_endpoint + Python E2E test) — split into its own slice. Tracked as issue feat: Python bridge for bidirectional engine (multimodal streaming input 3.5/N) #9673 (3.5/N).
• Runtime-crate latent bug — GLOBAL_TCP_SERVER's accept loop dies with whichever #[tokio::test] runtime first lazy-initializes it. Worked around by running the E2E test in its own tests/ binary (its own process). Followup written up at .feature-development/followup-global-tcp-server-runtime-lifetime.md; file at PR-open time.
• TODO(#171) — three new panic! paths in handle_request_reader follow the existing handle_reader precedent; ci: Trigger TRTLLM build when the associated files are updated #171 grows in urgency.

Related Issues

• Built on feat(llm/realtime): wire /v1/realtime through ModelManager + bidirectional PushRouter #9308 (2/N PR) — adds the bidirectional PushRouter shell with the bail! this PR fills. Currently the base of this branch; will retarget to main once feat(llm/realtime): wire /v1/realtime through ModelManager + bidirectional PushRouter #9308 squash-merges.
• Closes feat: implement remote bidirectional dispatch in PushRouter (3/N) #9361 — 3/N tracking issue.
• Relates to feat: Python bridge for bidirectional engine (multimodal streaming input 3.5/N) #9673 — 3.5/N follow-up (Python bridge).
• Relates to feat: bidirectional streaming-input WebSocket frontend (1/N) #9173 (1/N issue, closed) / PR feat: bidirectional streaming-input WebSocket frontend #9079 (merged) — /v1/realtime route this dispatch serves.
• Relates to feat: ModelManager accessor for bidirectional engines (2/N) #9174 (2/N issue, open) / PR feat(llm/realtime): wire /v1/realtime through ModelManager + bidirectional PushRouter #9308 (in review) — ModelManager accessor surface the watcher arm uses to register discovered realtime workers.
• Relates to DIS-1941 — Backend / Migration bidirectional variants (out of scope here).
• Relates to DIS-1942 — RouterMode::KV / Direct bidirectional support (out of scope here).

Internal Linear tickets DIS-1861 and DGH-860 mirror the GitHub issues above (cross-linked per .ai/linear-ticket-refs.md).

Summary by CodeRabbit

• New Features

  • Added support for bidirectional streaming in the pipeline and request routing system, enabling full duplex communication patterns.

• Refactor

  • Improved request stream handling with mutex-based single-consumption model for thread-safe operation.
  • Reorganized stream connection information tracking for better bidirectional dispatch workflows.
  • Consolidated duplicate request/response dispatch logic into shared helper utilities.

• Tests

  • Added comprehensive end-to-end test validating bidirectional streaming functionality.

[coderabbitai]

Walkthrough

This PR implements bidirectional streaming for the dynamo runtime. It introduces a single-consumption RequestStream<T> model, extends the network protocol and routers to dispatch bidirectional requests, adds ingress handling for bidirectional streams, and validates the full implementation with an end-to-end test.

Changes

Bidirectional Streaming Architecture

Layer / File(s)	Summary
Stream type model: RequestStream and ManyIn lib/runtime/src/engine.rs, lib/runtime/src/pipeline.rs	New RequestStream<T> wraps DataStream<T> in Mutex<Option<_>> for single-consumption. ManyIn<T> alias changes from EngineStream<T> to Context<RequestStream<T>>. Old RequestStream type alias removed from engine. Documentation updated to focus on output-side canonical forms.
Network protocol: RequestControlMessage extension lib/runtime/src/pipeline/network.rs	RequestControlMessage gains optional request_stream_connection_info: Option<ConnectionInfo> field with serde defaults and conditional serialization, enabling bidirectional flow metadata on the wire.
Addressed router: bidirectional dispatch core lib/runtime/src/pipeline/network/egress/addressed_router.rs	New helpers: decode_response_stream (metrics + frame deserialize), build_request_envelope (serialize control + payload), spawn_request_stream_forwarder (async frame forward), subject_of (extract TCP subject). New public generate_bidirectional method. Dispatch logic factored into dispatch_and_finalize (stream registration, tombstone check, request dispatch, forwarder spawn, transport handshake). Unary generate refactored to use shared core. Response-stream error handling refined; test control messages updated.
Push router: fault detection and bidirectional routing lib/runtime/src/pipeline/network/egress/push_router.rs	Unary routing refactored into three phases: overload check, transport resolution with fallback, stream wrapping with quarantine/timeout. New bidirectional_dispatch helper. Bidirectional AsyncEngine::generate changed from first-frame wait with immediate bail to worker selection + dispatch. Test imports and bidirectional input construction updated to use Context::new(RequestStream::new(...)).
Ingress: bidirectional dispatch and work handler lib/runtime/src/pipeline/network/ingress/push_handler.rs	New IngressDispatch impl for Ingress<ManyIn<T>, ManyOut<U>> parses header-only envelopes, validates request type, extracts request-stream connection info, opens request TCP stream, spawns async frame forwarder to deserialize and feed engine input. New PushWorkHandler impl wires shared payload handler. Imports and docs updated.
LLM layer: stream model adaptation lib/llm/src/engines.rs, lib/llm/src/http/service/realtime.rs	Realtime service constructs Context::new(RequestStream::new(...)) instead of old wrapped form. Mock EchoBidirectionalEngine destructures ManyIn, extracts RequestStream via take(). Test helper make_input uses new stream construction. Imports restructured.
End-to-end bidirectional integration test lib/runtime/tests/bidirectional_e2e.rs	New comprehensive test: EchoEngine implements AsyncEngine<ManyIn<u64>, ManyOut<EchoResponse>>. Full distributed runtime setup with server registration and client, validates round-trip frame echoing across three items.

🎯 4 (Complex) | ⏱️ ~60 minutes

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately and specifically describes the main change: implementing bidirectional request-plane dispatch and runtime-side ingress for streaming input functionality.
Description check	✅ Passed	The description is comprehensive and well-structured. It includes an overview of the changes, detailed explanations of wire-format choices, specific files to review, test plan results, and related issues. All required template sections are covered.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% 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.

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

---

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

🤖 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/runtime/src/pipeline/network/egress/addressed_router.rs`:
- Around line 154-162: The control message currently sets
RequestControlMessage.frontend_send_ts_ns to None, which disables
WORK_HANDLER_NETWORK_TRANSIT_SECONDS; update the constructor for
RequestControlMessage (in addressed_router.rs) to read the frontend send
timestamp emitted by dispatch_buffer (the request-plane header), parse/convert
it to the same type used by frontend_send_ts_ns, and set frontend_send_ts_ns =
Some(parsed_value) so the timestamp is serialized into the control envelope;
ensure the code path that constructs the control message (function creating
control_message) gracefully handles a missing header (leave None) and matches
the deserialization logic in Ingress::handle_payload_shared.

In `@lib/runtime/src/pipeline/network/egress/push_router.rs`:
- Around line 1084-1088: The current bidi path calls select_next_worker() which
returns None for RouterMode::PowerOfTwoChoices, RouterMode::LeastLoaded, and
RouterMode::DeviceAwareWeighted, losing occupancy tracking and incorrectly
erroring; modify the bidi handling in push_router.rs so that before calling
bidirectional_dispatch() you branch on the router mode and either (a) perform
the mode-specific selection/permit flow used by the corresponding unidirectional
paths (replicating the occupancy/permit acquisition logic used by those
selectors) and pass the chosen instance_id to
bidirectional_dispatch(instance_id, input).await, or (b) if bidi is not
supported for a given mode, explicitly return a clear error rejecting those
RouterMode variants (PowerOfTwoChoices, LeastLoaded, DeviceAwareWeighted)
instead of calling select_next_worker(); locate logic around
select_next_worker() and bidirectional_dispatch() to implement the mode-based
routing/permit acquisition or explicit rejection.

🪄 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: ec4b0be4-4000-4ee1-9b5d-8fb661132b86

📥 Commits

Reviewing files that changed from the base of the PR and between 1699aef and 2805e9b.

📒 Files selected for processing (9)

• lib/llm/src/engines.rs
• lib/llm/src/http/service/realtime.rs
• lib/runtime/src/engine.rs
• lib/runtime/src/pipeline.rs
• lib/runtime/src/pipeline/network.rs
• lib/runtime/src/pipeline/network/egress/addressed_router.rs
• lib/runtime/src/pipeline/network/egress/push_router.rs
• lib/runtime/src/pipeline/network/ingress/push_handler.rs
• lib/runtime/tests/bidirectional_e2e.rs

[furionw]

Thanks for working on this, @GuanLuo .

• one command on reuse decode_control_message
• another blocking comment on avoid panic
• others are generated by code-review skills and you can address as you see fit

[GuanLuo]

RE: review feedback

Thanks @furionw, @krishung5. Addressed in 5 commits (per-thread replies point at each):

• Avoid panic (blocking): c00e576c43 — take()/builder expect/unwrap → Err; transport post-conditions → debug_assert_eq!.
• Reuse decode_control_message: d66857efea — shared helper returning (RequestControlMessage, Option<data>); unary requires the data half, bidi rejects it.
• Reject unsupported router modes: 27fb870b10 — explicit bail! for PowerOfTwoChoices/LeastLoaded/DeviceAwareWeighted (+ test); full support tracked in Bidirectional PushRouter only wires RoundRobin/Random worker selection #10320.
• Cancellation symmetry: a5accd6f36 — bidi recv forwarder now honors stop_generating().
• Nits (clone, typos, {e}, scoped use, into_parts, SPDX, docstring): 4caa158707.

One dismissed: the frontend_send_ts_ns flag — not a regression (the transit metric is header-driven at shared_tcp_endpoint.rs:232, same as main); details in that thread.

[krishung5]

LGTM, thanks!