Chapter 2: Streaming Responses

Project: Streamline

Spark works. Spark also makes the user stare at a spinner for four seconds and then dumps a wall of text on them.

That's not a small aesthetic complaint. It's the difference between an app that feels alive and one that feels broken. The model is producing tokens the whole time — roughly the first one within a few hundred milliseconds — and we are throwing that away by waiting for the complete response before rendering anything.

In this chapter we build Streamline, a chat app where text appears as it's generated. Along the way:

• Server-Sent Events over Ktor, and how the Responses API's event stream is actually shaped.
• Turning a stream of network events into a Flow<StreamEvent> and then into Compose state without recomposing the world on every token.
• Multi-turn conversation state — client-side history versus server-side previous_response_id.
• Cancellation, which is not a nicety: an abandoned stream you don't cancel is output tokens you pay for and never show anyone.

---

2.1 What the stream actually looks like

Add "stream": true to the same POST /v1/responses request from Chapter 1 and the response becomes text/event-stream instead of JSON. Each event is a event: line and a data: line:

event: response.created
data: {"type":"response.created","response":{"id":"resp_abc","status":"in_progress"}}

event: response.output_item.added
data: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rs_1"}}

event: response.output_item.added
data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","id":"msg_1","role":"assistant"}}

event: response.output_text.delta
data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":1,"delta":"Sun"}

event: response.output_text.delta
data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":1,"delta":"light"}

event: response.output_text.done
data: {"type":"response.output_text.done","item_id":"msg_1","text":"Sunlight scatters..."}

event: response.completed
data: {"type":"response.completed","response":{"id":"resp_abc","status":"completed","usage":{"input_tokens":29,"output_tokens":17,"total_tokens":46}}}

There are dozens of event types. You need four:

Event	Meaning	What you do
response.output_text.delta	A chunk of visible text	Append it
response.completed	Done, with final usage	Finalize, record cost
response.incomplete	Stopped early (token cap, filter)	Show what you have, explain why
response.failed / error	Something broke mid-stream	Map to OpenAiError

Everything else — reasoning item lifecycle, content part boundaries, annotation events — you can ignore in a text app, and you must ignore gracefully, because OpenAI adds new event types without warning. Match on the ones you know; drop the rest silently.

The reasoning trap, again. Notice that the first output_item.added is a reasoning item, and it produces no output_text.delta events at all. With reasoning.effort set higher than none, the model can think for seconds before the first visible token arrives. Your "streaming" UI will show a blank screen for that entire period and users will think it's frozen. This is the single strongest argument for effort: "none" or "low" in interactive mobile features — see §2.6.

---

2.2 Streaming with Ktor

Ktor's SSE plugin does the framing for you. Add it:

# libs.versions.toml
ktor-client-sse = { module = "io.ktor:ktor-client-sse", version.ref = "ktor" }

// core/openai/di/OpenAiModule.kt — add to the HttpClient block
install(SSE)

Now the streaming call. This lives beside respond() in OpenAiClient:

// core/openai/src/main/kotlin/dev/spark/core/openai/OpenAiClient.kt
sealed interface StreamEvent {
    data class Delta(val text: String) : StreamEvent
    data class Completed(val usage: TokenUsage, val responseId: String) : StreamEvent
    data class Incomplete(val reason: String) : StreamEvent
}

fun respondStreaming(
    input: List<InputMessage>,
    instructions: String? = null,
    model: String = Models.DEFAULT,
    reasoningEffort: String = "none",
    previousResponseId: String? = null,
): Flow<StreamEvent> = flow {

    http.sse(
        urlString = "${config.baseUrl}/responses",
        request = {
            method = HttpMethod.Post
            contentType(ContentType.Application.Json)
            setBody(
                StreamingRequest(
                    model = model,
                    input = input,
                    instructions = instructions,
                    reasoning = Reasoning(reasoningEffort),
                    previousResponseId = previousResponseId,
                    store = previousResponseId != null,
                    stream = true,
                )
            )
        }
    ) {
        incoming.collect { sse ->
            val data = sse.data ?: return@collect
            if (data == "[DONE]") return@collect

            val event = json.decodeFromString<StreamEnvelope>(data)

            when (event.type) {
                "response.output_text.delta" ->
                    event.delta?.let { emit(StreamEvent.Delta(it)) }

                "response.completed" -> {
                    val r = event.response ?: return@collect
                    emit(
                        StreamEvent.Completed(
                            usage = TokenUsage(
                                inputTokens = r.usage?.inputTokens ?: 0,
                                outputTokens = r.usage?.outputTokens ?: 0,
                            ),
                            responseId = r.id,
                        )
                    )
                }

                "response.incomplete" ->
                    emit(StreamEvent.Incomplete(
                        event.response?.incompleteDetails?.reason ?: "unknown"
                    ))

                "response.failed", "error" ->
                    throw OpenAiError.Server(event.response?.error?.message ?: "Stream failed")

                else -> Unit   // unknown event types are not errors
            }
        }
    }
}.flowOn(Dispatchers.IO)
    .catch { e ->
        when (e) {
            is CancellationException -> throw e
            is OpenAiError -> throw e
            is IOException -> throw OpenAiError.Network(e)
            else -> throw OpenAiError.Server(e.message ?: "Stream error")
        }
    }

With the DTOs:

@Serializable
data class StreamEnvelope(
    val type: String,
    val delta: String? = null,
    val response: ResponseResult? = null,
)

@Serializable
data class StreamingRequest(
    val model: String,
    val input: List<InputMessage>,
    val instructions: String? = null,
    val reasoning: Reasoning? = null,
    @SerialName("previous_response_id") val previousResponseId: String? = null,
    val store: Boolean = false,
    val stream: Boolean = true,
)

@Serializable
data class InputMessage(
    val role: String,          // "user" | "assistant"
    val content: String,
)

Note else -> Unit. That branch is the whole reliability story. A when with an else -> error("unknown event") will crash your chat screen the day OpenAI ships a new event type, which they do routinely.

---

2.3 Cancellation, and why it costs money

The user starts a long answer, reads two lines, decides it's wrong, and hits back. What happens?

If you do nothing: the coroutine is cancelled when the ViewModel clears, the SSE connection closes, and OpenAI stops generating. Good — viewModelScope handles that for free.

If you collect the flow in the wrong scope — say a GlobalScope, or an Application-scoped repository that outlives the screen — the stream keeps running. The model keeps generating. You keep getting billed for output tokens that will never be displayed. This is a real bill, on a real invoice, for text no human read.

Two rules:

1. Collect stream flows in viewModelScope or a scope tied to the screen. Never in a scope that outlives the UI, unless you have a specific reason (background generation — Chapter 6 — where you do want it to survive, and you handle that explicitly with WorkManager).
2. Never swallow CancellationException. The catch block above rethrows it explicitly before mapping anything else. A catch (e: Exception) that turns cancellation into an error state will make your app show "Something went wrong" every time a user navigates away — a bug that is maddening to track down because it only reproduces when you leave the screen.

---

2.4 The ViewModel

@HiltViewModel
class ChatViewModel @Inject constructor(
    private val client: OpenAiClient,
) : ViewModel() {

    private val _state = MutableStateFlow(ChatUiState())
    val state: StateFlow<ChatUiState> = _state.asStateFlow()

    private var streamJob: Job? = null

    fun onSend() {
        val text = _state.value.draft.trim()
        if (text.isEmpty() || _state.value.isStreaming) return

        val userMessage = Message(Role.User, text)

        _state.update {
            it.copy(
                draft = "",
                messages = it.messages + userMessage + Message(Role.Assistant, ""),
                isStreaming = true,
                error = null,
            )
        }

        streamJob = viewModelScope.launch {
            val history = _state.value.messages
                .dropLast(1)                         // drop the empty placeholder
                .map { InputMessage(it.role.wire, it.text) }

            client.respondStreaming(input = history, instructions = SYSTEM)
                .onEach { event ->
                    when (event) {
                        is StreamEvent.Delta -> appendToLast(event.text)
                        is StreamEvent.Completed -> _state.update {
                            it.copy(isStreaming = false, totalCostUsd = it.totalCostUsd + estimateCostUsd(event.usage))
                        }
                        is StreamEvent.Incomplete -> _state.update {
                            it.copy(isStreaming = false, error = "Response was cut short (${event.reason}).")
                        }
                    }
                }
                .catch { e ->
                    _state.update {
                        it.copy(isStreaming = false, error = e.toUserMessage())
                    }
                }
                .collect()
        }
    }

    fun onStop() {
        streamJob?.cancel()
        _state.update { it.copy(isStreaming = false) }
    }

    private fun appendToLast(delta: String) {
        _state.update { state ->
            val messages = state.messages.toMutableList()
            val last = messages.last()
            messages[messages.lastIndex] = last.copy(text = last.text + delta)
            state.copy(messages = messages)
        }
    }
}

Two design points.

The empty assistant placeholder. We append a Message(Assistant, "") before the stream starts. That gives the UI something to grow, so the bubble appears immediately with a cursor in it and then fills. Adding the bubble on the first delta instead produces a visible pop that feels worse than a spinner.

An explicit Stop button. onStop() cancels the job. Users want this — every serious chat UI has it — and it directly saves you money. Wire it to replace the Send button while isStreaming is true.

---

2.5 Rendering a stream without melting the frame budget

Here's the performance problem nobody warns you about. Each delta triggers a StateFlow emission, which recomposes the message list. At 30–60 tokens per second, that's 30–60 recompositions of a LazyColumn per second. If your item composables aren't stable, you drop frames badly on mid-range devices.

Three fixes, in order of importance:

1. Key your list, and make items stable.

LazyColumn(state = listState) {
    items(
        items = state.messages,
        key = { it.id },                    // stable identity → no full rebuild
    ) { message ->
        MessageBubble(message)
    }
}

Give Message an id (a UUID assigned on creation, not the index). Without a key, LazyColumn re-creates every item's composition when the list changes.

2. Only the last bubble is changing. With a proper key, Compose knows only the last item's text changed, so only that bubble recomposes. Verify it: turn on Layout Inspector's recomposition counts and confirm the other bubbles stay at zero. If they don't, something in your Message class is unstable — usually a List field, which Compose can't prove immutable. Mark the data class with @Immutable or wrap collections in ImmutableList from kotlinx.collections.immutable.

3. Autoscroll, cheaply.

LaunchedEffect(state.messages.size, state.isStreaming) {
    if (state.isStreaming) {
        listState.animateScrollToItem(state.messages.lastIndex)
    }
}

Keying on messages.size rather than the last message's text means you scroll once per new message, not once per token. Scrolling on every token fights the user's own scrolling and burns frames.

If you want to be fancy: throttle deltas. Buffer them in the client and emit every 50ms instead of on every token:

.conflate()   // simplest option: drop intermediate emissions if the UI is behind

I'd reach for that only if you measure a problem. Correct keys usually solve it.

---

2.6 Conversation state: two strategies

Multi-turn chat means the model needs to see history. There are two ways, and they trade off differently.

Client-side history. Send the whole array of messages every turn, as input. This is what the code above does.

• Full control; nothing stored on OpenAI's side (store: false).
• You pay for the entire history as input tokens on every single turn. A 20-turn conversation re-sends 19 turns of context each time. Costs grow quadratically with conversation length.
• Prompt caching (Chapter 14) claws a lot of that back, because the prefix is identical each turn.

Server-side state. Send only the new message, plus previous_response_id. OpenAI keeps the thread.

• Much smaller requests. Simpler client code.
• Requires store: true — the conversation is retained on OpenAI's infrastructure. That's the privacy decision we flagged in Chapter 1 §1.9, and you must make it consciously.
• You lose the ability to edit or prune history client-side, which matters when you want to drop a bad turn or inject a summary.

My recommendation for mobile: client-side history plus caching, unless conversations are long-running and you've read the data terms and are comfortable. The privacy posture is cleaner and you keep the ability to trim. But make it a config flag, not a hardcoded assumption — previousResponseId is already a parameter in the client above.

Trimming

Whatever you choose, conversations grow. A naive app eventually sends 200k tokens of history and either bankrupts you or hits the context ceiling. The simplest workable policy:

private fun List<Message>.trimmed(maxTurns: Int = 20): List<Message> =
    if (size <= maxTurns) this
    else takeLast(maxTurns)

Better, and worth the extra call: when history exceeds a threshold, ask the model to summarize the older half, replace it with the summary, and continue. That's a fifteen-line addition and it's how every production chat app works. It is Exercise 3.

---

2.7 Exercises

1. Add the Stop button and prove it works: start a long generation, hit Stop, and confirm in your token meter that the final usage never arrives (you'll need to log partial output token counts to see the saving).
2. Measure time-to-first-token across reasoning.effort of none, low, and medium. Log the milliseconds between request start and first Delta. The gap between none and medium will tell you more about mobile UX than any blog post.
3. Implement summarization-based trimming. When history exceeds 20 messages, summarize the oldest 10 into a single system-level note and continue. Compare token counts before and after over a 40-turn conversation.
4. Break the stream. Kill the network mid-generation. Your UI should keep the partial text on screen, show a clear error, and offer retry — not blank the bubble.

---

2.8 What's next

Streaming makes a chatbot feel good. But a chatbot is the least interesting thing you can build with a language model in a mobile app — it offloads all the work onto the user's typing and all the interpretation onto their reading.

Chapter 3 changes that. Instead of asking the model for prose, we ask it for structured data conforming to a JSON schema we define, map it straight to a Kotlin data class, and render it as real UI: cards, chips, form fields, a chart. That's the move that turns "AI feature" from a chat box into an actual product feature.