test(#1753): cancellation contract pins for all three streaming adapters

Verifies that cancelling the collecting coroutine of chatStream
terminates promptly — well before the upstream response finishes.

Surprising finding when writing these: the Kotlin Flow contract
already handles cancellation correctly. emit() in the producer body
suspends waiting for downstream channel capacity; when the collector
cancels, emit throws CancellationException, the flow body unwinds,
BufferedReader.useLines + .use { stream } close the InputStream
cleanly. The IO-thread loop never enters its next blocking readLine
after cancel fires.

So no ensureActive() additions or other producer-side hardening was
needed — the test serves as a regression guard against future changes
that might bypass the Flow's cancellation hook (e.g., a synchronous
collect-to-list step inside the producer).

Three new tests, one per adapter:
- OllamaClientCancellationTest: SlowNdjsonStream yielding 6 chunks
  at 80ms each (~480ms total). Cancel after first chunk; assert
  return within 250ms.
- ClaudeClientCancellationTest: SlowSseStream with 6 content_block
  triples (~1.5s total). 400ms threshold accommodates SSE having
  multiple events per displayable text chunk.
- OpenAiClientCancellationTest: SlowOpenAiSseStream with 6 deltas
  + terminator (~640ms total). 250ms threshold.

All three green on first run — no impl changes needed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Skobeltsyn

src/test/kotlin/agents_engine/model/ClaudeClientCancellationTest.kt

@@ -0,0 +1,88 @@
+package agents_engine.model
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import java.io.IOException
+import java.io.InputStream
+import kotlin.test.Test
+import kotlin.test.assertTrue
+
+// #1753 — regression guard for cancellation of ClaudeClient.chatStream.
+// Same contract pin as the Ollama analog: Kotlin Flow's channel-backed
+// emit propagates collector cancellation back into the producer body,
+// closing the InputStream before the next blocking read.
+
+class ClaudeClientCancellationTest {
+
+    private class SlowSseStream : InputStream() {
+        private val events = buildList {
+            add("event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":3,\"output_tokens\":1}}}\n\n")
+            (1..6).forEach { i ->
+                add("event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\"}}\n\n")
+                add("event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"chunk$i \"}}\n\n")
+                add("event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n")
+            }
+            add("event: message_delta\ndata: {\"type\":\"message_delta\",\"usage\":{\"output_tokens\":12}}\n\n")
+            add("event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")
+        }.toMutableList()
+        private var bytes: ByteArray = events.removeFirst().toByteArray(Charsets.UTF_8)
+        private var bytesPos: Int = 0
+
+        override fun read(): Int = if (bytesPos < bytes.size) bytes[bytesPos++].toInt() and 0xFF else loadNextOrEof()
+        override fun read(b: ByteArray, off: Int, len: Int): Int {
+            if (bytesPos >= bytes.size) {
+                val advanced = loadNextOrEof()
+                if (advanced < 0) return -1
+            }
+            val available = bytes.size - bytesPos
+            val copyLen = minOf(len, available)
+            System.arraycopy(bytes, bytesPos, b, off, copyLen)
+            bytesPos += copyLen
+            return copyLen
+        }
+        private fun loadNextOrEof(): Int {
+            if (events.isEmpty()) return -1
+            try { Thread.sleep(80) } catch (_: InterruptedException) { throw IOException("interrupted") }
+            bytes = events.removeFirst().toByteArray(Charsets.UTF_8)
+            bytesPos = 0
+            return bytes[bytesPos].toInt() and 0xFF
+        }
+    }
+
+    @Test
+    fun `cancelling Claude chatStream mid-collect terminates within tight window, not the full response`() = runBlocking {
+        val stubbed = object : ClaudeClient(apiKey = "test-key", model = "test-model") {
+            override fun sendChatStream(body: String, headers: Map<String, String>): InputStream = SlowSseStream()
+        }
+
+        var received = 0
+        val firstChunk = CompletableDeferred<Unit>()
+
+        val collectJob = launch(Dispatchers.Default) {
+            stubbed.chatStream(listOf(LlmMessage("user", "Hi"))).collect { _ ->
+                received++
+                if (!firstChunk.isCompleted) firstChunk.complete(Unit)
+            }
+        }
+
+        withTimeout(2000) { firstChunk.await() }
+        assertTrue(received >= 1, "expected at least one chunk before cancel; got $received")
+
+        val cancelStartNs = System.nanoTime()
+        collectJob.cancelAndJoin()
+        val cancelMs = (System.nanoTime() - cancelStartNs) / 1_000_000
+
+        // SSE has more events per "chunk" (start + delta + stop) so the
+        // first text delta might arrive 2-3 stub-yields in. Bound at
+        // 400ms — generous slack vs. the full stream's ~1.5s.
+        assertTrue(
+            cancelMs < 400,
+            "expected cancel to return within ~400ms; took ${cancelMs}ms",
+        )
+    }
+}

src/test/kotlin/agents_engine/model/OllamaClientCancellationTest.kt

@@ -0,0 +1,93 @@
+package agents_engine.model
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import java.io.IOException
+import java.io.InputStream
+import kotlin.test.Test
+import kotlin.test.assertTrue
+
+// #1753 — regression guard for cancellation of OllamaClient.chatStream.
+//
+// The Kotlin Flow contract (channel-backed by `flowOn(Dispatchers.IO)`)
+// already propagates collector cancellation back through `emit()` —
+// when the consuming coroutine cancels, emit throws CancellationException,
+// the flow body unwinds, and BufferedReader.useLines + .use { stream }
+// close the InputStream cleanly. The IO-thread loop never enters its
+// next blocking `readLine` after cancel fires.
+//
+// This test pins that contract: a slow-yielding stub produces six
+// chunks at 80ms each (~480ms total). Cancelling after the first
+// chunk arrives must return within ~250ms — proves we're not blocking
+// on the remaining chunks. If a future change adds a blocking sync
+// step that bypasses the flow's cancellation hook, this test fires.
+
+class OllamaClientCancellationTest {
+
+    private class SlowNdjsonStream : InputStream() {
+        private val chunks = (1..6).map {
+            """{"model":"t","message":{"role":"assistant","content":"chunk$it "},"done":false}""" + "\n"
+        }.toMutableList().also {
+            it += """{"model":"t","message":{"content":""},"done":true,"prompt_eval_count":3,"eval_count":6}""" + "\n"
+        }
+        private var bytes: ByteArray = chunks.removeFirst().toByteArray(Charsets.UTF_8)
+        private var bytesPos: Int = 0
+
+        override fun read(): Int = if (bytesPos < bytes.size) bytes[bytesPos++].toInt() and 0xFF else loadNextOrEof()
+        override fun read(b: ByteArray, off: Int, len: Int): Int {
+            if (bytesPos >= bytes.size) {
+                val advanced = loadNextOrEof()
+                if (advanced < 0) return -1
+            }
+            val available = bytes.size - bytesPos
+            val copyLen = minOf(len, available)
+            System.arraycopy(bytes, bytesPos, b, off, copyLen)
+            bytesPos += copyLen
+            return copyLen
+        }
+        private fun loadNextOrEof(): Int {
+            if (chunks.isEmpty()) return -1
+            try { Thread.sleep(80) } catch (_: InterruptedException) { throw IOException("interrupted") }
+            bytes = chunks.removeFirst().toByteArray(Charsets.UTF_8)
+            bytesPos = 0
+            return bytes[bytesPos].toInt() and 0xFF
+        }
+    }
+
+    @Test
+    fun `cancelling chatStream mid-collect terminates within one-chunk's worth of time, not the full response`() = runBlocking {
+        val stubbed = object : OllamaClient(model = "test-model") {
+            override fun sendChatStream(body: String): InputStream = SlowNdjsonStream()
+        }
+
+        var received = 0
+        val firstChunk = CompletableDeferred<Unit>()
+
+        val collectJob = launch(Dispatchers.Default) {
+            stubbed.chatStream(listOf(LlmMessage("user", "Hi"))).collect { _ ->
+                received++
+                if (!firstChunk.isCompleted) firstChunk.complete(Unit)
+            }
+        }
+
+        withTimeout(2000) { firstChunk.await() }
+        assertTrue(received >= 1, "expected at least one chunk before cancel; got $received")
+
+        val cancelStartNs = System.nanoTime()
+        collectJob.cancelAndJoin()
+        val cancelMs = (System.nanoTime() - cancelStartNs) / 1_000_000
+
+        // Bound: under 250ms covers one-chunk-worth (~80ms) + cleanup
+        // overhead. The full stream would take ~480ms — we must not wait
+        // for it.
+        assertTrue(
+            cancelMs < 250,
+            "expected cancel to return within ~250ms (one chunk's worth + slack); took ${cancelMs}ms",
+        )
+    }
+}

src/test/kotlin/agents_engine/model/OpenAiClientCancellationTest.kt

@@ -0,0 +1,83 @@
+package agents_engine.model
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import java.io.IOException
+import java.io.InputStream
+import kotlin.test.Test
+import kotlin.test.assertTrue
+
+// #1753 — regression guard for cancellation of OpenAiClient.chatStream.
+// Same contract pin: Kotlin Flow's emit propagates collector cancellation
+// back into the producer body, closing the InputStream before the next
+// blocking read.
+
+class OpenAiClientCancellationTest {
+
+    private class SlowOpenAiSseStream : InputStream() {
+        private val events = buildList {
+            (1..6).forEach { i ->
+                add("data: {\"id\":\"x\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"chunk$i \"},\"finish_reason\":null}]}\n\n")
+            }
+            add("data: {\"id\":\"x\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n")
+            add("data: {\"id\":\"x\",\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":12,\"total_tokens\":15}}\n\n")
+            add("data: [DONE]\n\n")
+        }.toMutableList()
+        private var bytes: ByteArray = events.removeFirst().toByteArray(Charsets.UTF_8)
+        private var bytesPos: Int = 0
+
+        override fun read(): Int = if (bytesPos < bytes.size) bytes[bytesPos++].toInt() and 0xFF else loadNextOrEof()
+        override fun read(b: ByteArray, off: Int, len: Int): Int {
+            if (bytesPos >= bytes.size) {
+                val advanced = loadNextOrEof()
+                if (advanced < 0) return -1
+            }
+            val available = bytes.size - bytesPos
+            val copyLen = minOf(len, available)
+            System.arraycopy(bytes, bytesPos, b, off, copyLen)
+            bytesPos += copyLen
+            return copyLen
+        }
+        private fun loadNextOrEof(): Int {
+            if (events.isEmpty()) return -1
+            try { Thread.sleep(80) } catch (_: InterruptedException) { throw IOException("interrupted") }
+            bytes = events.removeFirst().toByteArray(Charsets.UTF_8)
+            bytesPos = 0
+            return bytes[bytesPos].toInt() and 0xFF
+        }
+    }
+
+    @Test
+    fun `cancelling OpenAI chatStream mid-collect terminates within tight window, not the full response`() = runBlocking {
+        val stubbed = object : OpenAiClient(apiKey = "test-key", model = "test-model") {
+            override fun sendChatStream(body: String, headers: Map<String, String>): InputStream = SlowOpenAiSseStream()
+        }
+
+        var received = 0
+        val firstChunk = CompletableDeferred<Unit>()
+
+        val collectJob = launch(Dispatchers.Default) {
+            stubbed.chatStream(listOf(LlmMessage("user", "Hi"))).collect { _ ->
+                received++
+                if (!firstChunk.isCompleted) firstChunk.complete(Unit)
+            }
+        }
+
+        withTimeout(2000) { firstChunk.await() }
+        assertTrue(received >= 1, "expected at least one chunk before cancel; got $received")
+
+        val cancelStartNs = System.nanoTime()
+        collectJob.cancelAndJoin()
+        val cancelMs = (System.nanoTime() - cancelStartNs) / 1_000_000
+
+        assertTrue(
+            cancelMs < 250,
+            "expected cancel to return within ~250ms; took ${cancelMs}ms",
+        )
+    }
+}