| 4489 | } |
| 4490 | |
| 4491 | func (p *Server) publishToStream(chatID uuid.UUID, event codersdk.ChatStreamEvent) { |
| 4492 | state := p.getOrCreateStreamState(chatID) |
| 4493 | state.mu.Lock() |
| 4494 | switch event.Type { |
| 4495 | case codersdk.ChatStreamEventTypeRetry: |
| 4496 | if event.Retry != nil { |
| 4497 | retryCopy := *event.Retry |
| 4498 | state.currentRetry = &retryCopy |
| 4499 | } |
| 4500 | case codersdk.ChatStreamEventTypeMessagePart: |
| 4501 | // Any streamed part means the provider is making forward |
| 4502 | // progress again, so the stream has left the retry backoff |
| 4503 | // window regardless of role. |
| 4504 | state.currentRetry = nil |
| 4505 | case codersdk.ChatStreamEventTypeError: |
| 4506 | state.currentRetry = nil |
| 4507 | case codersdk.ChatStreamEventTypeStatus: |
| 4508 | if event.Status != nil && shouldClearRetryPhaseForStatus(event.Status.Status) { |
| 4509 | state.currentRetry = nil |
| 4510 | } |
| 4511 | } |
| 4512 | if event.Type == codersdk.ChatStreamEventTypeMessagePart { |
| 4513 | if !state.buffering { |
| 4514 | p.cleanupStreamIfIdle(chatID, state) |
| 4515 | state.mu.Unlock() |
| 4516 | return |
| 4517 | } |
| 4518 | if len(state.buffer) >= maxStreamBufferSize { |
| 4519 | p.metrics.RecordStreamBufferDropped() |
| 4520 | state.bufferDropCount++ |
| 4521 | now := p.clock.Now() |
| 4522 | if now.Sub(state.bufferLastWarnAt) >= streamDropWarnInterval { |
| 4523 | p.logger.Warn(context.Background(), "chat stream buffer full, dropping oldest event", |
| 4524 | slog.F("chat_id", chatID), |
| 4525 | slog.F("buffer_size", len(state.buffer)), |
| 4526 | slog.F("dropped_count", state.bufferDropCount), |
| 4527 | ) |
| 4528 | state.bufferDropCount = 0 |
| 4529 | state.bufferLastWarnAt = now |
| 4530 | } |
| 4531 | // Zero the dropped slot so its *ChatStreamMessagePart is |
| 4532 | // GC-eligible; the later append reuses this slot in place |
| 4533 | // whenever cap > len. |
| 4534 | state.buffer[0] = bufferedStreamPart{} |
| 4535 | state.buffer = state.buffer[1:] |
| 4536 | } |
| 4537 | state.buffer = append(state.buffer, bufferedStreamPart{ |
| 4538 | event: event, |
| 4539 | // committedMessageID stays 0 here: the part belongs to |
| 4540 | // the in-progress turn until publishMessage claims it |
| 4541 | // with the committed assistant message ID. |
| 4542 | }) |
| 4543 | } |
| 4544 | subscribers := make([]chan codersdk.ChatStreamEvent, 0, len(state.subscribers)) |
| 4545 | for _, ch := range state.subscribers { |
| 4546 | subscribers = append(subscribers, ch) |
| 4547 | } |
| 4548 | state.mu.Unlock() |