Method subscribeToStream

coderd/x/chatd/chatd.go:4668–4700

subscribeToStream registers a subscriber to the per-chat in-memory stream and returns a snapshot of currently in-progress message_part events plus the current retry phase, the live subscriber channel, and a cancel func. Parts that were claimed by a committed durable assistant message (committedMess

(chatID uuid.UUID)

// so re-delivering their constituent parts here would render the
// same content twice.
func (p *Server) subscribeToStream(chatID uuid.UUID) (
	[]codersdk.ChatStreamEvent,
	*codersdk.ChatStreamRetry,
	<-chan codersdk.ChatStreamEvent,
	func(),
) {
	state := p.getOrCreateStreamState(chatID)
	state.mu.Lock()
	snapshot := snapshotBufferLocked(state.buffer)
	var currentRetry *codersdk.ChatStreamRetry
	if state.currentRetry != nil {
		retryCopy := *state.currentRetry
		currentRetry = &retryCopy
	}
	id := uuid.New()
	ch := make(chan codersdk.ChatStreamEvent, 128)
	state.subscribers[id] = ch
	state.mu.Unlock()

	cancel := func() {
		state.mu.Lock()
		// Remove the subscriber but do not close the channel.
		// publishToStream copies subscriber references under
		// the per-chat lock then sends outside; closing here
		// races with that send and can panic. The channel
		// becomes unreachable once removed and will be GC'd.
		delete(state.subscribers, id)
		p.cleanupStreamIfIdle(chatID, state)
		state.mu.Unlock()
	}

	return snapshot, currentRetry, ch, cancel
}

// getOrCreateStreamState returns the per-chat stream state,
// creating one atomically if it doesn't exist. The returned