Fix EventStream initial message handling to prevent hangs

Fixes #4435

Previously, codegen would always call `try_recv_initial()` for event
streams, which would block waiting for an initial-request or
initial-response message. This caused hangs when operations didn't have
modeled initial messages.

**Changes:**

Runtime (aws-smithy-http & aws-smithy-legacy-http):
- `recv()` now filters out initial-request/initial-response messages
  and stores them in a separate buffer
- `try_recv_initial()` checks the buffer first before blocking
- Initial messages are discarded if never explicitly read
- Added `try_recv_initial_with_preprocessor()` to legacy version for
  SigV4 support

Codegen:
- Server: Only calls `try_recv_initial()` when
  `httpBindingResolver.handlesEventStreamInitialRequest()` returns true
- Client: Only calls `try_recv_initial_response()` when there's a
  parser for non-event-stream output members
- Uses `let mut receiver = receiver` pattern for conditional mutability

Tests:
- Added `test_no_hang_without_initial_message()` to verify operations
  without modeled initial messages don't hang
- Added `test_waits_for_initial_message_when_modeled()` to verify
  operations with modeled initial messages still block correctly
- Added `wait_for_response_ready()` to ManualEventStreamClient for
  testing connection readiness

Documentation:
- Updated AGENTS.md with EventStream initial message handling behavior
- Added conditional mutability pattern to AGENTS.md

Author: rcoh

AGENTS.md

@@ -50,14 +50,18 @@ operation MyOperation {
 
 ### httpQueryParams Bug Investigation
 
-When investigating the `@httpQueryParams` bug (where query parameters weren't appearing in requests), the issue was in `RequestBindingGenerator.kt` line 173. The bug occurred when:
+When investigating the `@httpQueryParams` bug (where query parameters weren't appearing in requests), the issue was in
+`RequestBindingGenerator.kt` line 173. The bug occurred when:
 
 1. An operation had ONLY `@httpQueryParams` (no regular `@httpQuery` parameters)
-2. The condition `if (dynamicParams.isEmpty() && literalParams.isEmpty() && mapParams.isEmpty())` would skip generating the `uri_query` function
+2. The condition `if (dynamicParams.isEmpty() && literalParams.isEmpty() && mapParams.isEmpty())` would skip generating
+   the `uri_query` function
 
-The fix was to ensure `mapParams.isEmpty()` was included in the condition check. The current implementation correctly generates query parameters for `@httpQueryParams` even when no other query parameters exist.
+The fix was to ensure `mapParams.isEmpty()` was included in the condition check. The current implementation correctly
+generates query parameters for `@httpQueryParams` even when no other query parameters exist.
 
-**Testing httpQueryParams**: Create operations with only `@httpQueryParams` to ensure they generate proper query strings in requests.
+**Testing httpQueryParams**: Create operations with only `@httpQueryParams` to ensure they generate proper query strings
+in requests.
 
 ## rustTemplate Formatting
 
@@ -83,6 +87,28 @@ rustTemplate(
 ❌ Wrong: `"let result: Result<String, Error> = Ok(value);"`
 ✅ Correct: Use `*preludeScope` in templates
 
+## Conditional Mutability Pattern
+
+When a variable needs to be mutable only in certain codegen branches, use the rebinding pattern:
+
+```kotlin
+rustTemplate(
+    """
+    let receiver = create_receiver();
+    #{maybeModifyReceiver:W}
+    use_receiver(receiver);
+    """,
+    "maybeModifyReceiver" to writable {
+        if (needsMutability) {
+            rust("let mut receiver = receiver;")
+            rust("receiver.modify();")
+        }
+    }
+)
+```
+
+This avoids unused `mut` warnings when the modification code isn't generated.
+
 ## RuntimeType and Dependencies
 
 `RuntimeType` objects contain:
@@ -176,6 +202,7 @@ gh issue comment <number> --repo smithy-lang/smithy-rs --body 'markdown content
 ```
 
 **Comment Guidelines:**
+
 - Always ask for confirmation before posting comments
 - Always start comments with `*Comment from Claude*` in italics
 
@@ -215,12 +242,14 @@ gh workflow run "Invoke Canary as Maintainer" --repo smithy-lang/smithy-rs \
 Client changes often show the pattern for server-side implementation
 
 **Configuration Debugging:**
+
 - Server codegen settings go under `"codegen"` not `"codegenConfig"` in smithy-build.json
 - When settings aren't working, check the generated smithy-build.json structure first
 - Settings placement matters - wrong nesting means settings are ignored silently
 - Always verify actual generated configuration matches expectations
 
 **Testing Configuration Settings:**
+
 - Create separate services with different settings to test configuration behavior
 - Use integration tests that verify actual generated code behavior, not just compilation
 - Test both enabled and disabled states to ensure the setting actually controls behavior

codegen-client/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/generators/client/FluentBuilderGenerator.kt

@@ -194,55 +194,55 @@ class FluentBuilderGenerator(
 
         return writable {
             val eventStreamMemberName = symbolProvider.toMemberName(eventStreamMember)
+            val structuredDataParser = codegenContext.protocolImpl?.structuredDataParser()
+            val parser = structuredDataParser?.operationParser(operation)
+
             rustTemplate(
                 """
-                let mut output =
+                let output =
                     #{Operation}::orchestrate(
                         &runtime_plugins,
                         input,
                     )
                     .await?;
 
-                // Converts any error encountered beyond this point into an `SdkError` response error
-                // with an `HttpResponse`. However, since we have already exited the `orchestrate`
-                // function, the original `HttpResponse` is no longer available and cannot be restored.
-                // This means that header information from the original response has been lost.
-                //
-                // Note that the response body would have been consumed by the deserializer
-                // regardless, even if the initial message was hypothetically processed during
-                // the orchestrator's deserialization phase but later resulted in an error.
-                fn response_error(
-                    err: impl #{Into}<#{BoxError}>
-                ) -> #{SdkError}<#{OperationError}, #{HttpResponse}> {
-                    #{SdkError}::response_error(err, #{HttpResponse}::new(
-                        #{StatusCode}::try_from(200).expect("valid successful code"),
-                        #{SdkBody}::empty()))
-                }
-
-                let message = output.$eventStreamMemberName.try_recv_initial_response().await.map_err(response_error)?;
-
-                match message {
-                    #{Some}(_message) => {
-                        #{maybeRecreateOutputWithNonEventStreamMembers:W}
-                        #{Ok}(output)
-                    }
-                    #{None} => #{Ok}(output),
-                }
+                #{maybeReadInitialResponse:W}
+                #{Ok}(output)
                 """,
                 *scope,
-                "maybeRecreateOutputWithNonEventStreamMembers" to
+                "maybeReadInitialResponse" to
                     writable {
-                        val structuredDataParser = codegenContext.protocolImpl?.structuredDataParser()
-                        structuredDataParser?.operationParser(operation)?.also { parser ->
+                        if (parser != null) {
                             rustTemplate(
                                 """
-                                let mut builder = output.into_builder();
-                                builder = #{parser}(
-                                    _message.payload(),
-                                    builder
-                                )
-                                .map_err(response_error)?;
-                                let output = builder.build().map_err(response_error)?;
+                                // Converts any error encountered beyond this point into an `SdkError` response error
+                                // with an `HttpResponse`. However, since we have already exited the `orchestrate`
+                                // function, the original `HttpResponse` is no longer available and cannot be restored.
+                                // This means that header information from the original response has been lost.
+                                //
+                                // Note that the response body would have been consumed by the deserializer
+                                // regardless, even if the initial message was hypothetically processed during
+                                // the orchestrator's deserialization phase but later resulted in an error.
+                                fn response_error(
+                                    err: impl #{Into}<#{BoxError}>
+                                ) -> #{SdkError}<#{OperationError}, #{HttpResponse}> {
+                                    #{SdkError}::response_error(err, #{HttpResponse}::new(
+                                        #{StatusCode}::try_from(200).expect("valid successful code"),
+                                        #{SdkBody}::empty()))
+                                }
+                                let mut output = output;
+
+                                let message = output.$eventStreamMemberName.try_recv_initial_response().await.map_err(response_error)?;
+
+                                if let #{Some}(_message) = message {
+                                    let mut builder = output.into_builder();
+                                    builder = #{parser}(
+                                        _message.payload(),
+                                        builder
+                                    )
+                                    .map_err(response_error)?;
+                                    output = builder.build().map_err(response_error)?;
+                                }
                                 """,
                                 "parser" to parser,
                                 *scope,

codegen-server-test/integration-tests/Cargo.lock

@@ -18,8 +18,9 @@ use hyper::{
 };
 use hyper_util::{client::legacy::Client, rt::TokioExecutor};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use std::time::Duration;
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, Notify};
 use tokio::time::timeout;
 use tokio_stream::wrappers::ReceiverStream;
 
@@ -33,6 +34,7 @@ pub enum RecvError {
 pub struct ManualEventStreamClient {
     message_sender: mpsc::Sender<Message>,
     response_receiver: mpsc::Receiver<Result<Message, RecvError>>,
+    response_ready: Arc<Notify>,
     _handle: tokio::task::JoinHandle<()>,
 }
 
@@ -83,11 +85,17 @@ impl ManualEventStreamClient {
         let body = StreamBody::new(stream);
 
         let request = req.body(body).expect("failed to construct request");
+        let response_ready = Arc::new(Notify::new());
+        let response_ready_clone = response_ready.clone();
         let handle = tokio::spawn(async move {
             let response = timeout(Duration::from_secs(1), client.request(request))
                 .await
                 .expect("timeout making initial request")
                 .expect("failed to make initial contact with server");
+
+            // Notify that the response is ready
+            response_ready_clone.notify_one();
+
             let mut body = response.into_body();
             let mut decoder = MessageFrameDecoder::new();
 
@@ -122,10 +130,16 @@ impl ManualEventStreamClient {
         Ok(Self {
             message_sender,
             response_receiver,
+            response_ready,
             _handle: handle,
         })
     }
 
+    /// Waits for the response stream to be ready (server has accepted the connection).
+    pub async fn wait_for_response_ready(&self) {
+        self.response_ready.notified().await;
+    }
+
     /// Sends a message.
     pub async fn send(&mut self, message: Message) -> Result<(), String> {
         self.message_sender

codegen-server-test/integration-tests/eventstreams/src/lib.rs

@@ -476,6 +476,54 @@ async fn test_streaming_operation_with_initial_data() {
     assert_eq!(harness.server.initial_data(), Some("test-data".to_string()));
 }
 
+/// Test that operations without modeled initial messages don't hang.
+/// This verifies the fix for issue #4435 where recv() would block waiting
+/// for an initial message that was never modeled.
+#[tokio::test]
+async fn test_no_hang_without_initial_message() {
+    let mut harness = TestHarness::new("StreamingOperation").await;
+
+    // Wait for the response to be ready without sending any messages
+    // This should not hang because StreamingOperation has no modeled initial-request
+    harness.client.wait_for_response_ready().await;
+
+    // Now send an event and verify we can receive it
+    harness.send_event("A").await;
+    let resp = harness.expect_message().await;
+    assert_eq!(get_event_type(&resp), "A");
+}
+
+/// Test that operations WITH modeled initial messages DO wait for them.
+/// This verifies that try_recv_initial() blocks and the response doesn't become ready
+/// until the initial-request is sent.
+#[tokio::test]
+async fn test_waits_for_initial_message_when_modeled() {
+    let mut harness = TestHarness::new("StreamingOperationWithInitialData").await;
+
+    // Try to wait for response with a timeout - it should timeout because we haven't sent initial data
+    let result = tokio::time::timeout(
+        tokio::time::Duration::from_millis(100),
+        harness.client.wait_for_response_ready(),
+    )
+    .await;
+    assert!(
+        result.is_err(),
+        "Response should not be ready without initial data"
+    );
+
+    // Now send the initial data
+    harness.send_initial_data("test-data").await;
+
+    // Now the response should become ready
+    harness.client.wait_for_response_ready().await;
+
+    // Send an event and verify it works
+    harness.send_event("A").await;
+    let resp = harness.expect_message().await;
+    assert_eq!(get_event_type(&resp), "A");
+    assert_eq!(harness.server.initial_data(), Some("test-data".to_string()));
+}
+
 /// StreamingOperationWithInitialData has a mandatory initial data field.
 /// If we don't send this field, we'll never hit the handler.
 #[tokio::test]