From 5b1cf4a6a3b4b126241f08ac561306a3f723bede Mon Sep 17 00:00:00 2001 From: Kyle Carberry Date: Tue, 3 Mar 2026 16:27:59 -0500 Subject: [PATCH] fix(chatd): start stream buffering before publishing running status (#22571) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem There is a race condition in the chat stream reconnect path. When a client connects (or reconnects) to `/stream`, sometimes they only see a `status: running` event but never receive any `message_part` events — the stream appears stuck. ## Root Cause In `processChat`, the sequence is: 1. `publishStatus(running)` — broadcasts `status: running` to all subscribers and via pubsub. 2. `runChat()` is called. 3. Inside `runChat`, there's significant setup work (model resolution, DB queries, title generation, prompt building, instruction resolution). 4. Only **after** all that setup does `runChat` set `buffering = true` on the stream state. If a client connects to `/stream` between steps 1 and 4: - `Subscribe()` reads `chat.Status == running` from the DB, so it includes `status: running` in the snapshot. - But `buffering` is still `false`, so `subscribeToStream` returns an **empty** local snapshot (no message_parts). - `publishToStream` **drops** all `message_part` events when `buffering` is false. - Result: client sees `running` but never gets any streaming content. ## Fix Move the `buffering = true` setup (and its deferred cleanup) from `runChat` into `processChat`, right before `publishStatus(running)`. This guarantees the buffer is active before any subscriber can observe `status: running`, so: - The snapshot always includes any in-flight `message_part` events. - `publishToStream` never drops parts because buffering is already on. --- coderd/chatd/chatd.go | 42 ++++++++++++++++++++++++++---------------- 1 file changed, 26 insertions(+), 16 deletions(-) diff --git a/coderd/chatd/chatd.go b/coderd/chatd/chatd.go index 4fff70af4f..31748e47ce 100644 --- a/coderd/chatd/chatd.go +++ b/coderd/chatd/chatd.go @@ -1674,6 +1674,27 @@ func (p *Server) processChat(ctx context.Context, chat database.Chat) { } }() + // Start buffering stream events BEFORE publishing the running + // status. This closes a race where a subscriber sees + // status=running but misses message_part events because + // buffering hasn't started yet — the subscriber gets an empty + // snapshot and publishToStream drops message_parts while + // buffering is false. + p.streamMu.Lock() + startState := p.streamStateLocked(chat.ID) + startState.buffer = nil + startState.buffering = true + p.streamMu.Unlock() + defer func() { + p.streamMu.Lock() + if stopState, ok := p.chatStreams[chat.ID]; ok { + stopState.buffer = nil + stopState.buffering = false + p.cleanupStreamIfIdleLocked(chat.ID, stopState) + } + p.streamMu.Unlock() + }() + p.publishStatus(chat.ID, database.ChatStatusRunning, uuid.NullUUID{ UUID: p.workerID, Valid: true, @@ -1920,22 +1941,11 @@ func (p *Server) runChat( prompt = chatprompt.InsertSystem(prompt, defaultSubagentInstruction) } - // Start buffering stream events for this chat so that new - // subscribers receive a snapshot of in-flight message parts. - p.streamMu.Lock() - startState := p.streamStateLocked(chat.ID) - startState.buffer = nil - startState.buffering = true - p.streamMu.Unlock() - defer func() { - p.streamMu.Lock() - if stopState, ok := p.chatStreams[chat.ID]; ok { - stopState.buffer = nil - stopState.buffering = false - p.cleanupStreamIfIdleLocked(chat.ID, stopState) - } - p.streamMu.Unlock() - }() + // NOTE: Buffering was already started in processChat before + // the running status was published, so message_part events + // are captured from the moment subscribers can see + // status=running. The deferred cleanup also lives in + // processChat. currentChat := chat loadChatSnapshot := func(