fix(retry): handle UnexpectedEof and connection closed errors as retryable

Detect two additional connection-closed-by-peer scenarios in the retry
logic that were previously not treated as retryable:

1. TLS (rustls): peer closes connection without sending a TLS close_notify
   alert, surfaced as std::io::Error(UnexpectedEof) in the source chain.

2. Plain HTTP / hyper: peer closes the TCP connection before sending a
   complete HTTP response, surfaced as hyper's IncompleteMessage error
   with the message 'connection closed before message completed'.

Both cases are safe to retry because the request was never acknowledged
by the server. This fixes the error observed against the ChatGPT codex
endpoint:

  POST https://chatgpt.com/backend-api/codex/responses
  peer closed connection without sending TLS close_notify

Adds has_unexpected_eof() helper that walks the reqwest::Error source
chain to detect either condition, and two async tests that simulate a
peer closing the TCP connection to verify the detection and retry path.

Co-Authored-By: ForgeCode <noreply@forgecode.dev>

Author: tusharmath

crates/forge_repo/Cargo.toml

@@ -21,6 +21,7 @@ tokio.workspace = true
 tokio-stream.workspace = true
 tracing.workspace = true
 reqwest.workspace = true
+hyper = { version = "1" }
 url.workspace = true
 bytes.workspace = true
 strum.workspace = true

crates/forge_repo/src/provider/retry.rs

@@ -95,7 +95,32 @@ fn is_empty_error(error: &anyhow::Error) -> bool {
 fn is_req_transport_error(error: &anyhow::Error) -> bool {
     error
         .downcast_ref::<reqwest::Error>()
-        .is_some_and(|e| e.is_timeout() || e.is_connect())
+        .is_some_and(|e| e.is_timeout() || e.is_connect() || is_incomplete_message(e))
+}
+
+/// Returns true if the reqwest error was caused by a peer closing the
+/// connection before the response was complete. This covers:
+/// - TLS peers that omit the close_notify alert (rustls surfaces this as
+///   `io::Error(UnexpectedEof)`)
+/// - HTTP peers that drop the TCP connection mid-response (hyper surfaces
+///   this as `hyper::Error::is_incomplete_message()`)
+fn is_incomplete_message(error: &reqwest::Error) -> bool {
+    use std::error::Error as StdError;
+    let mut source: Option<&dyn StdError> = error.source();
+    while let Some(err) = source {
+        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
+            if io_err.kind() == std::io::ErrorKind::UnexpectedEof {
+                return true;
+            }
+        }
+        if let Some(hyper_err) = err.downcast_ref::<hyper::Error>() {
+            if hyper_err.is_incomplete_message() {
+                return true;
+            }
+        }
+        source = err.source();
+    }
+    false
 }
 
 fn is_event_transport_error(error: &anyhow::Error) -> bool {
@@ -293,4 +318,24 @@ mod tests {
         assert!(get_req_status_code(&error).is_none());
         assert!(get_event_req_status_code(&error).is_none());
     }
+
+    #[tokio::test]
+    async fn test_incomplete_message_is_retryable() {
+        use tokio::net::TcpListener;
+
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            let (_socket, _) = listener.accept().await.unwrap();
+        });
+
+        let req_err = reqwest::Client::new()
+            .get(format!("http://{addr}"))
+            .send()
+            .await
+            .unwrap_err();
+
+        let retry_config = fixture_retry_config(vec![]);
+        assert!(is_retryable(into_retry(req_err.into(), &retry_config)));
+    }
 }