From dd687dfb7b24a11f87916b4947eb94006d8fdb0e Mon Sep 17 00:00:00 2001 From: PratikDhanave Date: Fri, 24 Jul 2026 08:29:47 +0530 Subject: [PATCH] Close the streaming HTTP body on the Anthropic and OpenAI Responses paths The Anthropic Messages and OpenAI Responses streaming paths iterated the SSE stream but never called Close(), so on early consumer termination (yield returns false) or context cancellation the HTTP response body was never returned to the connection pool, leaking the underlying connection. Add a deferred stream.Close() immediately after each NewStreaming and GetStreaming call, mirroring the Chat Completions path which already does this and matching the .NET/Python SDKs that dispose the streaming response on early enumeration. --- provider/anthropicprovider/agent.go | 1 + provider/anthropicprovider/agent_test.go | 69 ++++++++++++++++++++ provider/openaiprovider/responses.go | 2 + provider/openaiprovider/responses_test.go | 76 +++++++++++++++++++++++ 4 files changed, 148 insertions(+) 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