diff --git a/AGENTS.md b/AGENTS.md index 35dca3e..c7c84ca 100644 --- a/AGENTS.md +++ b/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. diff --git a/README.md b/README.md index 1ff5b97..389aa2f 100644 --- a/README.md +++ b/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 diff --git a/chat.go b/chat.go index 66668d9..3db975c 100644 --- a/chat.go +++ b/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() diff --git a/models_flow_test.go b/models_flow_test.go index 7153e1e..5c16281 100644 --- a/models_flow_test.go +++ b/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) } } diff --git a/non_sse_test.go b/non_sse_test.go new file mode 100644 index 0000000..242c0f6 --- /dev/null +++ b/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) + } +}