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

lib/llm/src/engines.rs

@@ -10,13 +10,12 @@ use async_stream::stream;
 use async_trait::async_trait;
 
 use dynamo_runtime::engine::{AsyncEngine, AsyncEngineContextProvider, ResponseStream};
+#[cfg(test)]
+use dynamo_runtime::pipeline::RequestStream;
 use dynamo_runtime::pipeline::{Error, ManyIn, ManyOut, SingleIn};
 use dynamo_runtime::protocols::annotated::Annotated;
 use futures::StreamExt;
 
-#[cfg(test)]
-use dynamo_runtime::engine::RequestStream;
-
 use crate::protocols::openai::{
     chat_completions::{NvCreateChatCompletionRequest, NvCreateChatCompletionStreamResponse},
     completions::{NvCreateCompletionRequest, NvCreateCompletionResponse, prompt_to_string},
@@ -152,11 +151,15 @@ impl AsyncEngine<ManyIn<RealtimeClientEvent>, ManyOut<Annotated<RealtimeServerEv
 {
     async fn generate(
         &self,
-        mut incoming: ManyIn<RealtimeClientEvent>,
+        incoming: ManyIn<RealtimeClientEvent>,
     ) -> Result<ManyOut<Annotated<RealtimeServerEvent>>, Error> {
         let ctx = incoming.context();
         let session_id = ctx.id().to_string();
         let ctx_for_stream = ctx.clone();
+        let (request_stream, _ctx_unit) = incoming.into_parts();
+        let mut incoming = request_stream.take().ok_or_else(|| {
+            anyhow::anyhow!("RequestStream::take called twice on bidirectional input")
+        })?;
 
         let output = stream! {
             let ctx = ctx_for_stream;
@@ -561,7 +564,7 @@ mod tests {
     use futures::stream;
 
     fn make_input(events: Vec<RealtimeClientEvent>) -> ManyIn<RealtimeClientEvent> {
-        RequestStream::new(Box::pin(stream::iter(events)), Context::new(()).context())
+        Context::new(RequestStream::new(Box::pin(stream::iter(events))))
     }
 
     fn parse_client_event(json: serde_json::Value) -> RealtimeClientEvent {

lib/llm/src/http/service/realtime.rs

@@ -43,8 +43,8 @@ use axum::{
     response::Response,
     routing::get,
 };
-use dynamo_runtime::engine::{AsyncEngineContextProvider, RequestStream};
-use dynamo_runtime::pipeline::Context;
+use dynamo_runtime::engine::AsyncEngineContextProvider;
+use dynamo_runtime::pipeline::{Context, RequestStream};
 use futures::{SinkExt, StreamExt, stream::SplitSink};
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
@@ -145,7 +145,7 @@ async fn handle_socket(socket: WebSocket, state: Arc<service_v2::State>) {
     }
 
     let request_stream = Box::pin(ReceiverStream::new(req_rx));
-    let input = RequestStream::new(request_stream, Context::new(()).context());
+    let input = Context::new(RequestStream::new(request_stream));
 
     let mut response_stream = match engine.generate(input).await {
         Ok(s) => s,

lib/runtime/src/engine.rs

@@ -180,14 +180,9 @@ pub trait AsyncEngineUnary<Resp: Data>:
 ///
 /// This trait combines `Stream` semantics with context provider capabilities,
 /// representing a continuous async operation that produces multiple messages over time.
-///
-/// - **Output side:** wrapped as [`EngineStream<T>`] = `crate::pipeline::ManyOut<T>`
-///   — the stream of response chunks an engine emits.
-/// - **Input side:** same `EngineStream<T>` shape, exposed as
-///   `crate::pipeline::ManyIn<T>` for documentary clarity at the call site.
-///
-/// [`ResponseStream`] is the canonical concrete implementor; [`RequestStream`]
-/// is a type alias of it for the input side.
+/// The canonical boxed form is [`EngineStream<T>`] (= `crate::pipeline::ManyOut<T>`),
+/// the stream of response chunks an engine emits; [`ResponseStream`] is the
+/// canonical concrete implementor.
 pub trait AsyncEngineStream<T: Data>: Stream<Item = T> + AsyncEngineContextProvider + Send {}
 
 /// Engine is a trait that defines the interface for a streaming engine.
@@ -267,16 +262,6 @@ impl<R: Data> Debug for ResponseStream<R> {
     }
 }
 
-/// Input-side type alias of [`ResponseStream`] — same struct, different name to
-/// signal role at the call site.
-///
-/// The shape is identical: a `(stream, ctx)` pair that implements [`Stream`],
-/// [`AsyncEngineContextProvider`], and [`AsyncEngineStream`]. Use `RequestStream`
-/// when you're constructing a value to feed into the `Req` slot of an engine,
-/// and [`ResponseStream`] when constructing a value to emit from the `Resp` slot.
-/// Functionally interchangeable.
-pub type RequestStream<R> = ResponseStream<R>;
-
 impl<T: Data> AsyncEngineContextProvider for Pin<Box<dyn AsyncEngineUnary<T>>> {
     fn context(&self) -> Arc<dyn AsyncEngineContext> {
         AsyncEngineContextProvider::context(&**self)

lib/runtime/src/pipeline.rs

@@ -19,8 +19,8 @@ pub use network::egress::push_router::{PushRouter, RouterMode, WorkerLoadMonitor
 pub mod registry;
 
 pub use crate::engine::{
-    self as engine, AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, Data, DataStream,
-    Engine, EngineStream, EngineUnary, RequestStream, ResponseStream, async_trait,
+    self as engine, AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, AsyncEngineStream,
+    Data, DataStream, Engine, EngineStream, EngineUnary, ResponseStream, async_trait,
 };
 pub use anyhow::Error;
 pub use context::Context;
@@ -30,18 +30,44 @@ pub use error::{PipelineError, PipelineErrorExt, TwoPartCodecError};
 /// about the request. This information propagates through the stages, both local and distributed.
 pub type SingleIn<T> = Context<T>;
 
-/// Pipeline input for streaming-request engines: a trait-object alias around a
-/// stream that carries its own cancellation context (via the
-/// [`AsyncEngineContextProvider`] half of [`AsyncEngineStream`]).
-///
-/// This is the input-side mirror of [`ManyOut`]. Both aliases resolve to the
-/// same underlying [`EngineStream<T>`] type; the directional names are
-/// documentary. Earlier definitions wrapped the stream in a `Context<…>`
-/// (`Context<DataStream<T>>`); that shape was uninstantiable because
-/// `DataStream<T>` is `!Sync` while `Context<T: Data>` requires `Sync`. The
-/// trait-object form solves that cleanly: the cancellation surface is part of
-/// the trait contract rather than an outer wrapper.
-pub type ManyIn<T> = EngineStream<T>;
+/// `Sync` ownership cell for a non-`Sync` [`DataStream<T>`]. [`Self::take`]
+/// moves the inner stream out; the mutex serialises concurrent attempts so
+/// the first caller observes `Some(stream)` and all later callers see
+/// `None`. Iteration happens on the returned `DataStream<T>`.
+pub struct RequestStream<T: Data> {
+    inner: std::sync::Mutex<Option<DataStream<T>>>,
+}
+
+impl<T: Data> RequestStream<T> {
+    /// Wrap a [`DataStream<T>`] in a `Sync` ownership cell.
+    pub fn new(stream: DataStream<T>) -> Self {
+        Self {
+            inner: std::sync::Mutex::new(Some(stream)),
+        }
+    }
+
+    /// Atomically move the inner stream out. Returns `Some(stream)` exactly
+    /// once across all threads racing on the same `RequestStream`; every
+    /// subsequent call (on any thread) returns `None`. The returned stream
+    /// is the unique owner; the wrapper retains nothing.
+    pub fn take(&self) -> Option<DataStream<T>> {
+        self.inner.lock().unwrap().take()
+    }
+}
+
+impl<T: Data> std::fmt::Debug for RequestStream<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let taken = self.inner.lock().map(|g| g.is_none()).unwrap_or(true);
+        f.debug_struct("RequestStream")
+            .field("taken", &taken)
+            .finish()
+    }
+}
+
+/// Pipeline input for streaming-request engines: a [`RequestStream<T>`]
+/// payload wrapped in a [`Context`], symmetric to [`SingleIn<T>`] for unary
+/// inputs.
+pub type ManyIn<T> = Context<RequestStream<T>>;
 
 /// Type alias for the output of pipeline that returns a single value
 pub type SingleOut<T> = EngineUnary<T>;

lib/runtime/src/pipeline/network.rs

@@ -28,10 +28,12 @@ use super::{AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, Respons
 use serde::{Deserialize, Serialize};
 
 use super::{
-    AsyncTransportEngine, Context, Data, Error, ManyOut, PipelineError, PipelineIO, SegmentSource,
-    ServiceBackend, ServiceEngine, SingleIn, Source, context,
+    AsyncTransportEngine, Context, Data, Error, ManyIn, ManyOut, PipelineError, PipelineIO,
+    SegmentSource, ServiceBackend, ServiceEngine, SingleIn, Source, context,
 };
 use crate::metrics::MetricsHierarchy;
+use crate::metrics::prometheus_names::work_handler;
+use crate::protocols::maybe_error::MaybeError;
 use ingress::push_handler::WorkHandlerMetrics;
 use prometheus::{CounterVec, Histogram, IntCounter, IntCounterVec, IntGauge};
 
@@ -94,6 +96,11 @@ pub(crate) struct RequestControlMessage {
     /// Reliable for single-machine profiling; treat cross-host values as approximate.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub(crate) frontend_send_ts_ns: Option<u64>,
+    /// For bidirectional dispatch (`request_type == ManyIn`): connection info the
+    /// worker dials back to in order to receive subsequent request frames. `None`
+    /// for the unary path, which is the wire-compatible default.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) request_stream_connection_info: Option<ConnectionInfo>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]