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(