plugins v2 architecture

Author: akshaydeo

.github/workflows/release-pipeline.yml

@@ -3,7 +3,7 @@ name: Release Pipeline
 # Triggers automatically on push to main when any version file changes
 on:
   push:
-    branches: ["main"]
+    branches: ["main", "v1.4.0"]
 
 # Prevent concurrent runs
 concurrency:
@@ -594,7 +594,7 @@ jobs:
           fi
 
           # Build the message with proper formatting
-          MESSAGE=$(printf "🚀 **Release Pipeline Complete**\n\n**Components:**\n• Core: %s\n• Framework: %s\n• Plugins: %s\n• Bifrost HTTP: %s\n\n**Details:**\n• Branch: \`main\`\n• Commit: \`%.8s\`\n• Author: %s\n\n[View Workflow Run](${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }})" "$CORE_STATUS" "$FRAMEWORK_STATUS" "$PLUGINS_STATUS" "$BIFROST_STATUS" "${{ github.sha }}" "${{ github.actor }}")
+          MESSAGE=$(printf "🚀 **Release Pipeline Complete**\n\n**Components:**\n• Core: %s\n• Framework: %s\n• Plugins: %s\n• Bifrost HTTP: %s\n\n**Details:**\n• Branch: \`${{ github.ref_name }}\`\n• Commit: \`%.8s\`\n• Author: %s\n\n[View Workflow Run](${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }})" "$CORE_STATUS" "$FRAMEWORK_STATUS" "$PLUGINS_STATUS" "$BIFROST_STATUS" "${{ github.sha }}" "${{ github.actor }}")
 
           payload="$(jq -n --arg content "$MESSAGE" '{content:$content}')"
           curl -sS -H "Content-Type: application/json" -d "$payload" "$DISCORD_WEBHOOK"

.github/workflows/scripts/push-mintlify-changelog.sh

@@ -236,7 +236,8 @@ if ! grep -q "\"$route\"" docs/docs.json; then
 fi
 
 # Pulling again before committing
-git pull origin main
+CURRENT_BRANCH="$(git rev-parse --abbrev-ref HEAD)"
+git pull origin "$CURRENT_BRANCH"
 # Commit and push changes
 git add docs/changelogs/$VERSION.mdx
 git add docs/docs.json
@@ -247,4 +248,4 @@ done
 git config user.name "github-actions[bot]"
 git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
 git commit -m "Adds changelog for $VERSION --skip-pipeline"
-git push origin main
+git push origin "$CURRENT_BRANCH"

.github/workflows/scripts/release-bifrost-http.sh

@@ -358,7 +358,8 @@ echo "✅ Transport build validation successful"
 
 # Commit and push changes if any
 # First, pull latest changes to avoid conflicts
-git pull origin main
+CURRENT_BRANCH="$(git rev-parse --abbrev-ref HEAD)"
+git pull origin "$CURRENT_BRANCH"
 
 # Stage any changes made to transports/
 git add transports/

core/bifrost.go

@@ -68,6 +68,8 @@ func TestExecuteRequestWithRetries_SuccessScenarios(t *testing.T) {
 			schemas.ChatCompletionRequest,
 			schemas.OpenAI,
 			"gpt-4",
+			schemas.DefaultTracer(),
+			nil,
 		)
 
 		if callCount != 1 {
@@ -101,6 +103,8 @@ func TestExecuteRequestWithRetries_SuccessScenarios(t *testing.T) {
 			schemas.ChatCompletionRequest,
 			schemas.OpenAI,
 			"gpt-4",
+			schemas.DefaultTracer(),
+			nil,
 		)
 
 		if callCount != 3 {
@@ -134,6 +138,8 @@ func TestExecuteRequestWithRetries_RetryLimits(t *testing.T) {
 			schemas.ChatCompletionRequest,
 			schemas.OpenAI,
 			"gpt-4",
+			schemas.DefaultTracer(),
+			nil,
 		)
 
 		// Should try: initial + 2 retries = 3 total attempts
@@ -196,6 +202,8 @@ func TestExecuteRequestWithRetries_NonRetryableErrors(t *testing.T) {
 				schemas.ChatCompletionRequest,
 				schemas.OpenAI,
 				"gpt-4",
+				schemas.DefaultTracer(),
+				nil,
 			)
 
 			if callCount != 1 {
@@ -268,6 +276,8 @@ func TestExecuteRequestWithRetries_RetryableConditions(t *testing.T) {
 				schemas.ChatCompletionRequest,
 				schemas.OpenAI,
 				"gpt-4",
+				schemas.DefaultTracer(),
+				nil,
 			)
 
 			// Should try: initial + 1 retry = 2 total attempts
@@ -496,6 +506,8 @@ func TestExecuteRequestWithRetries_LoggingAndCounting(t *testing.T) {
 		schemas.ChatCompletionRequest,
 		schemas.OpenAI,
 		"gpt-4",
+		schemas.DefaultTracer(),
+		nil,
 	)
 
 	// Verify call progression

core/bifrost_test.go

@@ -1,3 +1,35 @@
 - feat: added code mode to mcp
 - feat: added health monitoring to mcp
-- feat: added responses format tool execution support to mcp
+- feat: added responses format tool execution support to mcp
+- feat: adds central tracer for e2e tracing
+
+### BREAKING CHANGES
+
+- **Plugin Interface: TransportInterceptor removed, replaced with HTTPTransportMiddleware**
+
+  The `TransportInterceptor` method has been removed from the `Plugin` interface in `schemas/plugin.go`. All plugins must now implement `HTTPTransportMiddleware()` instead.
+
+  **Old API (removed in core v1.3.0):**
+  ```go
+  TransportInterceptor(ctx *BifrostContext, url string, headers map[string]string, body map[string]any) (map[string]string, map[string]any, error)
+  ```
+
+  **New API (core v1.3.0+):**
+  ```go
+  HTTPTransportMiddleware() BifrostHTTPMiddleware
+  // where BifrostHTTPMiddleware = func(next fasthttp.RequestHandler) fasthttp.RequestHandler
+  ```
+
+  **Key changes:**
+  - Method renamed: `TransportInterceptor` -> `HTTPTransportMiddleware`
+  - Return type changed: Now returns a middleware function instead of modified headers/body
+  - New import required: `github.com/valyala/fasthttp`
+  - Flow control: Must call `next(ctx)` explicitly to continue the middleware chain
+  - New capability: Can now intercept and modify responses (not just requests)
+
+  **Migration for plugin consumers:**
+  1. Update your plugin to implement `HTTPTransportMiddleware()` instead of `TransportInterceptor()`
+  2. If your plugin doesn't need HTTP transport interception, return `nil` from `HTTPTransportMiddleware()`
+  3. Update tests to verify the new middleware signature
+
+  See [Plugin Migration Guide](/docs/plugins/migration-guide) for complete instructions and code examples.

core/changelog.md

@@ -1135,7 +1135,6 @@ func (m *MCPManager) createInProcessConnection(config schemas.MCPClientConfig) (
 	if config.InProcessServer == nil {
 		return nil, MCPClientConnectionInfo{}, fmt.Errorf("InProcess connection requires a server instance")
 	}
-
 	// Create in-process client directly connected to the provided server
 	inProcessClient, err := client.NewInProcessClient(config.InProcessServer)
 	if err != nil {

core/mcp.go

@@ -877,12 +877,27 @@ func SendInProgressEventResponsesChunk(ctx context.Context, postHookRunner schem
 // This utility reduces code duplication across streaming implementations by encapsulating
 // the common pattern of running post hooks, handling errors, and sending responses with
 // proper context cancellation handling.
+// It also completes the deferred LLM span when the final chunk is sent (StreamEndIndicator is true).
 func ProcessAndSendResponse(
 	ctx context.Context,
 	postHookRunner schemas.PostHookRunner,
 	response *schemas.BifrostResponse,
 	responseChan chan *schemas.BifrostStream,
 ) {
+	// Accumulate chunk for tracing (common for all providers)
+	if tracer, ok := ctx.Value(schemas.BifrostContextKeyTracer).(schemas.Tracer); ok && tracer != nil {
+		if traceID, ok := ctx.Value(schemas.BifrostContextKeyTraceID).(string); ok && traceID != "" {
+			tracer.AddStreamingChunk(traceID, response)
+		}
+	}
+
+	// Check if this is the final chunk and complete deferred span if so
+	if isFinalChunk := ctx.Value(schemas.BifrostContextKeyStreamEndIndicator); isFinalChunk != nil {
+		if final, ok := isFinalChunk.(bool); ok && final {
+			completeDeferredSpan(&ctx, response, nil)
+		}
+	}
+
 	// Run post hooks on the response
 	processedResponse, processedError := postHookRunner(&ctx, response, nil)
 
@@ -913,13 +928,21 @@ func ProcessAndSendResponse(
 // This utility reduces code duplication across streaming implementations by encapsulating
 // the common pattern of running post hooks, handling errors, and sending responses with
 // proper context cancellation handling.
+// It also completes the deferred LLM span when the final chunk is sent (StreamEndIndicator is true).
 func ProcessAndSendBifrostError(
 	ctx context.Context,
 	postHookRunner schemas.PostHookRunner,
 	bifrostErr *schemas.BifrostError,
 	responseChan chan *schemas.BifrostStream,
 	logger schemas.Logger,
 ) {
+	// Check if this is the final chunk and complete deferred span if so
+	if isFinalChunk := ctx.Value(schemas.BifrostContextKeyStreamEndIndicator); isFinalChunk != nil {
+		if final, ok := isFinalChunk.(bool); ok && final {
+			completeDeferredSpan(&ctx, nil, bifrostErr)
+		}
+	}
+
 	// Send scanner error through channel
 	processedResponse, processedError := postHookRunner(&ctx, nil, bifrostErr)
 
@@ -1385,3 +1408,94 @@ func GetBudgetTokensFromReasoningEffort(
 
 	return budget, nil
 }
+
+// completeDeferredSpan completes the deferred LLM span for streaming requests.
+// This is called when the final chunk is processed (when StreamEndIndicator is true).
+// It retrieves the deferred span handle from TraceStore using the trace ID from context,
+// populates response attributes from accumulated chunks, and ends the span.
+func completeDeferredSpan(ctx *context.Context, result *schemas.BifrostResponse, err *schemas.BifrostError) {
+	if ctx == nil {
+		return
+	}
+
+	// Get the trace ID from context (this IS available in the provider's goroutine)
+	traceID, ok := (*ctx).Value(schemas.BifrostContextKeyTraceID).(string)
+	if !ok || traceID == "" {
+		return
+	}
+
+	// Get the tracer from context
+	tracerVal := (*ctx).Value(schemas.BifrostContextKeyTracer)
+	if tracerVal == nil {
+		return
+	}
+	tracer, ok := tracerVal.(schemas.Tracer)
+	if !ok || tracer == nil {
+		return
+	}
+
+	// Get the deferred span handle from TraceStore using trace ID
+	handle := tracer.GetDeferredSpanHandle(traceID)
+	if handle == nil {
+		return
+	}
+
+	// Set total latency from the final chunk
+	if result != nil {
+		extraFields := result.GetExtraFields()
+		if extraFields.Latency > 0 {
+			tracer.SetAttribute(handle, "gen_ai.response.total_latency_ms", extraFields.Latency)
+		}
+	}
+
+	// Get accumulated response with full data (content, tool calls, reasoning, etc.)
+	// This builds a complete BifrostResponse from all the streaming chunks
+	accumulatedResp, ttftMs, chunkCount := tracer.GetAccumulatedChunks(traceID)
+	if accumulatedResp != nil {
+		// Use accumulated response for attributes (includes full content, tool calls, etc.)
+		tracer.PopulateLLMResponseAttributes(handle, accumulatedResp, err)
+
+		// Set Time to First Token (TTFT) attribute
+		if ttftMs > 0 {
+			tracer.SetAttribute(handle, schemas.AttrTimeToFirstToken, ttftMs)
+		}
+
+		// Set total chunks attribute
+		if chunkCount > 0 {
+			tracer.SetAttribute(handle, schemas.AttrTotalChunks, chunkCount)
+		}
+	} else if result != nil {
+		// Fall back to final chunk if no accumulated data (shouldn't happen normally)
+		tracer.PopulateLLMResponseAttributes(handle, result, err)
+	}
+
+	// Finalize aggregated post-hook spans before ending the LLM span
+	// This creates one span per plugin with average execution time
+	// We need to set the llm.call span ID in context so post-hook spans become its children
+	if finalizer, ok := (*ctx).Value(schemas.BifrostContextKeyPostHookSpanFinalizer).(func(context.Context)); ok && finalizer != nil {
+		// Get the deferred span ID (the llm.call span) to set as parent for post-hook spans
+		spanID := tracer.GetDeferredSpanID(traceID)
+		if spanID != "" {
+			finalizerCtx := context.WithValue(*ctx, schemas.BifrostContextKeySpanID, spanID)
+			finalizer(finalizerCtx)
+		} else {
+			finalizer(*ctx)
+		}
+	}
+
+	// End span with appropriate status
+	if err != nil {
+		if err.Error != nil {
+			tracer.SetAttribute(handle, "error", err.Error.Message)
+		}
+		if err.StatusCode != nil {
+			tracer.SetAttribute(handle, "status_code", *err.StatusCode)
+		}
+		tracer.EndSpan(handle, schemas.SpanStatusError, "streaming request failed")
+	} else {
+		tracer.EndSpan(handle, schemas.SpanStatusOk, "")
+	}
+
+	// Clear the deferred span from TraceStore
+	tracer.ClearDeferredSpan(traceID)
+}

core/providers/utils/utils.go

@@ -22,6 +22,7 @@ type BifrostConfig struct {
 	Account            Account
 	Plugins            []Plugin
 	Logger             Logger
+	Tracer             Tracer      // Tracer for distributed tracing (nil = NoOpTracer)
 	InitialPoolSize    int         // Initial pool size for sync pools in Bifrost. Higher values will reduce memory allocations but will increase memory usage.
 	DropExcessRequests bool        // If true, in cases where the queue is full, requests will not wait for the queue to be empty and will be dropped instead.
 	MCPConfig          *MCPConfig  // MCP (Model Context Protocol) configuration for tool integration
@@ -139,6 +140,13 @@ const (
 	BifrostMCPAgentOriginalRequestID                     BifrostContextKey = "bifrost-mcp-agent-original-request-id"            // string (to store the original request ID for MCP agent mode)
 	BifrostContextKeyStructuredOutputToolName            BifrostContextKey = "bifrost-structured-output-tool-name"              // string (to store the name of the structured output tool (set by bifrost))
 	BifrostContextKeyUserAgent                           BifrostContextKey = "bifrost-user-agent"                               // string (set by bifrost)
+	BifrostContextKeyTraceID                             BifrostContextKey = "bifrost-trace-id"                                 // string (trace ID for distributed tracing - set by tracing middleware)
+	BifrostContextKeySpanID                              BifrostContextKey = "bifrost-span-id"                                  // string (current span ID for child span creation - set by tracer)
+	BifrostContextKeyStreamStartTime                     BifrostContextKey = "bifrost-stream-start-time"                        // time.Time (start time for streaming TTFT calculation - set by bifrost)
+	BifrostContextKeyTracer                              BifrostContextKey = "bifrost-tracer"                                   // Tracer (tracer instance for completing deferred spans - set by bifrost)
+	BifrostContextKeyDeferTraceCompletion                BifrostContextKey = "bifrost-defer-trace-completion"                   // bool (signals trace completion should be deferred for streaming - set by streaming handlers)
+	BifrostContextKeyTraceCompleter                      BifrostContextKey = "bifrost-trace-completer"                          // func() (callback to complete trace after streaming - set by tracing middleware)
+	BifrostContextKeyPostHookSpanFinalizer               BifrostContextKey = "bifrost-posthook-span-finalizer"                  // func(context.Context) (callback to finalize post-hook spans after streaming - set by bifrost)
 )
 
 // NOTE: for custom plugin implementation dealing with streaming short circuit,

core/schemas/bifrost.go

@@ -23,6 +23,7 @@ var reservedKeys = []any{
 	BifrostContextKeySkipKeySelection,
 	BifrostContextKeyExtraHeaders,
 	BifrostContextKeyURLPath,
+	BifrostContextKeyDeferTraceCompletion,
 }
 
 // BifrostContext is a custom context.Context implementation that tracks user-set values.