fix: preserve JSON responses to streaming requests

AGENTS.md

@@ -43,6 +43,7 @@ The canonical type system is OpenAI-shaped; `gemini.go`/`anthropic.go` translate
 1. **Canonical finish reasons** are `stop | length | tool_calls | content_filter | ""`. Unmapped provider stop reasons map to `""` on every format — provider-specific strings never leak.
 2. **API keys never appear** in error text, `String()`, or any typed error.
 3. **Streaming**: retries only before the first emitted delta; a failure after partial output returns the partial `*ChatResult` + wrapped error and is **never retried**; a premature close (no completion signal) is an error, never a silent empty success; the parser goroutine is always released (abort-safe `done` protocol).
+   A successful non-SSE response is parsed directly with the usual body cap and format mapper. It learns buffered mode for future requests, but never discards the current generation or makes a replacement request; parse/read errors remain terminal for that response.
 4. **Unknown message roles are rejected** at the SDK boundary (`ConfigError`) — never dropped or reinterpreted per format.
 5. **Unknown provider data stays unknown** (zero values) — no static guesses, no fallback model tables.
 6. **Learn-once fallback state is per-`Provider`**, monotonic, atomic, shared by every `ChatClient` — never move it back to per-client.

README.md

@@ -205,7 +205,7 @@ When a provider rejects a request pattern, the SDK learns the constraint **once
 | Names `/v1/responses` as the tools+reasoning path | retry on `POST /responses` (keeps reasoning on) |
 | Rejects `reasoning_effort` + tools (legacy) | pin `reasoning_effort: "none"` |
 | Rejects streaming itself | downgrade to buffered calls permanently |
-| Answers a streamed request with a non-SSE body | downgrade to buffered calls permanently |
+| Answers a streamed request with a non-SSE body | consume that JSON response directly, then use buffered calls permanently (no duplicate generation) |
 
 ## Retry policy
 

chat.go

@@ -106,9 +106,6 @@ var errStreamStop = errors.New("llm: stream complete")
 // completion signal; retryable only before the first delta.
 var errPrematureClose = errors.New("llm: provider closed the stream before completion")
 
-// errNonSSE marks a streamed request answered with a regular body.
-var errNonSSE = errors.New("llm: provider answered a streamed request with a non-event-stream body")
-
 // consumerAbort wraps a delta-handler error through pumpSSE.
 type consumerAbort struct{ err error }
 
@@ -526,6 +523,11 @@ func (pc *providerClient) callStream(ctx context.Context, req *ChatRequest, mode
 
 		out := pc.attemptStream(deadlineCtx, url, body, attemptMapper, onDelta, len(req.Tools) > 0)
 		switch {
+		case out.bufferedResponse:
+			// A successful HTTP response already contains this generation.
+			// Return its buffered parse (including errors) without replacing
+			// it with another generation or resetting the request deadline.
+			return out.result, out.err
 		case out.success():
 			return out.result, nil
 		case out.abort != nil:
@@ -585,12 +587,13 @@ func (pc *providerClient) callStream(ctx context.Context, req *ChatRequest, mode
 // streamOutcome is one streaming attempt's result. Exactly one of the
 // terminal fields is meaningful; see the cases in callStream.
 type streamOutcome struct {
-	result     *ChatResult         // final (success) or partial (abort/fail)
-	abort      *StreamAbortedError // consumer aborted; result is partial
-	apiErr     *APIError           // provider 4xx/5xx (retryable flag inside)
-	err        error               // transport / watchdog / parse error
-	retryAfter time.Duration       // Retry-After hint when apiErr is a 429
-	learnRetry bool                // learn-once flag set; retry immediately
+	result           *ChatResult         // final (success) or partial (abort/fail)
+	abort            *StreamAbortedError // consumer aborted; result is partial
+	apiErr           *APIError           // provider 4xx/5xx (retryable flag inside)
+	err              error               // transport / watchdog / parse error
+	retryAfter       time.Duration       // Retry-After hint when apiErr is a 429
+	learnRetry       bool                // learn-once flag set; retry immediately
+	bufferedResponse bool                // successful non-SSE body consumed; never regenerate
 }
 
 func (o *streamOutcome) success() bool {
@@ -637,9 +640,18 @@ func (pc *providerClient) attemptStream(ctx context.Context, url string, body []
 	}
 
 	if ct := resp.Header.Get("Content-Type"); !strings.Contains(ct, "text/event-stream") {
-		// Provider answered a streamed request with a regular body.
+		// Learn buffered mode for later requests, but consume this generation
+		// now: discarding it would lose notes and tool calls, and bill twice.
 		pc.learn.forceBuffered.Store(true)
-		return streamOutcome{learnRetry: true, err: errNonSSE}
+		data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseSize+1))
+		if err != nil {
+			return streamOutcome{bufferedResponse: true, err: err}
+		}
+		if len(data) > maxResponseSize {
+			return streamOutcome{bufferedResponse: true, err: fmt.Errorf("llm: response exceeds %d bytes", maxResponseSize)}
+		}
+		result, err := pc.parseResponse(data, url)
+		return streamOutcome{bufferedResponse: true, result: result, err: err}
 	}
 
 	acc := newStreamAccum()

models_flow_test.go

@@ -145,7 +145,7 @@ func TestCallDefinitiveFailureAfter429(t *testing.T) {
 }
 
 // A streamed request answered with a non-SSE body learns the buffered
-// fallback mid-stream and completes on the buffered path.
+// fallback while consuming that generation directly.
 func TestCallStreamLearnsBufferedFromNonSSEBody(t *testing.T) {
 	var n int
 	srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) {
@@ -164,8 +164,8 @@ func TestCallStreamLearnsBufferedFromNonSSEBody(t *testing.T) {
 	if res.Content != "buffered" {
 		t.Errorf("res = %q", res.Content)
 	}
-	if n != 2 {
-		t.Errorf("requests = %d, want 2 (learn, then buffered retry)", n)
+	if n != 1 {
+		t.Errorf("requests = %d, want 1 (consume the successful body)", n)
 	}
 	// The learned fallback must short-circuit the NEXT CallStream of the
 	// same client straight onto the buffered path.
@@ -175,8 +175,8 @@ func TestCallStreamLearnsBufferedFromNonSSEBody(t *testing.T) {
 	if err != nil || res2.Content != "buffered" {
 		t.Fatalf("second stream: res=%q err=%v", res2.Content, err)
 	}
-	if n != 3 {
-		t.Errorf("requests = %d, want 3 (entry fast-path, no re-learn)", n)
+	if n != 2 {
+		t.Errorf("requests = %d, want 2 (entry fast-path, no re-learn)", n)
 	}
 }
 

non_sse_test.go

@@ -0,0 +1,112 @@
+package llm
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestCallStream_NonSSEPreservesFirstGeneration(t *testing.T) {
+	for _, responses := range []bool{false, true} {
+		t.Run(fmt.Sprint(responses), func(t *testing.T) {
+			requests := 0
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				requests++
+				var request map[string]any
+				if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+					t.Error(err)
+				}
+				stream, _ := request["stream"].(bool)
+				if stream != (requests == 1) {
+					t.Errorf("request %d stream=%v", requests, stream)
+				}
+				w.Header().Set("Content-Type", "application/json")
+				if responses {
+					fmt.Fprintf(w, `{"status":"completed","output":[{"type":"reasoning","encrypted_content":"signature","summary":[{"type":"summary_text","text":"Plan"}]},{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Generation %d"}]},{"type":"function_call","call_id":"call","name":"inspect","arguments":"{}"}],"usage":{"input_tokens":12,"output_tokens":3}}`, requests)
+				} else {
+					fmt.Fprintf(w, `{"choices":[{"message":{"content":"Generation %d","reasoning_content":"Plan","tool_calls":[{"id":"call","type":"function","function":{"name":"inspect","arguments":"{}"}}]},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":12,"completion_tokens":3}}`, requests)
+				}
+			}))
+			defer srv.Close()
+			model := "gpt-4o"
+			if responses {
+				model = "gpt-5.6-luna"
+			}
+			cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv)
+			cc.model = model
+			req := &ChatRequest{Model: model, Messages: []Message{{Role: RoleUser, Content: "hi"}}, Tools: []ToolDef{{Name: "inspect", Parameters: json.RawMessage(`{"type":"object"}`)}}}
+			for i := 1; i <= 2; i++ {
+				res, err := cc.CallStream(context.Background(), req, func(Delta) error { t.Error("buffered body emitted a delta"); return nil })
+				if err != nil {
+					t.Fatal(err)
+				}
+				if res.Content != fmt.Sprintf("Generation %d", i) || requests != i {
+					t.Fatalf("first generation discarded: response=%+v requests=%d", res, requests)
+				}
+				if res.ReasoningContent != "Plan" || len(res.ToolCalls) != 1 || res.ToolCalls[0].Name != "inspect" || res.FinishReason != FinishToolCalls || res.Usage.PromptTokens != 12 || res.Usage.CompletionTokens != 3 {
+					t.Fatalf("response metadata lost: %+v", res)
+				}
+				if responses && res.ThinkingSignature != "signature" {
+					t.Errorf("signature lost: %+v", res)
+				}
+			}
+		})
+	}
+}
+
+type nonSSETransport struct{ body io.ReadCloser }
+
+func (tr nonSSETransport) RoundTrip(*http.Request) (*http.Response, error) {
+	return &http.Response{StatusCode: 200, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: tr.body}, nil
+}
+
+type nonSSELargeReader struct{ bytes int }
+
+func (r *nonSSELargeReader) Read(p []byte) (int, error) { r.bytes += len(p); return len(p), nil }
+
+func TestAttemptStream_NonSSEBodyBounds(t *testing.T) {
+	large := &nonSSELargeReader{}
+	for _, tc := range []struct {
+		name      string
+		reader    io.Reader
+		errorText string
+	}{
+		{"read_failure", errReader{}, "socket gone"},
+		{"oversized", large, "response exceeds"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			client := &http.Client{Transport: nonSSETransport{body: io.NopCloser(tc.reader)}}
+			pc := newProviderClient(ProviderConfig{Format: FormatOpenAI}, client, client)
+			out := pc.attemptStream(context.Background(), "http://provider.test/chat/completions", []byte(`{}`), mapOpenAIStreamEvent, func(Delta) error { t.Error("unexpected delta"); return nil }, false)
+			if !out.bufferedResponse || out.result != nil || out.err == nil || !strings.Contains(out.err.Error(), tc.errorText) {
+				t.Fatalf("outcome=%+v", out)
+			}
+			if !pc.learn.forceBuffered.Load() {
+				t.Error("buffered mode not learned")
+			}
+		})
+	}
+	if large.bytes != maxResponseSize+1 {
+		t.Errorf("read=%d, bound=%d", large.bytes, maxResponseSize+1)
+	}
+}
+
+func TestCallStream_NonSSEMalformedBody(t *testing.T) {
+	requests := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, `{"choices":`)
+	}))
+	defer srv.Close()
+	cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv)
+	res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { return nil })
+	if err == nil || !strings.Contains(err.Error(), "parse") || res != nil || requests != 1 {
+		t.Fatalf("res=%+v err=%v requests=%d", res, err, requests)
+	}
+}