| 6014 | } |
| 6015 | |
| 6016 | func (p *Server) finishActiveChat( |
| 6017 | ctx context.Context, |
| 6018 | logger slog.Logger, |
| 6019 | chat database.Chat, |
| 6020 | status database.ChatStatus, |
| 6021 | lastError pqtype.NullRawMessage, |
| 6022 | ) (finishActiveChatResult, error) { |
| 6023 | result := finishActiveChatResult{} |
| 6024 | |
| 6025 | err := p.db.InTx(func(tx database.Store) error { |
| 6026 | // Re-read the chat status under lock — another caller |
| 6027 | // (e.g. promote) may have already set it to pending. |
| 6028 | latestChat, lockErr := tx.GetChatByIDForUpdate(ctx, chat.ID) |
| 6029 | if lockErr != nil { |
| 6030 | return xerrors.Errorf("lock chat for release: %w", lockErr) |
| 6031 | } |
| 6032 | |
| 6033 | // If another worker has already acquired this chat, |
| 6034 | // bail out — we must not overwrite their running |
| 6035 | // status or publish spurious events. |
| 6036 | if latestChat.Status == database.ChatStatusRunning && |
| 6037 | latestChat.WorkerID.Valid && |
| 6038 | latestChat.WorkerID.UUID != p.workerID { |
| 6039 | return errChatTakenByOtherWorker |
| 6040 | } |
| 6041 | |
| 6042 | // If someone else already set the chat to pending (e.g. |
| 6043 | // the promote endpoint), don't overwrite it — just clear |
| 6044 | // the worker and let the processor pick it back up. |
| 6045 | switch { |
| 6046 | case latestChat.Status == database.ChatStatusPending: |
| 6047 | status = database.ChatStatusPending |
| 6048 | case latestChat.Status == database.ChatStatusWaiting && status != database.ChatStatusWaiting && !latestChat.Archived: |
| 6049 | // PromoteQueued's deferred path won the status race. |
| 6050 | // Insert synthetic tool results before auto-promoting, |
| 6051 | // or a RequiresAction worker outcome reintroduces the |
| 6052 | // stops-dead bug this PR exists to fix. |
| 6053 | inserted, synthErr := insertSyntheticToolResultsTx( |
| 6054 | ctx, tx, latestChat, |
| 6055 | "Tool execution interrupted by queued message promotion", |
| 6056 | ) |
| 6057 | if synthErr != nil { |
| 6058 | return xerrors.Errorf("insert synthetic tool results during promote-driven cleanup: %w", synthErr) |
| 6059 | } |
| 6060 | result.syntheticToolResults = inserted |
| 6061 | var promoteErr error |
| 6062 | result.promotedMessage, result.remainingQueuedMessages, result.shouldPublishQueueUpdate, promoteErr = p.tryAutoPromoteQueuedMessage(ctx, tx, latestChat) |
| 6063 | if promoteErr != nil { |
| 6064 | logger.Error(ctx, "auto-promote queued message failed during promote-driven cleanup", slog.Error(promoteErr)) |
| 6065 | return xerrors.Errorf("auto-promote queued message: %w", promoteErr) |
| 6066 | } |
| 6067 | if result.promotedMessage != nil { |
| 6068 | status = database.ChatStatusPending |
| 6069 | } else { |
| 6070 | // Queue drained between snapshot and lock; honor |
| 6071 | // the external Waiting. |
| 6072 | status = database.ChatStatusWaiting |
| 6073 | } |