fix hang caused by StreamableHttpClientTransport (#226) (#227)

Re-throw exceptions in `StreamableHttpClientTransport` so `Protocol`
doesn't hang indefinitely.


## Motivation and Context
Fixes the issue described in #226

## How Has This Been Tested?
I've verified that my test case reproduces the error without my fix, and
verified with breakpoints that the hang occurred where I described it.

## Breaking Changes
None

## Types of changes
<!-- What types of changes does your code introduce? Put an `x` in all
the boxes that apply: -->
- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update

## Checklist
<!-- Go over all the following points, and put an `x` in all the boxes
that apply. -->
- [x] I have read the [MCP
Documentation](https://modelcontextprotocol.io)
- [x] My code follows the repository's style guidelines
- [x] New and existing tests pass locally
- [x] I have added appropriate error handling
- [ ] I have added or updated documentation as needed

## Additional context
I am unsure about the dispatcher I used for the real-time timeout here.
I could not get the test to work otherwise, as `kotlinx.coroutines.test`
was always tripping my timeout, with or without the fix. This is
seemingly unrelated to the test case though, as removing my timeout
results in the [timeout in
Protocol](6bae9874c0/kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/Protocol.kt (L454))
being tripped, even though that's not the point of the indefinite hang.

kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt

@@ -142,7 +142,10 @@ public class StreamableHttpClientTransport(
             ContentType.Application.Json -> response.bodyAsText().takeIf { it.isNotEmpty() }?.let { json ->
                 runCatching { McpJson.decodeFromString<JSONRPCMessage>(json) }
                     .onSuccess { _onMessage(it) }
-                    .onFailure(_onError)
+                    .onFailure {
+                        _onError(it)
+                        throw it
+                    }
             }
 
             ContentType.Text.EventStream -> handleInlineSse(
@@ -313,7 +316,10 @@ public class StreamableHttpClientTransport(
                             _onMessage(msg)
                         }
                     }
-                    .onFailure(_onError)
+                    .onFailure {
+                        _onError(it)
+                        throw it
+                    }
             }
             // reset
             id = null

kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransportTest.kt

@@ -12,13 +12,18 @@ import io.ktor.http.HttpStatusCode
 import io.ktor.http.content.TextContent
 import io.ktor.http.headersOf
 import io.ktor.utils.io.ByteReadChannel
+import io.modelcontextprotocol.kotlin.sdk.Implementation
 import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
 import io.modelcontextprotocol.kotlin.sdk.JSONRPCNotification
 import io.modelcontextprotocol.kotlin.sdk.JSONRPCRequest
 import io.modelcontextprotocol.kotlin.sdk.RequestId
 import io.modelcontextprotocol.kotlin.sdk.shared.McpJson
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.JsonPrimitive
 import kotlinx.serialization.json.buildJsonObject
@@ -27,6 +32,7 @@ import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNull
 import kotlin.test.assertTrue
+import kotlin.test.fail
 import kotlin.time.Duration.Companion.seconds
 
 class StreamableHttpClientTransportTest {
@@ -380,4 +386,46 @@ class StreamableHttpClientTransportTest {
         assertEquals("resume-100", resumptionTokenReceived)
         transport.close()
     }
+
+    @Test
+    fun testClientConnectWithInvalidJson() = runTest {
+        // Transport under test: respond with invalid JSON for the initialize request
+        val transport = createTransport { _ ->
+            respond(
+                "this is not valid json",
+                status = HttpStatusCode.OK,
+                headers = headersOf(HttpHeaders.ContentType, ContentType.Application.Json.toString()),
+            )
+        }
+
+        val client = Client(
+            clientInfo = Implementation(
+                name = "test-client",
+                version = "1.0",
+            ),
+        )
+
+        runCatching {
+            // Real time-keeping is needed; otherwise Protocol will always throw TimeoutCancellationException in tests
+            withContext(Dispatchers.Default.limitedParallelism(1)) {
+                withTimeout(5.seconds) {
+                    client.connect(transport)
+                }
+            }
+        }.onSuccess {
+            fail("Expected client.connect to fail on invalid JSON response")
+        }.onFailure { e ->
+            when (e) {
+                is TimeoutCancellationException -> fail("Client connect caused a hang", e)
+
+                is IllegalStateException -> {
+                    // Expected behavior: connect finishes and fails with an exception.
+                }
+
+                else -> fail("Unexpected exception during client.connect", e)
+            }
+        }.also {
+            transport.close()
+        }
+    }
 }