Merge current Pulse upstream for publication

Change-source: pulse-maintainer

internal/ai/providers/openai.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentcapabilities"
@@ -43,6 +44,10 @@ type OpenAIClient struct {
 	baseURL      string
 	client       *http.Client
 	streamClient *http.Client
+	// completionBudgetLearned records that this endpoint rejected max_tokens
+	// and asked for max_completion_tokens, so every later request sends the
+	// budget in that field without paying another round trip (#1837).
+	completionBudgetLearned atomic.Bool
 	// The configured request timeout bounds how long Pulse waits for the
 	// stream to START (response headers and first chunk). Once deltas flow
 	// there is deliberately no overall wall-clock deadline: reasoning models
@@ -466,16 +471,61 @@ func (c *OpenAIClient) requestMaxTokens(req ChatRequest) int {
 	return 0
 }
 
-// requiresMaxCompletionTokens returns true for models that need max_completion_tokens instead of max_tokens
-// Per OpenAI docs, o1/o3/o4 reasoning models require max_completion_tokens; max_tokens will error.
+// requiresMaxCompletionTokens returns true for models that need max_completion_tokens instead of max_tokens.
+// OpenAI's reasoning models (o-series and the GPT-5 family) reject max_tokens
+// with "Use 'max_completion_tokens' instead", on api.openai.com and on Azure
+// deployments of the same models. Azure deployment names need not carry the
+// model name, so an endpoint that has already rejected max_tokens is also
+// remembered for the life of the client.
 func (c *OpenAIClient) requiresMaxCompletionTokens(model string) bool {
 	if c.isOpenRouter() {
 		return true
 	}
-	if !c.usesOfficialOpenAIEndpoint() {
+	if c.completionBudgetLearned.Load() {
+		return true
+	}
+	if !c.usesOfficialOpenAIEndpoint() && !c.usesAzureOpenAIEndpoint() {
 		return false
 	}
-	return strings.HasPrefix(model, "o1") || strings.HasPrefix(model, "o3") || strings.HasPrefix(model, "o4")
+	return isOpenAIReasoningFamily(model)
+}
+
+func isOpenAIReasoningFamily(model string) bool {
+	normalized := strings.ToLower(strings.TrimSpace(model))
+	for _, prefix := range []string{"o1", "o3", "o4", "gpt-5"} {
+		if strings.HasPrefix(normalized, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
+// usesAzureOpenAIEndpoint reports whether the openai provider points at an
+// Azure OpenAI or Azure AI Foundry host, which serves official OpenAI models
+// under their official parameter rules.
+func (c *OpenAIClient) usesAzureOpenAIEndpoint() bool {
+	if c.Name() != "openai" {
+		return false
+	}
+	u, err := url.Parse(c.baseURL)
+	if err != nil {
+		return false
+	}
+	host := strings.ToLower(u.Hostname())
+	return strings.HasSuffix(host, ".openai.azure.com") ||
+		strings.HasSuffix(host, ".services.ai.azure.com") ||
+		strings.HasSuffix(host, ".cognitiveservices.azure.com")
+}
+
+// openAIRejectsMaxTokens reports whether a 400 response is the model telling
+// us to send max_completion_tokens instead of max_tokens. The request is
+// then re-sent once in the form the endpoint asked for (#1837).
+func openAIRejectsMaxTokens(status int, errMsg string) bool {
+	if status != http.StatusBadRequest {
+		return false
+	}
+	lower := strings.ToLower(errMsg)
+	return strings.Contains(lower, "max_tokens") && strings.Contains(lower, "max_completion_tokens")
 }
 
 func (c *OpenAIClient) supportsStreamOptions() bool {
@@ -638,9 +688,11 @@ func (c *OpenAIClient) Chat(ctx context.Context, req ChatRequest) (*ChatResponse
 	var respBody []byte
 	var lastErr error
 	maxRetries := openaiMaxRetries
+	budgetSwitched := false
+	skipBackoff := false
 
 	for attempt := 0; attempt <= maxRetries; attempt++ {
-		if attempt > 0 {
+		if attempt > 0 && !skipBackoff {
 			// Exponential backoff: 2s, 4s, 8s
 			backoff := openaiInitialBackoff * time.Duration(1<<(attempt-1))
 			log.Warn().
@@ -662,6 +714,7 @@ func (c *OpenAIClient) Chat(ctx context.Context, req ChatRequest) (*ChatResponse
 			case <-backoffTimer.C:
 			}
 		}
+		skipBackoff = false
 
 		httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL, bytes.NewReader(body))
 		if err != nil {
@@ -708,11 +761,29 @@ func (c *OpenAIClient) Chat(ctx context.Context, req ChatRequest) (*ChatResponse
 		// Non-retryable error
 		if resp.StatusCode != http.StatusOK {
 			var errResp openaiError
+			errMsg := string(respBody)
 			if err := json.Unmarshal(respBody, &errResp); err == nil && errResp.Error.Message != "" {
-				errMsg := appendRateLimitInfo(errResp.Error.Message, resp)
-				return nil, fmt.Errorf("API error (%d): %s", resp.StatusCode, errMsg)
+				errMsg = errResp.Error.Message
+			}
+			errMsg = appendRateLimitInfo(errMsg, resp)
+			if !budgetSwitched && openaiReq.MaxTokens > 0 && openAIRejectsMaxTokens(resp.StatusCode, errMsg) {
+				// The endpoint named the field it wants. Re-send once in that
+				// form and remember it; these models also refuse a non-default
+				// temperature, so it is dropped with the budget field (#1837).
+				openaiReq.MaxCompletionTokens = openaiReq.MaxTokens
+				openaiReq.MaxTokens = 0
+				openaiReq.Temperature = 0
+				if body, err = json.Marshal(openaiReq); err != nil {
+					return nil, fmt.Errorf("failed to marshal request: %w", err)
+				}
+				c.completionBudgetLearned.Store(true)
+				budgetSwitched = true
+				skipBackoff = true
+				attempt--
+				log.Info().Str("provider", c.Name()).Str("model", openaiReq.Model).
+					Msg("endpoint requires max_completion_tokens; re-sending request in that form")
+				continue
 			}
-			errMsg := appendRateLimitInfo(string(respBody), resp)
 			return nil, fmt.Errorf("API error (%d): %s", resp.StatusCode, errMsg)
 		}
 
@@ -1030,9 +1101,11 @@ func (c *OpenAIClient) ChatStream(ctx context.Context, req ChatRequest, callback
 		streamClient = c.client
 	}
 	maxRetries := openaiStreamMaxRetries
+	budgetSwitched := false
+	skipBackoff := false
 
 	for attempt := 0; attempt <= maxRetries; attempt++ {
-		if attempt > 0 {
+		if attempt > 0 && !skipBackoff {
 			backoff := openaiStreamInitialBackoff * time.Duration(1<<(attempt-1))
 			log.Warn().
 				Int("attempt", attempt).
@@ -1053,6 +1126,7 @@ func (c *OpenAIClient) ChatStream(ctx context.Context, req ChatRequest, callback
 			case <-backoffTimer.C:
 			}
 		}
+		skipBackoff = false
 
 		httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL, bytes.NewReader(body))
 		if err != nil {
@@ -1094,6 +1168,22 @@ func (c *OpenAIClient) ChatStream(ctx context.Context, req ChatRequest, callback
 		if isRetryableOpenAIStreamStatus(resp.StatusCode) {
 			continue
 		}
+		if !budgetSwitched && openaiReq.MaxTokens > 0 && openAIRejectsMaxTokens(resp.StatusCode, errMsg) {
+			// Same one-shot correction as the buffered path (#1837).
+			openaiReq.MaxCompletionTokens = openaiReq.MaxTokens
+			openaiReq.MaxTokens = 0
+			openaiReq.Temperature = 0
+			if body, err = json.Marshal(openaiReq); err != nil {
+				return fmt.Errorf("failed to marshal request: %w", err)
+			}
+			c.completionBudgetLearned.Store(true)
+			budgetSwitched = true
+			skipBackoff = true
+			attempt--
+			log.Info().Str("provider", c.Name()).Str("model", openaiReq.Model).
+				Msg("endpoint requires max_completion_tokens; re-sending stream request in that form")
+			continue
+		}
 		if openAIStreamingExplicitlyUnsupported(resp.StatusCode, errMsg) {
 			// Some otherwise compatible endpoints only implement buffered chat
 			// completions. Retry exactly once without stream=true, then emit the

internal/ai/providers/openai_test.go

@@ -1,6 +1,7 @@
 package providers
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -1379,6 +1380,153 @@ func TestOpenAIClient_HelperFlags(t *testing.T) {
 	assert.False(t, client.requiresMaxCompletionTokens("gpt-4"))
 }
 
+// Issue #1837: GPT-5 family models reject max_tokens on api.openai.com and on
+// Azure deployments of the same models.
+func TestOpenAIClient_RequiresMaxCompletionTokens_GPT5FamilyAndAzure(t *testing.T) {
+	official := NewOpenAIClient("sk", "gpt-5.6-luna-2026-07-09", "https://api.openai.com/v1", 0)
+	assert.True(t, official.requiresMaxCompletionTokens("gpt-5.6-luna-2026-07-09"))
+	assert.True(t, official.requiresMaxCompletionTokens("GPT-5-mini"))
+	assert.False(t, official.requiresMaxCompletionTokens("gpt-4o"))
+
+	azure := NewOpenAIClient("key", "gpt-5.6-luna", "https://example.services.ai.azure.com/openai/v1/chat/completions", 0)
+	assert.True(t, azure.usesAzureOpenAIEndpoint())
+	assert.True(t, azure.requiresMaxCompletionTokens("gpt-5.6-luna"))
+	assert.False(t, azure.requiresMaxCompletionTokens("gpt-4o"))
+
+	compatible := NewOpenAIClient("key", "gpt-5.6-luna", "https://gateway.example.com/v1", 0)
+	assert.False(t, compatible.requiresMaxCompletionTokens("gpt-5.6-luna"))
+	compatible.completionBudgetLearned.Store(true)
+	assert.True(t, compatible.requiresMaxCompletionTokens("anything"))
+}
+
+func maxTokensRejection() []byte {
+	body, _ := json.Marshal(openaiError{Error: openaiErrorDetail{
+		Message: "Unsupported parameter: 'max_tokens' is not supported with this model. Use 'max_completion_tokens' instead.",
+		Type:    "invalid_request_error",
+		Code:    "unsupported_parameter",
+	}})
+	return body
+}
+
+func TestOpenAIClient_Chat_ResendsWithMaxCompletionTokensWhenRejected(t *testing.T) {
+	var requests []openaiRequest
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var req openaiRequest
+		require.NoError(t, json.NewDecoder(r.Body).Decode(&req))
+		requests = append(requests, req)
+		if req.MaxTokens > 0 {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusBadRequest)
+			_, _ = w.Write(maxTokensRejection())
+			return
+		}
+		_ = json.NewEncoder(w).Encode(openaiResponse{
+			ID:    "chatcmpl-1",
+			Model: req.Model,
+			Choices: []openaiChoice{{
+				Message:      openaiRespMsg{Role: "assistant", Content: "Hello"},
+				FinishReason: "stop",
+			}},
+			Usage: openaiUsage{PromptTokens: 2, CompletionTokens: 3},
+		})
+	}))
+	defer server.Close()
+
+	// A deployment name that does not reveal the model family, as on Azure.
+	client := NewOpenAIClient("key", "luna-prod", server.URL, 0)
+	request := ChatRequest{
+		MaxTokens:   123,
+		Temperature: 0.7,
+		Messages:    []Message{{Role: "user", Content: "Hi"}},
+	}
+	resp, err := client.Chat(context.Background(), request)
+	require.NoError(t, err)
+	assert.Equal(t, "Hello", resp.Content)
+	require.Len(t, requests, 2)
+	assert.Equal(t, 123, requests[0].MaxTokens)
+	assert.Equal(t, 123, requests[1].MaxCompletionTokens)
+	assert.Zero(t, requests[1].MaxTokens)
+	assert.Zero(t, requests[1].Temperature)
+
+	// The endpoint's answer is remembered: the next request goes straight out
+	// in the accepted form.
+	_, err = client.Chat(context.Background(), request)
+	require.NoError(t, err)
+	require.Len(t, requests, 3)
+	assert.Equal(t, 123, requests[2].MaxCompletionTokens)
+	assert.Zero(t, requests[2].MaxTokens)
+}
+
+func TestOpenAIClient_Chat_OtherBadRequestsAreNotRetried(t *testing.T) {
+	calls := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusBadRequest)
+		_ = json.NewEncoder(w).Encode(openaiError{Error: openaiErrorDetail{Message: "Invalid value for 'messages'"}})
+	}))
+	defer server.Close()
+
+	client := NewOpenAIClient("key", "gpt-4o", server.URL, 0)
+	_, err := client.Chat(context.Background(), ChatRequest{
+		MaxTokens: 5,
+		Messages:  []Message{{Role: "user", Content: "Hi"}},
+	})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "Invalid value for 'messages'")
+	assert.Equal(t, 1, calls)
+	assert.False(t, client.completionBudgetLearned.Load())
+}
+
+func TestOpenAIClient_ChatStream_ResendsWithMaxCompletionTokensWhenRejected(t *testing.T) {
+	var requests []openaiStreamRequest
+	client := NewOpenAIClient("key", "luna-prod", "https://gateway.example.com/v1", 0)
+	client.streamClient = &http.Client{
+		Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+			var req openaiStreamRequest
+			if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+				return nil, err
+			}
+			requests = append(requests, req)
+			if req.MaxTokens > 0 {
+				return &http.Response{
+					StatusCode: http.StatusBadRequest,
+					Header:     http.Header{"Content-Type": []string{"application/json"}},
+					Body:       io.NopCloser(bytes.NewReader(maxTokensRejection())),
+				}, nil
+			}
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+				Body: io.NopCloser(strings.NewReader(
+					"data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n" +
+						"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1}}\n" +
+						"data: [DONE]\n",
+				)),
+			}, nil
+		}),
+	}
+
+	var content strings.Builder
+	err := client.ChatStream(context.Background(), ChatRequest{
+		MaxTokens:   99,
+		Temperature: 0.4,
+		Messages:    []Message{{Role: "user", Content: "Hi"}},
+	}, func(event StreamEvent) {
+		if event.Type == "content" {
+			content.WriteString(event.Data.(ContentEvent).Text)
+		}
+	})
+	require.NoError(t, err)
+	assert.Equal(t, "ok", content.String())
+	require.Len(t, requests, 2)
+	assert.Equal(t, 99, requests[0].MaxTokens)
+	assert.Equal(t, 99, requests[1].MaxCompletionTokens)
+	assert.Zero(t, requests[1].MaxTokens)
+	assert.Zero(t, requests[1].Temperature)
+	assert.True(t, client.completionBudgetLearned.Load())
+}
+
 func TestOpenAIClient_SupportsThinking(t *testing.T) {
 	client := NewOpenAIClient("sk", "deepseek-reasoner", "https://api.deepseek.com", 0)
 	assert.True(t, client.SupportsThinking("deepseek-reasoner"))