Support for images

anthropic.go

@@ -17,7 +17,24 @@ import (
 // actionable message to admins/owners while keeping the response vague for regular users.
 var ErrModelNotFound = errors.New("model not found or deprecated")
 
-func (b *Bot) getAnthropicResponse(ctx context.Context, messages []anthropic.BetaMessageParam, isNewChat, isOwner, isEmojiOnly bool, username string, firstName string, lastName string, isPremium bool, languageCode string, messageTime int) (string, error) {
+// maxFileNotFoundRetries caps the runtime 404 self-heal loop. If multiple
+// referenced file_ids are gone from Anthropic simultaneously (admin purge, AUP
+// enforcement, etc.), we strip them one at a time and retry. Three attempts
+// covers all realistic cascades without leaving the call hanging indefinitely.
+const maxFileNotFoundRetries = 3
+
+// getAnthropicResponse streams the model's response. Each completed text block
+// is delivered to onSegment as soon as the model finishes writing it — so the
+// caller can send segments to Telegram with natural rhythm around tool calls,
+// rather than batched at the very end of the turn. onSegment may be nil for
+// callers that only want the joined text (voice TTS, sticker reactions, etc.).
+// The returned string is every text segment joined by blank lines.
+//
+// chatID is required for the runtime 404 self-heal: when Anthropic returns
+// "File not found:" for a referenced file_id, the dead file_id is stripped
+// from this chat's in-memory ChatMemory and the corresponding DB rows are
+// stamped FilesCleanedAt so a reconciliation job can finish the cleanup.
+func (b *Bot) getAnthropicResponse(ctx context.Context, chatID int64, messages []anthropic.BetaMessageParam, isNewChat, isOwner, isEmojiOnly bool, username string, firstName string, lastName string, isPremium bool, languageCode string, messageTime int, onSegment func(string) error) (string, error) {
 	// Use prompts from config
 	var systemMessage string
 	if isNewChat {
@@ -95,6 +112,9 @@ func (b *Bot) getAnthropicResponse(ctx context.Context, messages []anthropic.Bet
 		MaxTokens: 1000,
 		Messages:  messages,
 		System:    []anthropic.BetaTextBlockParam{{Text: systemMessage}},
+		// Files API beta is always on: replayed conversation history may carry
+		// image content blocks that reference file_ids uploaded on prior turns.
+		Betas: []anthropic.AnthropicBeta{anthropic.AnthropicBetaFilesAPI2025_04_14},
 	}
 
 	// Apply temperature if set in config
@@ -136,40 +156,142 @@ func (b *Bot) getAnthropicResponse(ctx context.Context, messages []anthropic.Bet
 		}
 		params.MCPServers = mcpServers
 		params.Tools = tools
-		params.Betas = []anthropic.AnthropicBeta{anthropic.AnthropicBetaMCPClient2025_11_20}
+		params.Betas = append(params.Betas, anthropic.AnthropicBetaMCPClient2025_11_20)
 	}
 
-	resp, err := b.anthropicClient.Beta.Messages.New(ctx, params)
-	if err != nil {
+	// Streaming + 404 self-heal loop. A "File not found:" 404 from Anthropic
+	// (admin purge, AUP enforcement, accidental delete elsewhere) is caught
+	// here: the offending file_id is stripped from in-memory ChatMemory + the
+	// affected DB rows are stamped for the reconciliation job, and the call is
+	// re-issued. The loop caps at maxFileNotFoundRetries so cascading deletions
+	// can't pin the call indefinitely.
+	for attempt := 0; attempt < maxFileNotFoundRetries; attempt++ {
+		joined, streamErr := b.streamMessages(ctx, params, onSegment)
+		if streamErr == nil {
+			return joined, nil
+		}
 		var apiErr *anthropic.Error
-		if errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound {
+		if !errors.As(streamErr, &apiErr) || apiErr.StatusCode != http.StatusNotFound {
+			return "", fmt.Errorf("error creating Anthropic message: %w", streamErr)
+		}
+		missingFileID := extractMissingFileID(streamErr)
+		if missingFileID == "" {
+			// 404 without a "File not found:" body — interpret as model-not-found,
+			// matching the legacy behavior pre-Files-API.
 			return "", fmt.Errorf("%w: %s", ErrModelNotFound, b.config.Model)
 		}
-		return "", fmt.Errorf("error creating Anthropic message: %w", err)
+		ErrorLogger.Printf("[%s] self-heal: stripping dead file_id %s from chat %d (attempt %d/%d)",
+			b.config.ID, missingFileID, chatID, attempt+1, maxFileNotFoundRetries)
+		b.stripDeadFileIDFromMemory(chatID, missingFileID)
+		if _, cleanupErr := b.markFilesPendingCleanup(ctx, chatID, []string{missingFileID}); cleanupErr != nil {
+			ErrorLogger.Printf("[%s] mark files pending cleanup: %v", b.config.ID, cleanupErr)
+		}
+		params.Messages = b.prepareContextMessages(b.getOrCreateChatMemory(chatID))
 	}
+	return "", fmt.Errorf("max self-heal retries (%d) exceeded: too many file_ids gone from anthropic", maxFileNotFoundRetries)
+}
 
-	var sb strings.Builder
-	for _, block := range resp.Content {
-		switch block.Type {
-		case "text":
-			sb.WriteString(block.Text)
-		case "mcp_tool_use":
-			InfoLogger.Printf("[mcp] tool_use server=%q name=%q id=%q input=%s",
-				block.ServerName, block.Name, block.ID, string(block.Input))
-		case "mcp_tool_result":
-			preview := block.JSON.Content.Raw()
-			if len(preview) > 500 {
-				preview = preview[:500] + "...(truncated)"
+// streamMessages runs one streaming call against the Beta Messages API,
+// dispatching each completed text block to onSegment as it arrives. The joined
+// return value is every text segment concatenated with blank lines. Errors from
+// the SDK are returned raw; the caller wraps them (model-not-found, file 404
+// self-heal, etc.).
+func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNewParams, onSegment func(string) error) (string, error) {
+	stream := b.anthropicClient.Beta.Messages.NewStreaming(ctx, params)
+	defer func() {
+		if err := stream.Close(); err != nil {
+			ErrorLogger.Printf("[stream] close failed: %v", err)
+		}
+	}()
+
+	// Per-block accumulators. Reset on content_block_start, consumed on
+	// content_block_stop. Only one block is active at a time per the SSE
+	// contract; SDK guarantees deltas arrive between matching start/stop.
+	var (
+		allSegments                                       []string
+		currentKind                                       string
+		currentText                                       strings.Builder
+		currentInputJSON                                  strings.Builder
+		currentTUseName, currentTUseServer, currentTUseID string
+		currentTResultUseID, currentTResultServer         string
+		currentTResultIsError                             bool
+		currentTResultContent                             string
+	)
+
+	for stream.Next() {
+		e := stream.Current()
+		switch e.Type {
+		case "content_block_start":
+			cbs := e.AsContentBlockStart()
+			currentKind = cbs.ContentBlock.Type
+			currentText.Reset()
+			currentInputJSON.Reset()
+			switch currentKind {
+			case "mcp_tool_use":
+				currentTUseName = cbs.ContentBlock.Name
+				currentTUseServer = cbs.ContentBlock.ServerName
+				currentTUseID = cbs.ContentBlock.ID
+			case "mcp_tool_result":
+				currentTResultUseID = cbs.ContentBlock.ToolUseID
+				currentTResultServer = cbs.ContentBlock.ServerName
+				currentTResultIsError = cbs.ContentBlock.IsError
+				// Tool-result content arrives populated on start (server-side
+				// pre-assembled), not via subsequent deltas like text/JSON.
+				currentTResultContent = cbs.ContentBlock.JSON.Content.Raw()
 			}
-			InfoLogger.Printf("[mcp] tool_result tool_use_id=%q server=%q is_error=%v content=%s",
-				block.ToolUseID, block.ServerName, block.IsError, preview)
-		default:
-			InfoLogger.Printf("[mcp] block type=%q (unhandled)", block.Type)
+
+		case "content_block_delta":
+			cbd := e.AsContentBlockDelta()
+			switch cbd.Delta.Type {
+			case "text_delta":
+				if currentKind == "text" {
+					currentText.WriteString(cbd.Delta.Text)
+				}
+			case "input_json_delta":
+				if currentKind == "mcp_tool_use" {
+					currentInputJSON.WriteString(cbd.Delta.PartialJSON)
+				}
+			}
+
+		case "content_block_stop":
+			switch currentKind {
+			case "text":
+				seg := strings.TrimSpace(currentText.String())
+				if seg != "" {
+					allSegments = append(allSegments, seg)
+					if onSegment != nil {
+						if cbErr := onSegment(seg); cbErr != nil {
+							// Log but keep streaming — the model's response
+							// is still inbound; we want it recorded even if
+							// one Telegram send failed.
+							ErrorLogger.Printf("[stream] onSegment failed: %v", cbErr)
+						}
+					}
+				}
+			case "mcp_tool_use":
+				InfoLogger.Printf("[mcp] tool_use server=%q name=%q id=%q input=%s",
+					currentTUseServer, currentTUseName, currentTUseID, currentInputJSON.String())
+			case "mcp_tool_result":
+				preview := currentTResultContent
+				if len(preview) > 500 {
+					preview = preview[:500] + "...(truncated)"
+				}
+				InfoLogger.Printf("[mcp] tool_result tool_use_id=%q server=%q is_error=%v content=%s",
+					currentTResultUseID, currentTResultServer, currentTResultIsError, preview)
+			default:
+				if currentKind != "" {
+					InfoLogger.Printf("[mcp] block type=%q (unhandled)", currentKind)
+				}
+			}
+			currentKind = ""
 		}
 	}
-	out := sb.String()
-	if out == "" {
+
+	if err := stream.Err(); err != nil {
+		return "", err
+	}
+	if len(allSegments) == 0 {
 		return "", fmt.Errorf("unexpected response format from Anthropic")
 	}
-	return out, nil
+	return strings.Join(allSegments, "\n\n"), nil
 }