feat: Add browser tests for streaming RPC methods

Add comprehensive browser tests for all 4 streaming patterns:
- sum: client-to-server streaming (Push<T> → scalar)
- range: server-to-client streaming (scalar → Pull<T>)
- pipe: bidirectional streaming (Push<T> ↔ Pull<T>)
- stats: aggregating stream to compound result

Key fixes to make streaming work:
- Spawn streaming handlers as async tasks to avoid deadlock
- Flush outgoing Data/Close before Response for server-to-client
- Wrap streaming results in CallResult (like unary methods)
- Add Clone + 'static bounds for service types in dispatchers

Author: fasterthanlime

rust/codegen-test-consumer/src/bin/ws_server.rs

@@ -10,6 +10,7 @@ use std::env;
 use tokio::net::TcpListener;
 use tokio_tungstenite::accept_async;
 
+#[derive(Clone)]
 struct Calculator;
 
 #[allow(clippy::manual_async_fn)]

rust/codegen-test-consumer/src/lib.rs

@@ -17,6 +17,7 @@ mod tests {
     }
 
     // Test that we can implement the handler trait
+    #[derive(Clone)]
     struct TestCalculator;
 
     #[allow(clippy::manual_async_fn)]

rust/roam-codegen/src/targets/rust.rs

@@ -188,10 +188,13 @@ impl<'a> RustGenerator<'a> {
         }
 
         // impl ServiceDispatcher for Dispatcher
+        // S: Clone is required so dispatch_streaming can clone the service into 'static futures
         {
             let impl_block = self.scope.new_impl(&format!("{service_name}Dispatcher<S>"));
             impl_block.generic("S");
             impl_block.bound("S", &handler_trait);
+            impl_block.bound("S", "Clone");
+            impl_block.bound("S", "'static");
             impl_block.impl_trait("::roam_stream::ServiceDispatcher");
 
             generate_is_streaming(impl_block, self.service);
@@ -371,7 +374,8 @@ fn generate_dispatch_streaming(
     service: &ServiceDetail,
     options: &RustCodegenOptions,
 ) {
-    let ret_type = "::std::pin::Pin<::std::boxed::Box<dyn ::std::future::Future<Output = ::std::result::Result<::std::vec::Vec<u8>, ::std::string::String>> + Send + '_>>";
+    // Return type uses 'static so the future can be spawned by the connection
+    let ret_type = "::std::pin::Pin<::std::boxed::Box<dyn ::std::future::Future<Output = ::std::result::Result<::std::vec::Vec<u8>, ::std::string::String>> + Send + 'static>>";
 
     let func = impl_block.new_fn("dispatch_streaming");
     func.arg_ref_self();
@@ -497,6 +501,9 @@ fn generate_streaming_setup_and_dispatch(
     let arg_names: Vec<String> = method.args.iter().map(|a| a.name.to_snake_case()).collect();
     let call_args = arg_names.join(", ");
 
+    // Clone service before async block so the future is 'static
+    block.line("let service = self.service.clone();");
+
     let mut async_block = Block::new("::std::boxed::Box::pin(async move");
     async_block.after(")");
 
@@ -511,27 +518,31 @@ fn generate_streaming_setup_and_dispatch(
             .line("tracing::debug!(payload_len = payload.len(), \"streaming request received\");");
     }
 
+    // Call handler and encode response per spec r[unary.response.encoding]
+    // Response is always CallResult<T, Never> = Result<T, RoamError<Never>>
+    let return_ty = rust_type_server_return(&method.return_type);
+
     // For unit return types, don't bind to a variable (clippy: let_unit_value)
     if method.return_type == TypeDetail::Unit {
-        async_block.line(format!("self.service.{method_name}({call_args}).await"));
+        async_block.line(format!("service.{method_name}({call_args}).await"));
         async_block.line("    .map_err(|e| format!(\"method error: {e}\"))?;");
-
-        if options.tracing {
-            async_block.line("let response = facet_postcard::to_vec(&())");
-        } else {
-            async_block.line("facet_postcard::to_vec(&())");
-        }
+        async_block.line(format!(
+            "let call_result: CallResult<{return_ty}, Never> = Ok(());"
+        ));
     } else {
         async_block.line(format!(
-            "let result = self.service.{method_name}({call_args}).await"
+            "let result = service.{method_name}({call_args}).await"
         ));
         async_block.line("    .map_err(|e| format!(\"method error: {e}\"))?;");
+        async_block.line(format!(
+            "let call_result: CallResult<{return_ty}, Never> = Ok(result);"
+        ));
+    }
 
-        if options.tracing {
-            async_block.line("let response = facet_postcard::to_vec(&result)");
-        } else {
-            async_block.line("facet_postcard::to_vec(&result)");
-        }
+    if options.tracing {
+        async_block.line("let response = facet_postcard::to_vec(&call_result)");
+    } else {
+        async_block.line("facet_postcard::to_vec(&call_result)");
     }
 
     if options.tracing {

rust/roam-codegen/src/targets/typescript.rs

@@ -526,12 +526,14 @@ fn generate_handler_interface(service: &ServiceDetail) -> String {
         let can_encode_return = is_fully_supported(&method.return_type);
 
         if can_decode_args && can_encode_return {
-            // Decode all arguments
+            // Decode all arguments (using server context for proper Push/Pull inversion)
             out.push_str("      const buf = payload;\n");
             out.push_str("      let offset = 0;\n");
             for arg in &method.args {
                 let arg_name = arg.name.to_lower_camel_case();
-                let decode_stmt = generate_decode_stmt(&arg.type_info, &arg_name, "offset");
+                // Use server-side decode for proper Push/Pull inversion
+                // r[impl streaming.caller-pov] - Schema is caller's perspective, server inverts.
+                let decode_stmt = generate_decode_stmt_server(&arg.type_info, &arg_name, "offset");
                 out.push_str(&format!("      {decode_stmt}\n"));
             }
             out.push_str(
@@ -849,6 +851,35 @@ fn generate_decode_stmt_client(ty: &TypeDetail, var_name: &str, offset_var: &str
     }
 }
 
+/// Generate TypeScript code that decodes a value from a buffer for SERVER context.
+/// Schema types are from caller's perspective - server needs INVERTED types:
+/// - Schema Push (caller sends) → server receives → Pull for server
+/// - Schema Pull (caller receives) → server sends → Push for server
+///
+/// r[impl streaming.caller-pov] - Schema is from caller's perspective, server inverts.
+fn generate_decode_stmt_server(ty: &TypeDetail, var_name: &str, offset_var: &str) -> String {
+    match ty {
+        TypeDetail::Push(inner) => {
+            // Schema Push (caller sends) → server receives → Pull for server
+            // r[impl streaming.type] - Stream types decode as stream_id on wire.
+            let inner_type = ts_type_server_arg(inner);
+            format!(
+                "const _{var_name}_r = decodeU64(buf, {offset_var}); const {var_name} = {{ streamId: _{var_name}_r.value }} as Pull<{inner_type}>; {offset_var} = _{var_name}_r.next; /* TODO: create real Pull handle */"
+            )
+        }
+        TypeDetail::Pull(inner) => {
+            // Schema Pull (caller receives) → server sends → Push for server
+            // r[impl streaming.type] - Stream types decode as stream_id on wire.
+            let inner_type = ts_type_server_arg(inner);
+            format!(
+                "const _{var_name}_r = decodeU64(buf, {offset_var}); const {var_name} = {{ streamId: _{var_name}_r.value }} as Push<{inner_type}>; {offset_var} = _{var_name}_r.next; /* TODO: create real Push handle */"
+            )
+        }
+        // For non-streaming types, use the regular decode
+        _ => generate_decode_stmt(ty, var_name, offset_var),
+    }
+}
+
 /// Generate TypeScript code that decodes a value from a buffer.
 /// Returns the decoded value in a variable and updates offset.
 /// `var_name` is the variable to assign the result to.

rust/roam-session/src/lib.rs

@@ -692,7 +692,8 @@ pub trait ServiceDispatcher: Send + Sync {
     /// - Call the handler method with those handles
     /// - Serialize the response
     ///
-    /// Returns a boxed future since each streaming method may have different async block types.
+    /// Returns a boxed future with `'static` lifetime so it can be spawned.
+    /// Implementations should clone their service into the future to achieve this.
     ///
     /// Takes ownership of the payload to avoid copies - the caller already owns it from
     /// the decoded message frame.
@@ -703,7 +704,9 @@ pub trait ServiceDispatcher: Send + Sync {
         method_id: u64,
         payload: Vec<u8>,
         registry: &mut StreamRegistry,
-    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, String>> + Send + '_>>;
+    ) -> std::pin::Pin<
+        Box<dyn std::future::Future<Output = Result<Vec<u8>, String>> + Send + 'static>,
+    >;
 }
 
 /// A dispatcher that routes to one of two dispatchers based on method ID.
@@ -763,8 +766,9 @@ where
         method_id: u64,
         payload: Vec<u8>,
         registry: &mut StreamRegistry,
-    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, String>> + Send + '_>>
-    {
+    ) -> std::pin::Pin<
+        Box<dyn std::future::Future<Output = Result<Vec<u8>, String>> + Send + 'static>,
+    > {
         if self.first_methods.contains(&method_id) {
             self.first.dispatch_streaming(method_id, payload, registry)
         } else {

rust/roam-stream/src/connection.rs

@@ -54,6 +54,9 @@ impl From<std::io::Error> for ConnectionError {
     }
 }
 
+/// Result from a completed streaming handler: (request_id, serialized_result).
+type StreamingResult = (u64, Result<Vec<u8>, String>);
+
 /// A live connection with completed Hello exchange.
 ///
 /// Generic over the transport type `T` which must implement [`MessageTransport`].
@@ -70,6 +73,10 @@ pub struct Connection<T> {
     in_flight_requests: HashSet<u64>,
     #[allow(dead_code)]
     our_hello: Hello,
+    /// Channel for receiving completed streaming handler results.
+    /// Spawned tasks send (request_id, result) when they complete.
+    streaming_results_tx: tokio::sync::mpsc::Sender<StreamingResult>,
+    streaming_results_rx: tokio::sync::mpsc::Receiver<StreamingResult>,
 }
 
 impl<T> Connection<T> {
@@ -211,6 +218,27 @@ where
             tokio::select! {
                 biased;
 
+                // Handle completed streaming handlers
+                Some((request_id, result)) = self.streaming_results_rx.recv() => {
+                    let response_payload = result.map_err(ConnectionError::Dispatch)?;
+
+                    // Flush any outgoing stream data (Data/Close) BEFORE Response.
+                    // This ensures the client receives all streamed data before the
+                    // Response that signals call completion.
+                    // r[impl streaming.flush-before-response] - Stream data sent before Response.
+                    self.flush_outgoing().await?;
+
+                    // r[impl streaming.call-complete] - Call completes when Response sent.
+                    // r[impl streaming.lifecycle.response-closes-pulls] - Pull streams close with Response.
+                    let resp = Message::Response {
+                        request_id,
+                        metadata: Vec::new(),
+                        payload: response_payload,
+                    };
+                    self.io.send(&resp).await?;
+                    self.in_flight_requests.remove(&request_id);
+                }
+
                 // Prioritize incoming messages over outgoing flush
                 result = self.io.recv_timeout(Duration::from_secs(30)) => {
                     let msg = match result {
@@ -295,35 +323,50 @@ where
                 }
 
                 // Dispatch to service - use streaming dispatch if method has Push/Pull args
-                let response_payload = if dispatcher.is_streaming(method_id) {
-                    dispatcher
-                        .dispatch_streaming(method_id, payload, &mut self.stream_registry)
-                        .await
-                        .map_err(ConnectionError::Dispatch)?
+                if dispatcher.is_streaming(method_id) {
+                    // For streaming methods, we need to continue processing messages
+                    // (Data, Close) while the handler runs. The handler reads from
+                    // Pull<T> which is backed by an mpsc channel that we route to.
+                    //
+                    // dispatch_streaming registers streams synchronously, then returns
+                    // a future. We spawn that future as a task so the message loop
+                    // can continue processing Data messages.
+                    let handler_fut = dispatcher.dispatch_streaming(
+                        method_id,
+                        payload,
+                        &mut self.stream_registry,
+                    );
+
+                    // Spawn the handler as a task that sends its result to our channel
+                    let results_tx = self.streaming_results_tx.clone();
+                    tokio::spawn(async move {
+                        let result = handler_fut.await;
+                        // Send result to the connection's run loop
+                        // Ignore send error if connection closed
+                        let _ = results_tx.send((request_id, result)).await;
+                    });
                 } else {
-                    dispatcher
+                    let response_payload = dispatcher
                         .dispatch_unary(method_id, &payload)
                         .await
-                        .map_err(ConnectionError::Dispatch)?
-                };
-
-                // r[impl core.call] - Callee sends Response for caller's Request.
-                // r[impl core.call.request-id] - Response has same request_id.
-                // r[impl unary.complete] - Send Response with matching request_id.
-                // r[impl unary.lifecycle.single-response] - Exactly one Response per Request.
-                // r[impl unary.request-id.in-flight] - Request no longer in-flight after Response.
-                // r[impl streaming.call-complete] - Call completes when Response sent.
-                // r[impl streaming.lifecycle.response-closes-pulls] - Pull streams close with Response.
-                let resp = Message::Response {
-                    request_id,
-                    metadata: Vec::new(),
-                    payload: response_payload,
-                };
-                self.io.send(&resp).await?;
-                self.in_flight_requests.remove(&request_id);
-
-                // Flush any outgoing stream data that handlers may have queued
-                self.flush_outgoing().await?;
+                        .map_err(ConnectionError::Dispatch)?;
+
+                    // r[impl core.call] - Callee sends Response for caller's Request.
+                    // r[impl core.call.request-id] - Response has same request_id.
+                    // r[impl unary.complete] - Send Response with matching request_id.
+                    // r[impl unary.lifecycle.single-response] - Exactly one Response per Request.
+                    // r[impl unary.request-id.in-flight] - Request no longer in-flight after Response.
+                    let resp = Message::Response {
+                        request_id,
+                        metadata: Vec::new(),
+                        payload: response_payload,
+                    };
+                    self.io.send(&resp).await?;
+                    self.in_flight_requests.remove(&request_id);
+
+                    // Flush any outgoing stream data that handlers may have queued
+                    self.flush_outgoing().await?;
+                }
             }
             Message::Response { request_id, .. } => {
                 // Server doesn't expect Response messages (it sends them, not receives them).
@@ -472,6 +515,7 @@ where
         initial_credit: our_credit.min(peer_credit),
     };
 
+    let (streaming_results_tx, streaming_results_rx) = tokio::sync::mpsc::channel(64);
     Ok(Connection {
         io,
         role: Role::Acceptor,
@@ -481,6 +525,8 @@ where
         stream_registry: StreamRegistry::new_with_credit(negotiated.initial_credit),
         in_flight_requests: HashSet::new(),
         our_hello,
+        streaming_results_tx,
+        streaming_results_rx,
     })
 }
 
@@ -551,6 +597,7 @@ where
         initial_credit: our_credit.min(peer_credit),
     };
 
+    let (streaming_results_tx, streaming_results_rx) = tokio::sync::mpsc::channel(64);
     Ok(Connection {
         io,
         role: Role::Initiator,
@@ -560,5 +607,7 @@ where
         stream_registry: StreamRegistry::new_with_credit(negotiated.initial_credit),
         in_flight_requests: HashSet::new(),
         our_hello,
+        streaming_results_tx,
+        streaming_results_rx,
     })
 }

rust/subject-rust/src/main.rs

@@ -12,6 +12,7 @@ pub use spec_proto::{Canvas, Color, Message, Person, Point, Rectangle, Shape};
 include!(concat!(env!("OUT_DIR"), "/generated.rs"));
 
 // Service implementation using generated EchoHandler trait
+#[derive(Clone)]
 struct EchoService;
 
 #[allow(clippy::manual_async_fn)]

spec/spec-proto/src/lib.rs

@@ -79,30 +79,31 @@ pub enum Message {
 /// Streaming service for cross-language conformance testing.
 ///
 /// Tests Push/Pull semantics, stream lifecycle, and bidirectional streaming.
+/// r[impl streaming.caller-pov] - Types are from caller's perspective.
 #[service]
 pub trait Streaming {
     /// Client pushes numbers, server returns their sum.
     ///
     /// Tests: client-to-server streaming (`Push<T>` → scalar return).
     /// r[impl streaming.client-to-server] - Client sends stream, server returns scalar.
-    async fn sum(&self, numbers: Pull<i32>) -> i64;
+    async fn sum(&self, numbers: Push<i32>) -> i64;
 
-    /// Client sends a count, server returns that many numbers.
+    /// Client sends a count, server streams that many numbers back.
     ///
-    /// Tests: server-to-client streaming (scalar → `Pull<T>`).
+    /// Tests: server-to-client streaming (scalar → `Pull<T>` as output parameter).
     /// r[impl streaming.server-to-client] - Client sends scalar, server returns stream.
-    async fn range(&self, count: u32) -> Push<u32>;
+    async fn range(&self, count: u32, output: Pull<u32>);
 
     /// Client pushes strings, server echoes each back.
     ///
     /// Tests: bidirectional streaming (`Push<T>` ↔ `Pull<T>`).
     /// r[impl streaming.bidirectional] - Both sides stream simultaneously.
-    async fn pipe(&self, input: Pull<String>) -> Push<String>;
+    async fn pipe(&self, input: Push<String>, output: Pull<String>);
 
     /// Client pushes numbers, server returns (sum, count, average).
     ///
     /// Tests: aggregating a stream into a compound result.
-    async fn stats(&self, numbers: Pull<i32>) -> (i64, u64, f64);
+    async fn stats(&self, numbers: Push<i32>) -> (i64, u64, f64);
 }
 
 /// Complex types service for testing struct/enum encoding.