feat: responses added to logs

Author: Pratham-Mishra04

Makefile

@@ -6,7 +6,7 @@ PORT ?= 8080
 APP_DIR ?= 
 PROMETHEUS_LABELS ?=
 LOG_STYLE ?= json
-LOG_LEVEL ?= debug
+LOG_LEVEL ?= info
 
 # Colors for output
 RED=\033[0;31m

framework/logstore/tables.go

@@ -76,10 +76,11 @@ type Log struct {
 	Model               string    `gorm:"type:varchar(255);index;not null" json:"model"`
 	InputHistory        string    `gorm:"type:text" json:"-"` // JSON serialized []schemas.ChatMessage
 	OutputMessage       string    `gorm:"type:text" json:"-"` // JSON serialized *schemas.ChatMessage
+	ResponsesOutput     string    `gorm:"type:text" json:"-"` // JSON serialized *schemas.ResponsesMessage
 	EmbeddingOutput     string    `gorm:"type:text" json:"-"` // JSON serialized [][]float32
 	Params              string    `gorm:"type:text" json:"-"` // JSON serialized *schemas.ModelParameters
 	Tools               string    `gorm:"type:text" json:"-"` // JSON serialized []schemas.Tool
-	ToolCalls           string    `gorm:"type:text" json:"-"` // JSON serialized []schemas.ToolCall
+	ToolCalls           string    `gorm:"type:text" json:"-"` // JSON serialized []schemas.ToolCall (For backward compatibility, tool calls are now in the content)
 	SpeechInput         string    `gorm:"type:text" json:"-"` // JSON serialized *schemas.SpeechInput
 	TranscriptionInput  string    `gorm:"type:text" json:"-"` // JSON serialized *schemas.TranscriptionInput
 	SpeechOutput        string    `gorm:"type:text" json:"-"` // JSON serialized *schemas.BifrostSpeech
@@ -103,10 +104,11 @@ type Log struct {
 	// Virtual fields for JSON output - these will be populated when needed
 	InputHistoryParsed        []schemas.ChatMessage                  `gorm:"-" json:"input_history,omitempty"`
 	OutputMessageParsed       *schemas.ChatMessage                   `gorm:"-" json:"output_message,omitempty"`
+	ResponsesOutputParsed     []schemas.ResponsesMessage             `gorm:"-" json:"responses_output,omitempty"`
 	EmbeddingOutputParsed     []schemas.BifrostEmbedding             `gorm:"-" json:"embedding_output,omitempty"`
 	ParamsParsed              interface{}                            `gorm:"-" json:"params,omitempty"`
 	ToolsParsed               []schemas.ChatTool                     `gorm:"-" json:"tools,omitempty"`
-	ToolCallsParsed           []schemas.ChatAssistantMessageToolCall `gorm:"-" json:"tool_calls,omitempty"`
+	ToolCallsParsed           []schemas.ChatAssistantMessageToolCall `gorm:"-" json:"tool_calls,omitempty"` // For backward compatibility, tool calls are now in the content
 	TokenUsageParsed          *schemas.LLMUsage                      `gorm:"-" json:"token_usage,omitempty"`
 	ErrorDetailsParsed        *schemas.BifrostError                  `gorm:"-" json:"error_details,omitempty"`
 	SpeechInputParsed         *schemas.SpeechInput                   `gorm:"-" json:"speech_input,omitempty"`
@@ -157,6 +159,14 @@ func (l *Log) SerializeFields() error {
 		}
 	}
 
+	if l.ResponsesOutputParsed != nil {
+		if data, err := json.Marshal(l.ResponsesOutputParsed); err != nil {
+			return err
+		} else {
+			l.ResponsesOutput = string(data)
+		}
+	}
+
 	if l.EmbeddingOutputParsed != nil {
 		if data, err := json.Marshal(l.EmbeddingOutputParsed); err != nil {
 			return err
@@ -272,6 +282,13 @@ func (l *Log) DeserializeFields() error {
 		}
 	}
 
+	if l.ResponsesOutput != "" {
+		if err := json.Unmarshal([]byte(l.ResponsesOutput), &l.ResponsesOutputParsed); err != nil {
+			// Log error but don't fail the operation - initialize as nil
+			l.ResponsesOutputParsed = []schemas.ResponsesMessage{}
+		}
+	}
+
 	if l.EmbeddingOutput != "" {
 		if err := json.Unmarshal([]byte(l.EmbeddingOutput), &l.EmbeddingOutputParsed); err != nil {
 			// Log error but don't fail the operation - initialize as nil
@@ -393,6 +410,30 @@ func (l *Log) BuildContentSummary() string {
 		}
 	}
 
+	// Add responses output content
+	if l.ResponsesOutputParsed != nil {
+		for _, msg := range l.ResponsesOutputParsed {
+			if msg.Content != nil {
+				if msg.Content.ContentStr != nil && *msg.Content.ContentStr != "" {
+					parts = append(parts, *msg.Content.ContentStr)
+				}
+				// If content blocks exist, extract text from them
+				if msg.Content.ContentBlocks != nil {
+					for _, block := range msg.Content.ContentBlocks {
+						if block.Text != nil && *block.Text != "" {
+							parts = append(parts, *block.Text)
+						}
+					}
+				}
+			}
+			if msg.ResponsesReasoning != nil {
+				for _, summary := range msg.ResponsesReasoning.Summary {
+					parts = append(parts, summary.Text)
+				}
+			}
+		}
+	}
+
 	// Add speech input content
 	if l.SpeechInputParsed != nil && l.SpeechInputParsed.Input != "" {
 		parts = append(parts, l.SpeechInputParsed.Input)

plugins/logging/main.go

@@ -5,9 +5,7 @@ package logging
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -38,20 +36,19 @@ const (
 // Context keys for logging optimization
 const (
 	DroppedCreateContextKey ContextKey = "logging-dropped"
-	CreatedTimestampKey     ContextKey = "logging-created-timestamp"
 )
 
 // UpdateLogData contains data for log entry updates
 type UpdateLogData struct {
 	Status              string
 	TokenUsage          *schemas.LLMUsage
 	Cost                *float64 // Cost in dollars from pricing plugin
-	OutputMessage       *schemas.ChatMessage
+	ChatOutput          *schemas.ChatMessage
+	ResponsesOutput     []schemas.ResponsesMessage
 	EmbeddingOutput     []schemas.BifrostEmbedding
-	ToolCalls           []schemas.ChatAssistantMessageToolCall
 	ErrorDetails        *schemas.BifrostError
-	Model               string                     // May be different from request
-	Object              string                     // May be different from request
+	Model               string
+	Object              string
 	SpeechOutput        *schemas.BifrostSpeech     // For non-streaming speech responses
 	TranscriptionOutput *schemas.BifrostTranscribe // For non-streaming transcription responses
 }
@@ -62,6 +59,7 @@ type LogMessage struct {
 	RequestID          string                             // Unique ID for the request
 	ParentRequestID    string                             // Unique ID for the parent request
 	Timestamp          time.Time                          // Of the preHook/postHook call
+	Latency            int64                              // For latency updates
 	InitialData        *InitialLogData                    // For create operations
 	SemanticCacheDebug *schemas.BifrostCacheDebug         // For semantic cache operations
 	UpdateData         *UpdateLogData                     // For update operations
@@ -100,39 +98,6 @@ type LoggerPlugin struct {
 	accumulator     *streaming.Accumulator // Accumulator for streaming chunks
 }
 
-// retryOnNotFound retries a function up to 3 times with 1-second delays if it returns logstore.ErrNotFound
-func retryOnNotFound(ctx context.Context, operation func() error) error {
-	const maxRetries = 3
-	const retryDelay = time.Second
-
-	var lastErr error
-	for attempt := 0; attempt < maxRetries; attempt++ {
-		err := operation()
-		if err == nil {
-			return nil
-		}
-
-		// Check if the error is logstore.ErrNotFound
-		if !errors.Is(err, logstore.ErrNotFound) {
-			return err
-		}
-
-		lastErr = err
-
-		// Don't wait after the last attempt
-		if attempt < maxRetries-1 {
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			case <-time.After(retryDelay):
-				// Continue to next retry
-			}
-		}
-	}
-
-	return lastErr
-}
-
 // Init creates new logger plugin with given log store
 func Init(ctx context.Context, logger schemas.Logger, logsStore logstore.LogStore, pricingManager *pricing.PricingManager) (*LoggerPlugin, error) {
 	if logsStore == nil {
@@ -269,7 +234,7 @@ func (p *LoggerPlugin) PreHook(ctx *context.Context, req *schemas.BifrostRequest
 		initialData.Params = req.TranscriptionRequest.Params
 		initialData.TranscriptionInput = req.TranscriptionRequest.Input
 	}
-	*ctx = context.WithValue(*ctx, CreatedTimestampKey, createdTimestamp)
+
 	// Queue the log creation message (non-blocking) - Using sync.Pool
 	logMsg := p.getLogMessage()
 	logMsg.Operation = LogOperationCreate
@@ -344,7 +309,13 @@ func (p *LoggerPlugin) PostHook(ctx *context.Context, result *schemas.BifrostRes
 	// Queue the log update message (non-blocking) - use same pattern for both streaming and regular
 	logMsg := p.getLogMessage()
 	logMsg.RequestID = requestID
-	logMsg.Timestamp = time.Now()
+
+	if result != nil {
+		logMsg.Latency = result.ExtraFields.Latency
+	} else {
+		logMsg.Latency = 0
+	}
+
 	// If response is nil, and there is an error, we update log with error
 	if result == nil && bifrostErr != nil {
 		// If request type is streaming, then we trigger cleanup as well
@@ -357,7 +328,7 @@ func (p *LoggerPlugin) PostHook(ctx *context.Context, result *schemas.BifrostRes
 			ErrorDetails: bifrostErr,
 		}
 		processingErr := retryOnNotFound(p.ctx, func() error {
-			return p.updateLogEntry(p.ctx, logMsg.RequestID, logMsg.Timestamp, logMsg.SemanticCacheDebug, logMsg.UpdateData)
+			return p.updateLogEntry(p.ctx, logMsg.RequestID, logMsg.Latency, logMsg.SemanticCacheDebug, logMsg.UpdateData)
 		})
 		if processingErr != nil {
 			p.logger.Error("failed to process log update for request %s: %v", logMsg.RequestID, processingErr)
@@ -389,7 +360,7 @@ func (p *LoggerPlugin) PostHook(ctx *context.Context, result *schemas.BifrostRes
 			go func() {
 				defer p.putLogMessage(logMsg) // Return to pool when done
 				processingErr := retryOnNotFound(p.ctx, func() error {
-					return p.updateStreamingLogEntry(p.ctx, logMsg.RequestID, logMsg.Timestamp, logMsg.SemanticCacheDebug, logMsg.StreamResponse, streamResponse.Type == streaming.StreamResponseTypeFinal)
+					return p.updateStreamingLogEntry(p.ctx, logMsg.RequestID, logMsg.SemanticCacheDebug, logMsg.StreamResponse, streamResponse.Type == streaming.StreamResponseTypeFinal)
 				})
 				if processingErr != nil {
 					p.logger.Error("failed to process stream update for request %s: %v", logMsg.RequestID, processingErr)
@@ -420,7 +391,7 @@ func (p *LoggerPlugin) PostHook(ctx *context.Context, result *schemas.BifrostRes
 			// Success case
 			updateData.Status = "success"
 			if result.Model != "" {
-				updateData.Model = result.Model
+				updateData.Model = result.ExtraFields.ModelRequested
 			}
 			// Update object type if available
 			if result.Object != "" {
@@ -434,7 +405,7 @@ func (p *LoggerPlugin) PostHook(ctx *context.Context, result *schemas.BifrostRes
 			if len(result.Choices) > 0 {
 				choice := result.Choices[0]
 				if choice.BifrostTextCompletionResponseChoice != nil {
-					updateData.OutputMessage = &schemas.ChatMessage{
+					updateData.ChatOutput = &schemas.ChatMessage{
 						Role: schemas.ChatMessageRoleAssistant,
 						Content: schemas.ChatMessageContent{
 							ContentStr: choice.BifrostTextCompletionResponseChoice.Text,
@@ -443,29 +414,11 @@ func (p *LoggerPlugin) PostHook(ctx *context.Context, result *schemas.BifrostRes
 				}
 				// Check if this is a non-stream response choice
 				if choice.BifrostNonStreamResponseChoice != nil {
-					updateData.OutputMessage = &choice.BifrostNonStreamResponseChoice.Message
-					// Extract tool calls if present
-					if choice.BifrostNonStreamResponseChoice.Message.ChatAssistantMessage != nil &&
-						choice.BifrostNonStreamResponseChoice.Message.ChatAssistantMessage.ToolCalls != nil {
-						updateData.ToolCalls = choice.BifrostNonStreamResponseChoice.Message.ChatAssistantMessage.ToolCalls
-					}
+					updateData.ChatOutput = &choice.BifrostNonStreamResponseChoice.Message
 				}
 			}
 			if result.ResponsesResponse != nil {
-				outputMessages := result.ResponsesResponse.Output
-				if len(outputMessages) > 0 {
-					chatMessages := schemas.ToChatMessages(outputMessages)
-					if len(chatMessages) > 0 {
-						lastMessage := chatMessages[len(chatMessages)-1]
-						updateData.OutputMessage = &lastMessage
-
-						// Extract tool calls if present
-						if lastMessage.ChatAssistantMessage != nil &&
-							lastMessage.ChatAssistantMessage.ToolCalls != nil {
-							updateData.ToolCalls = lastMessage.ChatAssistantMessage.ToolCalls
-						}
-					}
-				}
+				updateData.ResponsesOutput = result.ResponsesResponse.Output
 			}
 			if result.Data != nil {
 				updateData.EmbeddingOutput = result.Data
@@ -519,7 +472,7 @@ func (p *LoggerPlugin) PostHook(ctx *context.Context, result *schemas.BifrostRes
 			}
 			// Here we pass plugin level context for background processing to avoid context cancellation
 			processingErr := retryOnNotFound(p.ctx, func() error {
-				return p.updateLogEntry(p.ctx, logMsg.RequestID, logMsg.Timestamp, logMsg.SemanticCacheDebug, logMsg.UpdateData)
+				return p.updateLogEntry(p.ctx, logMsg.RequestID, 0, logMsg.SemanticCacheDebug, logMsg.UpdateData)
 			})
 			if processingErr != nil {
 				p.logger.Error("failed to process log update for request %s: %v", logMsg.RequestID, processingErr)
@@ -553,92 +506,3 @@ func (p *LoggerPlugin) Cleanup() error {
 	// GORM handles connection cleanup automatically
 	return nil
 }
-
-// Helper methods
-
-// determineObjectType determines the object type from request input
-func (p *LoggerPlugin) determineObjectType(requestType schemas.RequestType) string {
-	switch requestType {
-	case schemas.TextCompletionRequest, schemas.TextCompletionStreamRequest:
-		return "text.completion"
-	case schemas.ChatCompletionRequest:
-		return "chat.completion"
-	case schemas.ChatCompletionStreamRequest:
-		return "chat.completion.chunk"
-	case schemas.ResponsesRequest:
-		return "response"
-	case schemas.ResponsesStreamRequest:
-		return "response.completion.chunk"
-	case schemas.EmbeddingRequest:
-		return "list"
-	case schemas.SpeechRequest:
-		return "audio.speech"
-	case schemas.SpeechStreamRequest:
-		return "audio.speech.chunk"
-	case schemas.TranscriptionRequest:
-		return "audio.transcription"
-	case schemas.TranscriptionStreamRequest:
-		return "audio.transcription.chunk"
-	}
-	return "unknown"
-}
-
-// extractInputHistory extracts input history from request input
-// extractInputHistory extracts input history from request input
-func (p *LoggerPlugin) extractInputHistory(request *schemas.BifrostRequest) []schemas.ChatMessage {
-	if request.ChatRequest != nil {
-		return request.ChatRequest.Input
-	}
-	if request.ResponsesRequest != nil {
-		messages := schemas.ToChatMessages(request.ResponsesRequest.Input)
-		if len(messages) > 0 {
-			return messages
-		}
-	}
-	if request.TextCompletionRequest != nil {
-		var text string
-		if request.TextCompletionRequest.Input.PromptStr != nil {
-			text = *request.TextCompletionRequest.Input.PromptStr
-		} else {
-			var stringBuilder strings.Builder
-			for _, prompt := range request.TextCompletionRequest.Input.PromptArray {
-				stringBuilder.WriteString(prompt)
-			}
-			text = stringBuilder.String()
-		}
-		return []schemas.ChatMessage{
-			{
-				Role: schemas.ChatMessageRoleUser,
-				Content: schemas.ChatMessageContent{
-					ContentStr: &text,
-				},
-			},
-		}
-	}
-	if request.EmbeddingRequest != nil {
-		texts := request.EmbeddingRequest.Input.Texts
-
-		if len(texts) == 0 && request.EmbeddingRequest.Input.Text != nil {
-			texts = []string{*request.EmbeddingRequest.Input.Text}
-		}
-
-		contentBlocks := make([]schemas.ChatContentBlock, len(texts))
-		for i, text := range texts {
-			// Create a per-iteration copy to avoid reusing the same memory address
-			t := text
-			contentBlocks[i] = schemas.ChatContentBlock{
-				Type: schemas.ChatContentBlockTypeText,
-				Text: &t,
-			}
-		}
-		return []schemas.ChatMessage{
-			{
-				Role: schemas.ChatMessageRoleUser,
-				Content: schemas.ChatMessageContent{
-					ContentBlocks: contentBlocks,
-				},
-			},
-		}
-	}
-	return []schemas.ChatMessage{}
-}