diff --git a/openai-java-core/src/main/kotlin/com/openai/core/http/AsyncStreamResponse.kt b/openai-java-core/src/main/kotlin/com/openai/core/http/AsyncStreamResponse.kt
index 019dc6aff..8111d1db0 100644
--- a/openai-java-core/src/main/kotlin/com/openai/core/http/AsyncStreamResponse.kt
+++ b/openai-java-core/src/main/kotlin/com/openai/core/http/AsyncStreamResponse.kt
@@ -91,6 +91,15 @@ internal fun <T> CompletableFuture<StreamResponse<T>>.toAsync(streamHandlerExecu
                     else "Cannot subscribe after the response is closed"
                 }
 
+                val guardedExecutor = Executor { task ->
+                    try {
+                        executor.execute(task)
+                    } catch (error: Throwable) {
+                        if (onCompleteFuture.completeExceptionally(error)) close()
+                        throw error
+                    }
+                }
+
                 this@toAsync.whenCompleteAsync(
                     { streamResponse, futureError ->
                         if (state.get() == State.CLOSED) {
@@ -129,7 +138,7 @@ internal fun <T> CompletableFuture<StreamResponse<T>>.toAsync(streamHandlerExecu
                             }
                         }
                     },
-                    executor,
+                    guardedExecutor,
                 )
             }
 
diff --git a/openai-java-core/src/test/kotlin/com/openai/core/http/AsyncStreamResponseTest.kt b/openai-java-core/src/test/kotlin/com/openai/core/http/AsyncStreamResponseTest.kt
index 976bcd6da..8014d1311 100644
--- a/openai-java-core/src/test/kotlin/com/openai/core/http/AsyncStreamResponseTest.kt
+++ b/openai-java-core/src/test/kotlin/com/openai/core/http/AsyncStreamResponseTest.kt
@@ -2,7 +2,10 @@ package com.openai.core.http
 
 import java.util.*
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ExecutionException
 import java.util.concurrent.Executor
+import java.util.concurrent.RejectedExecutionException
+import java.util.concurrent.TimeUnit
 import java.util.stream.Stream
 import kotlin.streams.asStream
 import org.assertj.core.api.Assertions.assertThat
@@ -257,6 +260,24 @@ internal class AsyncStreamResponseTest {
         verify(streamResponse, times(1)).close()
     }
 
+    @Test
+    fun subscribe_whenExecutorRejects_completesAndCloses() {
+        val future = CompletableFuture.completedFuture(streamResponse)
+        val asyncStreamResponse = future.toAsync(executor)
+        val rejected = RejectedExecutionException("executor rejected")
+        val rejectingExecutor = Executor { throw rejected }
+
+        asyncStreamResponse.subscribe(handler, rejectingExecutor)
+
+        val completionError = catchThrowable {
+            asyncStreamResponse.onCompleteFuture().get(100, TimeUnit.MILLISECONDS)
+        }
+        assertThat(completionError).isInstanceOf(ExecutionException::class.java).hasCause(rejected)
+        verify(streamResponse, times(1)).close()
+        verify(handler, never()).onNext(any())
+        verify(handler, never()).onComplete(any())
+    }
+
     @Test
     fun close_whenFutureErrors_doesNothing() {
         val future = CompletableFuture<StreamResponse<String>>()