](https://app.getpostman.com/run-collection/31642484-2ba0e658-4dcd-49f4-845a-0c7ed745b916?action=collection%2Ffork&source=rip_markdown&collection-url=entityId%3D31642484-2ba0e658-4dcd-49f4-845a-0c7ed745b916%26entityType%3Dcollection%26workspaceId%3D63e853c8-9aec-477f-909c-7f02f543150e)
diff --git a/cli/go.mod b/cli/go.mod
index c887aac505..f7260a3b86 100644
--- a/cli/go.mod
+++ b/cli/go.mod
@@ -1,6 +1,6 @@
module github.com/maximhq/bifrost/cli
-go 1.26.2
+go 1.26.1
require (
github.com/bytedance/sonic v1.15.0
diff --git a/core/bifrost.go b/core/bifrost.go
index 17543d4cbe..a386c7393a 100644
--- a/core/bifrost.go
+++ b/core/bifrost.go
@@ -4689,8 +4689,16 @@ func (bifrost *Bifrost) tryStreamRequest(ctx *schemas.BifrostContext, req *schem
// shared processedResponse or processedError objects.
streamResponse := providerUtils.BuildClientStreamChunk(ctx, processedResponse, processedError)
- // Send the processed message to the output stream
- outputStream <- streamResponse
+ // Guarded send: if the consumer abandons outputStream (client
+ // disconnect, ctx cancel), drain the upstream shortCircuit.Stream
+ // so its producer can exit cleanly instead of blocking on its send.
+ select {
+ case outputStream <- streamResponse:
+ case <-ctx.Done():
+ for range shortCircuit.Stream {
+ }
+ return
+ }
//TODO: Release the processed response immediately after use
}
@@ -4952,7 +4960,7 @@ func executeRequestWithRetries[T any](
// the SSE stream instead of returning proper HTTP error status codes.
if bifrostError == nil {
if streamChan, ok := any(result).(chan *schemas.BifrostStreamChunk); ok {
- checkedStream, drainDone, firstChunkErr := providerUtils.CheckFirstStreamChunkForError(streamChan)
+ checkedStream, drainDone, firstChunkErr := providerUtils.CheckFirstStreamChunkForError(ctx, streamChan)
if firstChunkErr != nil {
<-drainDone
bifrostError = firstChunkErr
@@ -5216,12 +5224,16 @@ func (bifrost *Bifrost) requestWorker(provider schemas.Provider, config *schemas
}
return resp, nil
}
- // Store a finalizer callback to create aggregated post-hook spans at stream end
- // This closure captures the pipeline reference and releases it after finalization
+ // Store a finalizer callback to create aggregated post-hook spans at stream end.
+ // Wrapped in sync.Once so the normal end-of-stream invocation and a deferred
+ // safety-net invocation (e.g. from a provider goroutine's panic path) cannot
+ // double-release the pipeline.
+ var finalizerOnce sync.Once
postHookSpanFinalizer := func(ctx context.Context) {
- pipeline.FinalizeStreamingPostHookSpans(ctx)
- // Release the pipeline AFTER finalizing spans (not before streaming completes)
- bifrost.releasePluginPipeline(pipeline)
+ finalizerOnce.Do(func() {
+ pipeline.FinalizeStreamingPostHookSpans(ctx)
+ bifrost.releasePluginPipeline(pipeline)
+ })
}
req.Context.SetValue(schemas.BifrostContextKeyPostHookSpanFinalizer, postHookSpanFinalizer)
}
diff --git a/core/go.mod b/core/go.mod
index 924e204ea2..b85c403ec6 100644
--- a/core/go.mod
+++ b/core/go.mod
@@ -1,6 +1,6 @@
module github.com/maximhq/bifrost/core
-go 1.26.2
+go 1.26.1
require (
cloud.google.com/go v0.123.0
diff --git a/core/providers/anthropic/anthropic.go b/core/providers/anthropic/anthropic.go
index ca23445d82..e012f3a13f 100644
--- a/core/providers/anthropic/anthropic.go
+++ b/core/providers/anthropic/anthropic.go
@@ -701,6 +701,7 @@ func HandleAnthropicChatCompletionStreaming(
// Start streaming in a goroutine
go func() {
+ defer providerUtils.EnsureStreamFinalizerCalled(ctx)
defer func() {
model := "unknown"
if meta != nil {
@@ -1187,6 +1188,7 @@ func HandleAnthropicResponsesStream(
// Start streaming in a goroutine
go func() {
+ defer providerUtils.EnsureStreamFinalizerCalled(ctx)
defer func() {
model := "