feat: add embedding_input column and ui display for embedding request inputs

framework/logstore/contenthidden_test.go

@@ -28,6 +28,9 @@ func newContentHiddenTestEntry(id string) *Log {
 		OutputMessageParsed: &schemas.ChatMessage{
 			Content: &schemas.ChatMessageContent{ContentStr: strPtr("sensitive answer")},
 		},
+		EmbeddingInputParsed: []schemas.EmbeddingInputItem{
+			{Content: schemas.EmbeddingContent{{Type: schemas.EmbeddingContentPartTypeText, Text: strPtr("sensitive embedding")}}},
+		},
 	}
 }
 
@@ -57,6 +60,7 @@ func TestHybrid_ContentHiddenStripsDBRowAndSkipsHydration(t *testing.T) {
 	assert.True(t, dbRow.HasObject)
 	assert.Empty(t, dbRow.InputHistory)
 	assert.Empty(t, dbRow.OutputMessage)
+	assert.Empty(t, dbRow.EmbeddingInput)
 	assert.Empty(t, dbRow.ContentSummary)
 	assert.NotEmpty(t, dbRow.TokenUsage)
 	require.NotNil(t, dbRow.TokenUsageParsed)
@@ -75,6 +79,8 @@ func TestHybrid_ContentHiddenStripsDBRowAndSkipsHydration(t *testing.T) {
 	assert.True(t, found.ContentHidden)
 	assert.Empty(t, found.InputHistory, "hidden log must not be hydrated")
 	assert.Empty(t, found.OutputMessage, "hidden log must not be hydrated")
+	assert.Empty(t, found.EmbeddingInput, "hidden log must not be hydrated")
+	assert.Nil(t, found.EmbeddingInputParsed, "hidden log must not be hydrated")
 	assert.Empty(t, found.ContentSummary)
 }
 

framework/logstore/hybrid_test.go

@@ -144,6 +144,38 @@ func TestHybrid_CreateAndFindByID(t *testing.T) {
 	assert.Contains(t, found.ContentSummary, "Hello, how are you?")
 }
 
+// TestHybrid_EmbeddingInputOffloaded pins that embedding_input leaves the DB row and is hydrated from the object.
+func TestHybrid_EmbeddingInputOffloaded(t *testing.T) {
+	hybrid, inner, _ := newTestHybrid(t)
+	defer hybrid.Close(context.Background())
+	ctx := context.Background()
+
+	text := "embed me"
+	entry := &Log{
+		ID:        "emb-1",
+		Timestamp: time.Now().UTC(),
+		Provider:  "openai",
+		Model:     "text-embedding-3-small",
+		Status:    "success",
+		Object:    "embedding",
+		EmbeddingInputParsed: []schemas.EmbeddingInputItem{
+			{Content: schemas.EmbeddingContent{{Type: schemas.EmbeddingContentPartTypeText, Text: &text}}},
+		},
+	}
+	require.NoError(t, entry.SerializeFields())
+	require.NoError(t, hybrid.CreateIfNotExists(ctx, entry))
+	waitForOffload(t, inner, "emb-1")
+
+	dbRow, err := inner.FindByID(ctx, "emb-1")
+	require.NoError(t, err)
+	assert.Empty(t, dbRow.EmbeddingInput, "embedding_input must be offloaded, not kept in the DB row")
+
+	found, err := hybrid.FindByID(ctx, "emb-1")
+	require.NoError(t, err)
+	require.Len(t, found.EmbeddingInputParsed, 1, "embedding_input should be hydrated from object storage")
+	assert.Equal(t, text, *found.EmbeddingInputParsed[0].Content[0].Text)
+}
+
 func TestHybrid_EmptyPayloadSkipsUpload(t *testing.T) {
 	hybrid, _, objStore := newTestHybrid(t)
 	defer hybrid.Close(context.Background())

framework/logstore/migrations.go

@@ -326,6 +326,7 @@ var logstoreMigrationSteps = []migrationStep{
 	{IDs: []string{"logs_add_warp_conversation_tables"}, run: migrationAddWarpConversationTables},
 	{IDs: []string{"logs_add_warp_conversations_updated_at_index"}, run: migrationAddWarpConversationsUpdatedAtIndex},
 	{IDs: []string{"logs_add_warp_message_outcome_columns"}, run: migrationAddWarpMessageOutcomeColumns},
+	{IDs: []string{"logs_add_embedding_input_column"}, run: migrationAddEmbeddingInputColumn},
 }
 
 // areThereAnyPendingMigrations returns true if there are any pending migrations to be applied.
@@ -5207,3 +5208,33 @@ func migrationAddWarpConversationsUpdatedAtIndex(ctx context.Context, db *gorm.D
 	}
 	return nil
 }
+
+// migrationAddEmbeddingInputColumn adds the embedding_input column to the logs table.
+func migrationAddEmbeddingInputColumn(ctx context.Context, db *gorm.DB, logger schemas.Logger) error {
+	migrationName := "logs_add_embedding_input_column"
+	logger.Info("[logstore] starting migration %s", migrationName)
+	defer logger.Info("[logstore] finished migration %s", migrationName)
+	opts := *migrator.DefaultOptions
+	opts.UseTransaction = true
+	m := migrator.New(db, &opts, []*migrator.Migration{{
+		ID: migrationName,
+		Migrate: func(tx *gorm.DB) error {
+			tx = tx.WithContext(ctx)
+			if err := boundDDLLockWait(tx); err != nil {
+				return err
+			}
+			return addColumnIfNotExists(tx, logger, &Log{}, "embedding_input")
+		},
+		Rollback: func(tx *gorm.DB) error {
+			tx = tx.WithContext(ctx)
+			if err := boundDDLLockWait(tx); err != nil {
+				return err
+			}
+			return dropColumnIfExists(tx, logger, &Log{}, "embedding_input")
+		},
+	}})
+	if err := m.Migrate(); err != nil {
+		return fmt.Errorf("error while adding embedding_input column: %s", err.Error())
+	}
+	return nil
+}

framework/logstore/migrations_test.go

@@ -751,3 +751,47 @@ func TestWarpRollbackLocksTablesInWriterOrder(t *testing.T) {
 	require.Less(t, conversationsLock, messagesLock,
 		"locks must follow writer order (conversation first), or a concurrent append can deadlock the rollback")
 }
+
+// ========== Embedding Input Column Migration Tests ==========
+
+// runEmbeddingInputColumnCases pins that the migration only adds the column and leaves existing rows NULL.
+func runEmbeddingInputColumnCases(t *testing.T, db *gorm.DB) {
+	t.Helper()
+	db.Exec("DROP TABLE IF EXISTS logs")
+	db.Exec("CREATE TABLE IF NOT EXISTS migrations (id VARCHAR(255) PRIMARY KEY)")
+	db.Exec("DELETE FROM migrations WHERE id = 'logs_add_embedding_input_column'")
+	require.NoError(t, db.Exec(`CREATE TABLE logs (id VARCHAR(255) PRIMARY KEY, object_type VARCHAR(255) NOT NULL, input_history TEXT)`).Error)
+	t.Cleanup(func() {
+		db.Exec("DROP TABLE IF EXISTS logs")
+		db.Exec("DELETE FROM migrations WHERE id = 'logs_add_embedding_input_column'")
+	})
+
+	history := `[{"role":"user","content":[{"type":"text","text":"hello"}]}]`
+	require.NoError(t, db.Exec("INSERT INTO logs (id, object_type, input_history) VALUES (?, ?, ?)", "emb-1", "embedding", history).Error)
+
+	ctx := context.Background()
+	require.NoError(t, migrationAddEmbeddingInputColumn(ctx, db, testLogger{}))
+	require.True(t, db.Migrator().HasColumn(&Log{}, "embedding_input"))
+
+	var result struct {
+		EmbeddingInput *string `gorm:"column:embedding_input"`
+	}
+	require.NoError(t, db.Table("logs").Select("embedding_input").Where("id = ?", "emb-1").Scan(&result).Error)
+	assert.Nil(t, result.EmbeddingInput, "historical rows are not backfilled; the UI falls back to input_history")
+
+	require.NoError(t, migrationAddEmbeddingInputColumn(ctx, db, testLogger{}), "re-run should be a no-op")
+}
+
+func TestMigrationAddEmbeddingInputColumn_Postgres(t *testing.T) {
+	db := trySetupPostgresDB(t)
+	if db == nil {
+		t.Skip("Postgres not available, skipping test")
+	}
+	runEmbeddingInputColumnCases(t, db)
+}
+
+func TestMigrationAddEmbeddingInputColumn_SQLite(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	runEmbeddingInputColumnCases(t, db)
+}

framework/logstore/payload.go

@@ -21,6 +21,7 @@ var payloadFields = []string{
 	"responses_input_history",
 	"output_message",
 	"responses_output",
+	"embedding_input",
 	"embedding_output",
 	"rerank_output",
 	"ocr_input",
@@ -64,6 +65,7 @@ func ExtractPayload(l *Log) map[string]string {
 	m["responses_input_history"] = l.ResponsesInputHistory
 	m["output_message"] = l.OutputMessage
 	m["responses_output"] = l.ResponsesOutput
+	m["embedding_input"] = l.EmbeddingInput
 	m["embedding_output"] = l.EmbeddingOutput
 	m["rerank_output"] = l.RerankOutput
 	m["ocr_input"] = l.OCRInput
@@ -208,6 +210,7 @@ func ClearPayload(l *Log) {
 	l.ResponsesInputHistory = ""
 	l.OutputMessage = ""
 	l.ResponsesOutput = ""
+	l.EmbeddingInput = ""
 	l.EmbeddingOutput = ""
 	l.RerankOutput = ""
 	l.OCRInput = ""
@@ -247,6 +250,7 @@ func ClearPayload(l *Log) {
 	l.ResponsesInputHistoryParsed = nil
 	l.OutputMessageParsed = nil
 	l.ResponsesOutputParsed = nil
+	l.EmbeddingInputParsed = nil
 	l.EmbeddingOutputParsed = nil
 	l.RerankOutputParsed = nil
 	l.OCRInputParsed = nil
@@ -297,6 +301,9 @@ func MergePayloadFromJSON(l *Log, data []byte) error {
 	if v, ok := m["responses_output"]; ok && v != "" {
 		l.ResponsesOutput = v
 	}
+	if v, ok := m["embedding_input"]; ok && v != "" {
+		l.EmbeddingInput = v
+	}
 	if v, ok := m["embedding_output"]; ok && v != "" {
 		l.EmbeddingOutput = v
 	}
@@ -866,6 +873,9 @@ func clearPayloadField(l *Log, name string) {
 	case "responses_output":
 		l.ResponsesOutput = ""
 		l.ResponsesOutputParsed = nil
+	case "embedding_input":
+		l.EmbeddingInput = ""
+		l.EmbeddingInputParsed = nil
 	case "embedding_output":
 		l.EmbeddingOutput = ""
 		l.EmbeddingOutputParsed = nil

framework/logstore/payload_test.go

@@ -18,6 +18,7 @@ func TestExtractPayload_RoundTrip(t *testing.T) {
 		ResponsesInputHistory:   `[{"role":"user","content":"hi"}]`,
 		OutputMessage:           `{"role":"assistant","content":"world"}`,
 		ResponsesOutput:         `[{"role":"assistant","content":"there"}]`,
+		EmbeddingInput:          `[{"content":[{"type":"text","text":"embed me"}]}]`,
 		EmbeddingOutput:         `[{"embedding":[0.1]}]`,
 		RerankOutput:            `[{"score":0.9}]`,
 		Params:                  `{"temperature":0.7}`,
@@ -56,6 +57,7 @@ func TestExtractPayload_RoundTrip(t *testing.T) {
 	assert.Equal(t, len(payloadFields)+1+6, len(payload), "payload map should have all payload fields plus metadata and index fields")
 	assert.Equal(t, `[{"role":"user","content":"hello"}]`, payload["input_history"])
 	assert.Equal(t, `{"role":"assistant","content":"world"}`, payload["output_message"])
+	assert.Equal(t, `[{"content":[{"type":"text","text":"embed me"}]}]`, payload["embedding_input"])
 	assert.Equal(t, `{"judge_calls":[{"total_tokens":18}]}`, payload["guardrail_debug"])
 	assert.Equal(t, `routing log`, payload["routing_engine_logs"])
 	assert.Equal(t, metadata, payload["metadata"], "metadata must be written to the snapshot for object consumers")
@@ -64,6 +66,8 @@ func TestExtractPayload_RoundTrip(t *testing.T) {
 	ClearPayload(log)
 	assert.Empty(t, log.InputHistory)
 	assert.Empty(t, log.OutputMessage)
+	assert.Empty(t, log.EmbeddingInput)
+	assert.Nil(t, log.EmbeddingInputParsed)
 	assert.Empty(t, log.RawRequest)
 	assert.Empty(t, log.GuardrailDebug)
 	assert.Empty(t, log.RoutingEngineLogs)
@@ -85,6 +89,7 @@ func TestExtractPayload_RoundTrip(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, `[{"role":"user","content":"hello"}]`, log.InputHistory)
 	assert.Equal(t, `{"role":"assistant","content":"world"}`, log.OutputMessage)
+	assert.Equal(t, `[{"content":[{"type":"text","text":"embed me"}]}]`, log.EmbeddingInput)
 	assert.Equal(t, `{"judge_calls":[{"total_tokens":18}]}`, log.GuardrailDebug)
 	assert.Equal(t, `routing log`, log.RoutingEngineLogs)
 	require.NotNil(t, log.Metadata)

framework/logstore/tables.go

@@ -266,6 +266,7 @@ type Log struct {
 	ResponsesInputHistory   string    `gorm:"type:text" json:"-"`                                                      // JSON serialized []schemas.ResponsesMessage
 	OutputMessage           string    `gorm:"type:text" json:"-"`                                                      // JSON serialized *schemas.ChatMessage
 	ResponsesOutput         string    `gorm:"type:text" json:"-"`                                                      // JSON serialized *schemas.ResponsesMessage
+	EmbeddingInput          string    `gorm:"type:text" json:"-"`                                                      // JSON serialized []schemas.EmbeddingInputItem
 	EmbeddingOutput         string    `gorm:"type:text" json:"-"`                                                      // JSON serialized [][]float32
 	RerankOutput            string    `gorm:"type:text" json:"-"`                                                      // JSON serialized []schemas.RerankResult
 	OCROutput               string    `gorm:"type:text" json:"-"`                                                      // JSON serialized *schemas.BifrostOCRResponse
@@ -396,6 +397,7 @@ type Log struct {
 	ResponsesInputHistoryParsed []schemas.ResponsesMessage              `gorm:"-" json:"responses_input_history,omitempty"`
 	OutputMessageParsed         *schemas.ChatMessage                    `gorm:"-" json:"output_message,omitempty"`
 	ResponsesOutputParsed       []schemas.ResponsesMessage              `gorm:"-" json:"responses_output,omitempty"`
+	EmbeddingInputParsed        []schemas.EmbeddingInputItem            `gorm:"-" json:"embedding_input,omitempty"`
 	EmbeddingOutputParsed       []schemas.EmbeddingData                 `gorm:"-" json:"embedding_output,omitempty"`
 	RerankOutputParsed          []schemas.RerankResult                  `gorm:"-" json:"rerank_output,omitempty"`
 	OCROutputParsed             *schemas.BifrostOCRResponse             `gorm:"-" json:"ocr_output,omitempty"`
@@ -590,6 +592,14 @@ func (l *Log) SerializeFields() error {
 		}
 	}
 
+	if l.EmbeddingInputParsed != nil {
+		if data, err := sonic.Marshal(l.EmbeddingInputParsed); err != nil {
+			return err
+		} else {
+			l.EmbeddingInput = string(data)
+		}
+	}
+
 	if l.EmbeddingOutputParsed != nil {
 		if data, err := sonic.Marshal(l.EmbeddingOutputParsed); err != nil {
 			return err
@@ -989,6 +999,12 @@ func (l *Log) DeserializeFields() error {
 		}
 	}
 
+	if l.EmbeddingInput != "" {
+		if err := sonic.Unmarshal([]byte(l.EmbeddingInput), &l.EmbeddingInputParsed); err != nil {
+			l.EmbeddingInputParsed = nil
+		}
+	}
+
 	if l.EmbeddingOutput != "" {
 		if err := sonic.Unmarshal([]byte(l.EmbeddingOutput), &l.EmbeddingOutputParsed); err != nil {
 			// Log error but don't fail the operation - initialize as nil
@@ -1909,6 +1925,15 @@ type MCPToolLogStats struct {
 func (l *Log) BuildContentSummary() string {
 	var parts []string
 
+	// Add embedding input text parts
+	for _, item := range l.EmbeddingInputParsed {
+		for _, part := range item.Content {
+			if part.Type == schemas.EmbeddingContentPartTypeText && part.Text != nil && *part.Text != "" {
+				parts = append(parts, *part.Text)
+			}
+		}
+	}
+
 	// Add input messages
 	for _, msg := range l.InputHistoryParsed {
 		if msg.Content != nil {

plugins/logging/main.go

@@ -246,6 +246,7 @@ func dropCapturedInput(entry *logstore.Log) {
 	entry.SpeechInputParsed = nil
 	entry.TranscriptionInputParsed = nil
 	entry.OCRInputParsed = nil
+	entry.EmbeddingInputParsed = nil
 	entry.ImageGenerationInputParsed = nil
 	entry.ImageEditInputParsed = nil
 	entry.ImageVariationInputParsed = nil
@@ -1107,6 +1108,7 @@ type InitialLogData struct {
 	Object                 string
 	InputHistory           []schemas.ChatMessage
 	ResponsesInputHistory  []schemas.ResponsesMessage
+	EmbeddingInput         []schemas.EmbeddingInputItem
 	Params                 any
 	SpeechInput            *schemas.SpeechInput
 	TranscriptionInput     *schemas.TranscriptionInput
@@ -1776,6 +1778,12 @@ func (p *LoggerPlugin) PreLLMHook(ctx *schemas.BifrostContext, req *schemas.Bifr
 			}
 		case schemas.EmbeddingRequest:
 			initialData.Params = req.EmbeddingRequest.Params
+			items := extractEmbeddingInput(req)
+			reqThreshold, _ := ctx.Value(schemas.BifrostContextKeyLargePayloadRequestThreshold).(int64)
+			if reqThreshold > 0 && embeddingMediaDataSize(items) > reqThreshold {
+				items = redactEmbeddingMediaData(items)
+			}
+			initialData.EmbeddingInput = items
 		case schemas.RerankRequest:
 			initialData.Params = req.RerankRequest.Params
 		case schemas.DecisionRequest:

plugins/logging/operations_test.go

@@ -2075,6 +2075,43 @@ func TestPreLLMHookNotificationCarriesNoInputWhileKeyIsUnresolved(t *testing.T)
 	}
 }
 
+// TestPreLLMHookNotificationCarriesNoEmbeddingInputWhileKeyIsUnresolved pins that embedding input,
+// captured provisionally because the presented key may turn content on, stays out of the notification.
+func TestPreLLMHookNotificationCarriesNoEmbeddingInputWhileKeyIsUnresolved(t *testing.T) {
+	store := newTestStore(t)
+	plugin, err := Init(context.Background(), &Config{DisableContentLogging: new(true)}, testLogger{}, store, nil, nil, nil)
+	if err != nil {
+		t.Fatalf("Init() error = %v", err)
+	}
+	t.Cleanup(func() { _ = plugin.Cleanup() })
+	var notified []*logstore.Log
+	plugin.SetLogCallback(func(_ context.Context, logEntry *logstore.Log) { notified = append(notified, logEntry) })
+
+	ctx := schemas.NewBifrostContext(context.Background(), schemas.NoDeadline)
+	ctx.SetValue(schemas.BifrostContextKeyRequestID, "req-notify-unresolved-embedding")
+	ctx.SetValue(schemas.BifrostContextKeyVirtualKey, "sk-bf-presented")
+	text := "secret embedding text"
+	_, _, err = plugin.PreLLMHook(ctx, &schemas.BifrostRequest{
+		RequestType: schemas.EmbeddingRequest,
+		EmbeddingRequest: &schemas.BifrostEmbeddingRequest{
+			Provider: schemas.OpenAI,
+			Model:    "text-embedding-3-small",
+			Input: []schemas.EmbeddingInputItem{
+				{Content: schemas.EmbeddingContent{{Type: schemas.EmbeddingContentPartTypeText, Text: &text}}},
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("PreLLMHook() error = %v", err)
+	}
+	if len(notified) == 0 {
+		t.Fatal("expected the processing notification")
+	}
+	if input := notified[0].EmbeddingInputParsed; input != nil {
+		t.Fatalf("the notification went out before the key was stamped and must carry no embedding input, got %#v", input)
+	}
+}
+
 // TestKeyTurningContentOnOverClientFlagKeepsEarlyCapturedContent pins the other direction of the
 // hook ordering hole: with the client flag off, a key that turns content on is still unstamped when
 // PreLLMHook and PreMCPHook capture, so they must capture provisionally and let the final policy in

plugins/logging/utils.go

@@ -731,6 +731,21 @@ func redactEmbeddingMediaData(items []schemas.EmbeddingInputItem) []schemas.Embe
 	return stripped
 }
 
+// embeddingMediaDataSize totals the inline media bytes across every item.
+func embeddingMediaDataSize(items []schemas.EmbeddingInputItem) int64 {
+	var total int64
+	for _, item := range items {
+		for _, part := range item.Content {
+			for _, media := range []*schemas.EmbeddingMediaPart{part.Image, part.Audio, part.File, part.Video} {
+				if media != nil && media.Data != nil {
+					total += int64(len(*media.Data))
+				}
+			}
+		}
+	}
+	return total
+}
+
 func extractRealtimeInputHistory(input []schemas.ResponsesMessage) []schemas.ChatMessage {
 	messages := make([]schemas.ChatMessage, 0, len(input))
 	for _, item := range input {