Dep upd

handlers.go

@@ -28,6 +28,12 @@ func (b *Bot) handleVoiceMessage(ctx context.Context, message *models.Message, u
 		return
 	}
 
+	// Voice is a committed turn: it waits for any running turn and is never
+	// restarted, so a text sent right after a voice note is answered afterwards.
+	turn, turnCtx := b.beginTurn(ctx, chatID)
+	defer b.endTurn(turn)
+	ctx = turnCtx
+
 	stopTyping := b.startChatAction(ctx, chatID, businessConnectionID, models.ChatActionTyping)
 	defer stopTyping()
 
@@ -54,9 +60,8 @@ func (b *Bot) handleVoiceMessage(ctx context.Context, message *models.Message, u
 	}
 	b.chatMemoriesMu.Unlock()
 
-	chatMemory := b.getOrCreateChatMemory(chatID)
-	contextMessages := b.prepareContextMessages(chatMemory)
-	response, err := b.getAnthropicResponse(ctx, chatID, contextMessages, false, username, firstName, lastName, isPremium, languageCode, messageTime, nil)
+	contextMessages := b.snapshotForTurn(turn, b.getOrCreateChatMemory(chatID))
+	response, err := b.getAnthropicResponse(ctx, chatID, contextMessages, false, username, firstName, lastName, isPremium, languageCode, messageTime, nil, nil)
 	if err != nil {
 		ErrorLogger.Printf("Error getting Anthropic response for voice: %v", err)
 		if err := b.sendResponse(ctx, chatID, b.anthropicErrorResponse(err, userID), businessConnectionID); err != nil {
@@ -120,11 +125,53 @@ func (b *Bot) handlePhotoMessage(
 		return
 	}
 
-	// Covers the Files API uploads as well as the model turn; on an album this
-	// is the longest wait in the bot.
+	// The Files API uploads depend on nothing the chat's running turn is doing,
+	// so they overlap it instead of queueing behind it. On an album this is the
+	// longest wait in the bot, and the typing indicator covers it.
 	stopTyping := b.startChatAction(ctx, chatID, businessConnectionID, models.ChatActionTyping)
-	defer stopTyping()
+	finalUploaded, caption, err := b.uploadPhotoItems(ctx, items, chatID, businessConnectionID)
+	stopTyping()
+	if err != nil || len(finalUploaded) == 0 {
+		return
+	}
 
+	req := turnRequest{
+		ctx: ctx, chatID: chatID, userID: userID,
+		username: username, firstName: firstName, lastName: lastName,
+		languageCode: languageCode, isPremium: isPremium, messageTime: messageTime,
+		businessConnectionID: businessConnectionID,
+	}
+	// Photos wait for any running turn and are never restarted.
+	turn, turnCtx := b.beginTurn(ctx, chatID)
+
+	b.runTurn(turn, turnCtx, req, func(ctx context.Context, out replyOutput) (string, error) {
+		chatMemory := b.getOrCreateChatMemory(chatID)
+		userMessage := b.createMessage(chatID, userID, username, "user", caption, true)
+		userMessage.ImageFileIDs = finalUploaded
+		userMessage.TelegramMessageID = items[0].ID
+		if err := b.storeMessage(&userMessage); err != nil {
+			b.compensatingDelete(ctx, finalUploaded)
+			ErrorLogger.Printf("[%s] store photo message failed: %v", b.config.ID, err)
+			if sendErr := b.sendResponse(ctx, chatID, "Sorry, I had trouble saving your message.", businessConnectionID); sendErr != nil {
+				ErrorLogger.Printf("Error sending store failure message: %v", sendErr)
+			}
+			return "", errTurnSilent
+		}
+		b.addMessageToChatMemory(chatMemory, userMessage)
+
+		contextMessages := b.snapshotForTurn(turn, chatMemory)
+		return b.getAnthropicResponse(
+			ctx, chatID, contextMessages, false,
+			username, firstName, lastName, isPremium, languageCode, messageTime,
+			out.onSegment, out.onProgress,
+		)
+	})
+}
+
+// uploadPhotoItems uploads every photo in items to the Files API in parallel
+// and returns their file ids with the album caption. On any failure it deletes
+// the uploads that did succeed, tells the user, and returns errTurnSilent.
+func (b *Bot) uploadPhotoItems(ctx context.Context, items []*models.Message, chatID int64, businessConnectionID string) ([]string, string, error) {
 	uploaded := make([]string, len(items))
 	caption := ""
 	g, gctx := errgroup.WithContext(ctx)
@@ -157,7 +204,7 @@ func (b *Bot) handlePhotoMessage(
 		if sendErr := b.sendResponse(ctx, chatID, "Sorry, I couldn't process one of your images.", businessConnectionID); sendErr != nil {
 			ErrorLogger.Printf("Error sending photo failure message: %v", sendErr)
 		}
-		return
+		return nil, "", errTurnSilent
 	}
 	finalUploaded := make([]string, 0, len(uploaded))
 	for _, fid := range uploaded {
@@ -165,41 +212,7 @@ func (b *Bot) handlePhotoMessage(
 			finalUploaded = append(finalUploaded, fid)
 		}
 	}
-	if len(finalUploaded) == 0 {
-		return
-	}
-
-	chatMemory := b.getOrCreateChatMemory(chatID)
-	userMessage := b.createMessage(chatID, userID, username, "user", caption, true)
-	userMessage.ImageFileIDs = finalUploaded
-	if err := b.storeMessage(&userMessage); err != nil {
-		b.compensatingDelete(ctx, finalUploaded)
-		ErrorLogger.Printf("[%s] store photo message failed: %v", b.config.ID, err)
-		if sendErr := b.sendResponse(ctx, chatID, "Sorry, I had trouble saving your message.", businessConnectionID); sendErr != nil {
-			ErrorLogger.Printf("Error sending store failure message: %v", sendErr)
-		}
-		return
-	}
-	b.addMessageToChatMemory(chatMemory, userMessage)
-
-	contextMessages := b.prepareContextMessages(chatMemory)
-	joined, err := b.getAnthropicResponse(
-		ctx, chatID, contextMessages, false,
-		username, firstName, lastName, isPremium, languageCode, messageTime,
-		func(seg string) error {
-			return b.sendOneSegment(ctx, chatID, seg, businessConnectionID)
-		},
-	)
-	if err != nil {
-		ErrorLogger.Printf("Error getting Anthropic response for photo: %v", err)
-		if sendErr := b.sendResponse(ctx, chatID, b.anthropicErrorResponse(err, userID), businessConnectionID); sendErr != nil {
-			ErrorLogger.Printf("Error sending anthropic error response: %v", sendErr)
-		}
-		return
-	}
-	if _, storeErr := b.screenOutgoingMessage(chatID, joined); storeErr != nil {
-		ErrorLogger.Printf("Error recording assistant turn: %v", storeErr)
-	}
+	return finalUploaded, caption, nil
 }
 
 // respondToChat runs one assistant turn against the chat's current memory and
@@ -207,40 +220,30 @@ func (b *Bot) handlePhotoMessage(
 // through here, so a coalesced turn is byte-for-byte the same request as a
 // single-message one: the messages were already written to memory at intake, and
 // the model simply sees more of them.
-func (b *Bot) respondToChat(
-	ctx context.Context,
-	chatID, userID int64,
-	isEmojiOnly bool,
-	username, firstName, lastName string,
-	isPremium bool,
-	languageCode string,
-	messageTime int,
-	businessConnectionID string,
-) {
-	stopTyping := b.startChatAction(ctx, chatID, businessConnectionID, models.ChatActionTyping)
-	defer stopTyping()
-
-	chatMemory := b.getOrCreateChatMemory(chatID)
-	contextMessages := b.prepareContextMessages(chatMemory)
-
-	joined, err := b.getAnthropicResponse(
-		ctx, chatID, contextMessages, isEmojiOnly,
-		username, firstName, lastName, isPremium, languageCode, messageTime,
-		func(seg string) error {
-			return b.sendOneSegment(ctx, chatID, seg, businessConnectionID)
-		},
-	)
-	if err != nil {
-		ErrorLogger.Printf("Error getting Anthropic response: %v", err)
-		if sendErr := b.sendResponse(ctx, chatID, b.anthropicErrorResponse(err, userID), businessConnectionID); sendErr != nil {
-			ErrorLogger.Printf("Error sending response: %v", sendErr)
-		}
+//
+// If a turn is already running for the chat, beginTextTurn decides whether this
+// flush is dropped, restarts it, or runs after it; see turn.go.
+func (b *Bot) respondToChat(req turnRequest) {
+	wantID := b.newestUserMessageIDInChat(req.chatID)
+	if wantID == 0 {
+		// Nothing of the user's is in memory, for example because /clear
+		// landed after this flush was collected. There is nothing to answer.
+		InfoLogger.Printf("[%s] turn: chat %d has no user message to answer; skipping", b.config.ID, req.chatID)
+		return
+	}
+	turn, turnCtx, ok := b.beginTextTurn(req, wantID)
+	if !ok {
 		return
 	}
 
-	if _, storeErr := b.screenOutgoingMessage(chatID, joined); storeErr != nil {
-		ErrorLogger.Printf("Error recording assistant turn: %v", storeErr)
-	}
+	b.runTurn(turn, turnCtx, req, func(ctx context.Context, out replyOutput) (string, error) {
+		contextMessages := b.snapshotForTurn(turn, b.getOrCreateChatMemory(req.chatID))
+		return b.getAnthropicResponse(
+			ctx, req.chatID, contextMessages, req.isEmojiOnly,
+			req.username, req.firstName, req.lastName, req.isPremium, req.languageCode, req.messageTime,
+			out.onSegment, out.onProgress,
+		)
+	})
 }
 
 func (b *Bot) anthropicErrorResponse(err error, userID int64) string {
@@ -274,6 +277,19 @@ func (b *Bot) anthropicErrorResponse(err error, userID int64) string {
 }
 
 func (b *Bot) handleUpdate(ctx context.Context, tgBot *bot.Bot, update *models.Update) {
+	if stopped := update.StoppedMessageGeneration; stopped != nil {
+		b.handleStopGeneration(stopped.Chat.ID, stopped.DraftID)
+		return
+	}
+	if edited := update.EditedMessage; edited != nil {
+		b.handleEditedMessage(ctx, edited)
+		return
+	}
+	if edited := update.EditedBusinessMessage; edited != nil {
+		b.handleEditedMessage(ctx, edited)
+		return
+	}
+
 	var message *models.Message
 
 	if update.Message != nil {
@@ -489,8 +505,10 @@ func (b *Bot) handleUpdate(ctx context.Context, tgBot *bot.Bot, update *models.U
 
 	if message.Sticker != nil {
 		b.cancelIntake(chatID)
-		contextMessages := b.prepareContextMessages(b.getOrCreateChatMemory(chatID))
-		b.handleStickerMessage(ctx, chatID, userMsg, message, contextMessages, businessConnectionID)
+		turn, turnCtx := b.beginTurn(ctx, chatID)
+		defer b.endTurn(turn)
+		contextMessages := b.snapshotForTurn(turn, b.getOrCreateChatMemory(chatID))
+		b.handleStickerMessage(turnCtx, chatID, userMsg, message, contextMessages, businessConnectionID)
 		return
 	}
 
@@ -510,9 +528,89 @@ func (b *Bot) handleUpdate(ctx context.Context, tgBot *bot.Bot, update *models.U
 		return
 	}
 
-	b.respondToChat(ctx, chatID, userID, isEmojiOnly,
-		username, firstName, lastName, isPremium, languageCode, messageTime,
-		businessConnectionID)
+	b.respondToChat(turnRequest{
+		ctx: ctx, chatID: chatID, userID: userID, isEmojiOnly: isEmojiOnly,
+		username: username, firstName: firstName, lastName: lastName,
+		isPremium: isPremium, languageCode: languageCode, messageTime: messageTime,
+		businessConnectionID: businessConnectionID,
+	})
+}
+
+// handleEditedMessage applies a Telegram edit to the stored copy of a user
+// message. Mobile clients deliver autocorrect fixes as edits within a few
+// seconds of the send, so the model should see the corrected text. What
+// happens next depends on where the message is:
+//
+//   - still in its quiet window: the window restarts as if it had just arrived;
+//   - being read by a turn that has shown nothing yet: that turn restarts;
+//   - already answered, or shown: only the record is corrected.
+func (b *Bot) handleEditedMessage(ctx context.Context, edited *models.Message) {
+	text := edited.Text
+	if text == "" {
+		text = edited.Caption
+	}
+	if edited.From == nil || text == "" {
+		return
+	}
+	chatID := edited.Chat.ID
+
+	// Memory is the gate: only a message still in memory can influence a turn,
+	// and the check is cheap. It also keeps edits from chats this bot never
+	// stored off the database; a business connection delivers every edit on
+	// the whole account.
+	var msgID uint
+	answered := false
+	b.chatMemoriesMu.Lock()
+	if mem, exists := b.chatMemories[chatID]; exists {
+		for i := len(mem.Messages) - 1; i >= 0; i-- {
+			m := &mem.Messages[i]
+			if m.IsUser && m.TelegramMessageID == edited.ID {
+				m.Text = text
+				msgID = m.ID
+				break
+			}
+			if !m.IsUser {
+				answered = true
+			}
+		}
+	}
+	b.chatMemoriesMu.Unlock()
+	if msgID == 0 {
+		return
+	}
+
+	if err := b.db.Model(&Message{}).Where("id = ?", msgID).Update("text", text).Error; err != nil {
+		ErrorLogger.Printf("[%s] applying edit to message %d in chat %d: %v", b.config.ID, edited.ID, chatID, err)
+	}
+	if answered {
+		return
+	}
+
+	isEmojiOnly := isOnlyEmojis(text)
+	if b.touchIntake(chatID, isEmojiOnly) {
+		InfoLogger.Printf("[%s] edit to message %d re-armed the quiet window for chat %d", b.config.ID, edited.ID, chatID)
+		return
+	}
+
+	req := turnRequest{
+		ctx: ctx, chatID: chatID, userID: edited.From.ID,
+		username: edited.From.Username, firstName: edited.From.FirstName, lastName: edited.From.LastName,
+		languageCode: edited.From.LanguageCode, isPremium: edited.From.IsPremium, messageTime: edited.Date,
+		businessConnectionID: edited.BusinessConnectionID, isEmojiOnly: isEmojiOnly,
+	}
+	if b.restartForEdit(req, msgID) {
+		InfoLogger.Printf("[%s] edit to message %d restarted the running turn for chat %d", b.config.ID, edited.ID, chatID)
+	}
+}
+
+// handleStopGeneration routes Telegram's stopped_message_generation update,
+// sent when the user taps Stop under a streaming draft.
+func (b *Bot) handleStopGeneration(chatID int64, draftID int) {
+	if b.stopTurn(chatID, draftID) {
+		InfoLogger.Printf("[%s] user stopped generation in chat %d (draft %d)", b.config.ID, chatID, draftID)
+		return
+	}
+	InfoLogger.Printf("[%s] stop for chat %d ignored: draft %d is not the running turn", b.config.ID, chatID, draftID)
 }
 
 func (b *Bot) sendRateLimitExceededMessage(ctx context.Context, chatID int64, businessConnectionID string) {
@@ -547,7 +645,7 @@ func (b *Bot) generateStickerResponse(ctx context.Context, message Message, cont
 
 	if message.StickerFileID != "" {
 		messageTime := int(message.Timestamp.Unix())
-		response, err := b.getAnthropicResponse(ctx, message.ChatID, contextMessages, true, message.Username, "", "", false, "", messageTime, nil)
+		response, err := b.getAnthropicResponse(ctx, message.ChatID, contextMessages, true, message.Username, "", "", false, "", messageTime, nil, nil)
 		if err != nil {
 			return "", err
 		}
@@ -641,6 +739,9 @@ func (b *Bot) clearChatHistory(ctx context.Context, chatID int64, currentUserID
 		InfoLogger.Printf("[%s] discarded %d buffered message(s) for chat %d on history clear",
 			b.config.ID, discarded, clearedChatID)
 	}
+	// Likewise a flush collected behind a running turn: rerun against the
+	// emptied chat, it would call the model with no user message at all.
+	b.discardRerun(clearedChatID)
 
 	b.chatMemoriesMu.Lock()
 	if targetUserID == currentUserID {