diff --git a/provider/anthropicprovider/agent.go b/provider/anthropicprovider/agent.go index 82822f09..be7b5ab2 100644 --- a/provider/anthropicprovider/agent.go +++ b/provider/anthropicprovider/agent.go @@ -123,6 +123,7 @@ func (a *client) run(ctx context.Context, messages []*message.Message, options . } return func(yield func(*agent.ResponseUpdate, error) bool) { stream := a.client.Messages.NewStreaming(ctx, params) + defer func() { _ = stream.Close() }() var messageID string var usage message.UsageDetails diff --git a/provider/anthropicprovider/agent_test.go b/provider/anthropicprovider/agent_test.go index 7ffecea2..ae12d96a 100644 --- a/provider/anthropicprovider/agent_test.go +++ b/provider/anthropicprovider/agent_test.go @@ -9,6 +9,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "github.com/anthropics/anthropic-sdk-go" @@ -575,3 +576,71 @@ func TestToolUseEmptyArgumentsSerializeAsObject(t *testing.T) { t.Fatal("tool_use block for toolu_1 not found in request") } } + +// countingReadCloser counts Close calls on an HTTP response body. +type countingReadCloser struct { + io.ReadCloser + closes *atomic.Int64 +} + +func (c *countingReadCloser) Close() error { + c.closes.Add(1) + return c.ReadCloser.Close() +} + +// closeCountingTransport wraps each response body so tests can assert the +// streaming HTTP body is released once the run completes. +type closeCountingTransport struct { + base http.RoundTripper + closes *atomic.Int64 +} + +func (t *closeCountingTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.base.RoundTrip(req) + if err != nil || resp == nil { + return resp, err + } + resp.Body = &countingReadCloser{ReadCloser: resp.Body, closes: t.closes} + return resp, nil +} + +// TestStreamingClosesResponseBody verifies the streaming path releases the HTTP +// response body when the consumer stops iterating early. Without an explicit +// stream.Close(), the body is never returned to the pool, leaking the +// underlying connection. This mirrors the defer-close already present on the +// Chat Completions streaming path and matches the .NET/Python SDKs, which +// dispose the streaming response on early enumeration. +func TestStreamingClosesResponseBody(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, minimalStreamingResponse("hello world")) + })) + defer server.Close() + + var closes atomic.Int64 + httpClient := &http.Client{Transport: &closeCountingTransport{base: http.DefaultTransport, closes: &closes}} + a := anthropicprovider.NewAgent( + anthropic.NewClient( + option.WithBaseURL(server.URL), + option.WithAPIKey("test"), + option.WithHTTPClient(httpClient), + ), + anthropicprovider.AgentConfig{ + Model: "claude-3-5-sonnet-20241022", + Config: agent.Config{DisableFuncAutoCall: true}, + }, + ) + + // Stop iterating after the first streamed update. The provider's run + // closure then returns via yield=false, which must close the body. + for _, err := range a.RunText(t.Context(), "hi", agent.Stream(true)) { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + break + } + + if got := closes.Load(); got == 0 { + t.Fatal("streaming response body was not closed after early consumer exit") + } +} diff --git a/provider/openaiprovider/responses.go b/provider/openaiprovider/responses.go index 1b712031..168952db 100644 --- a/provider/openaiprovider/responses.go +++ b/provider/openaiprovider/responses.go @@ -130,6 +130,7 @@ func (a *responsesClient) run(ctx context.Context, messages []*message.Message, streamResp := a.client.Responses.GetStreaming(ctx, ct.ResponseID, responses.ResponseGetParams{ StartingAfter: openai.Int(ct.SequenceNumber), }, telemetryRequestOption) + defer func() { _ = streamResp.Close() }() // Update conversation ID when resuming updateConversationID(ct.ResponseID) for streamResp.Next() { @@ -171,6 +172,7 @@ func (a *responsesClient) run(ctx context.Context, messages []*message.Message, if stream { // Create streaming response streamResp := a.client.Responses.NewStreaming(ctx, body, telemetryRequestOption) + defer func() { _ = streamResp.Close() }() responseID := "" createdAt := time.Time{} isBackground, _ := agent.GetOption(options, agent.AllowBackgroundResponses) diff --git a/provider/openaiprovider/responses_test.go b/provider/openaiprovider/responses_test.go index 4f060310..29645c73 100644 --- a/provider/openaiprovider/responses_test.go +++ b/provider/openaiprovider/responses_test.go @@ -10,6 +10,7 @@ import ( "net/http/httptest" "reflect" "strings" + "sync/atomic" "testing" "time" @@ -5524,6 +5525,81 @@ func TestResponsesMultipleRequiredFunctions(t *testing.T) { } } +// countingReadCloser counts Close calls on an HTTP response body. +type countingReadCloser struct { + io.ReadCloser + closes *atomic.Int64 +} + +func (c *countingReadCloser) Close() error { + c.closes.Add(1) + return c.ReadCloser.Close() +} + +// closeCountingTransport wraps each response body so tests can assert the +// streaming HTTP body is released once the run completes. +type closeCountingTransport struct { + base http.RoundTripper + closes *atomic.Int64 +} + +func (t *closeCountingTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.base.RoundTrip(req) + if err != nil || resp == nil { + return resp, err + } + resp.Body = &countingReadCloser{ReadCloser: resp.Body, closes: t.closes} + return resp, nil +} + +// TestResponsesStreamingClosesResponseBody verifies the Responses streaming path +// releases the HTTP response body when the consumer stops iterating early. +// Without an explicit streamResp.Close(), the body is never returned to the +// pool, leaking the underlying connection. This mirrors the defer-close already +// present on the Chat Completions streaming path and matches the .NET/Python +// SDKs, which dispose the streaming response on early enumeration. +func TestResponsesStreamingClosesResponseBody(t *testing.T) { + const output = `event: response.created +data: {"type":"response.created","response":{"id":"resp_close_test","object":"response","created_at":1741892091,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"gpt-4o-mini-2024-07-18","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"generate_summary":null},"store":true,"temperature":1.0,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1.0,"usage":null,"user":null,"metadata":{}}} + +event: response.output_text.delta +data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"Hello"} + +event: response.output_text.delta +data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":" world"} + +` + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, output) + })) + defer server.Close() + + var closes atomic.Int64 + httpClient := &http.Client{Transport: &closeCountingTransport{base: http.DefaultTransport, closes: &closes}} + a := openaiprovider.NewResponsesAgent( + openai.NewClient(option.WithBaseURL(server.URL), option.WithHTTPClient(httpClient)), + openaiprovider.AgentConfig{ + Model: "gpt-4o-mini", + Config: agent.Config{DisableFuncAutoCall: true}, + }, + ) + + // Stop iterating after the first streamed update. The provider's run + // closure then returns via yield=false, which must close the body. + for _, err := range a.RunText(t.Context(), "hi", agent.Stream(true)) { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + break + } + + if got := closes.Load(); got == 0 { + t.Fatal("streaming response body was not closed after early consumer exit") + } +} + func responsesBodyEqual(t *testing.T, got string, want string) { t.Helper() var gotObj any