[DO-NOT-MERGE-YET] feat(complexity): add session routing observability

core/schemas/bifrost.go

@@ -267,6 +267,10 @@ const (
 	BifrostContextKeyGovernanceComplexityTier            BifrostContextKey = "bifrost-governance-complexity-tier"      // string (complexity tier computed for routing, e.g. "SIMPLE"/"MEDIUM"/"COMPLEX"; only present when a routing rule referenced complexity_tier and classification produced a tier (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
 	BifrostContextKeyGovernanceComplexityMechanism       BifrostContextKey = "bifrost-governance-complexity-mechanism" // string (how the complexity tier was classified: "semantic", or "skipped" when classification was demanded but produced no tier; only present when a routing rule referenced complexity_tier (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
 	BifrostContextKeyGovernanceComplexityScore           BifrostContextKey = "bifrost-governance-complexity-score"     // float64 (classifier score behind the tier: the semantic classifier's similarity to the nearest reference phrase; only present alongside a computed tier (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+	BifrostContextKeyGovernanceComplexitySessionID          BifrostContextKey = "bifrost-governance-complexity-session-id"           // string (raw opaque session ID for exact log lookup)
+	BifrostContextKeyGovernanceComplexitySessionMode        BifrostContextKey = "bifrost-governance-complexity-session-mode"         // string ("pinned" or "cache_aware")
+	BifrostContextKeyGovernanceComplexitySessionTierSource  BifrostContextKey = "bifrost-governance-complexity-session-tier-source"  // string ("classified", "memoised", or "held"; absent when no tier was established)
+	BifrostContextKeyGovernanceComplexitySessionSwitchCount BifrostContextKey = "bifrost-governance-complexity-session-switch-count" // int (absent when a store failure makes the persisted count unknowable)
 	BifrostContextKeyRoutingPinnedAPIKeyID               BifrostContextKey = "bifrost-routing-pinned-api-key-id"       // string (provider key ID pinned by a matched routing rule target; resolved against the configured key pool during key selection and takes precedence over a caller-supplied pin (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
 	BifrostContextKeySelectedPromptName                  BifrostContextKey = "bifrost-selected-prompt-name"            // string (display name of the selected prompt (set by prompts plugin - DO NOT SET THIS MANUALLY))
 	BifrostContextKeySelectedPromptVersion               BifrostContextKey = "bifrost-selected-prompt-version"         // string (numeric version as string, e.g. "3" (set by prompts plugin - DO NOT SET THIS MANUALLY))

framework/logstore/billingprojection_test.go

@@ -68,6 +68,20 @@ func TestListProjectionIncludesBillingScalars(t *testing.T) {
 	}
 }
 
+func TestListProjectionIncludesComplexitySessionFields(t *testing.T) {
+	cols := newProjectionTestStore(t).listSelectColumns()
+	for _, field := range []string{
+		"complexity_session_id",
+		"complexity_session_mode",
+		"complexity_session_tier_source",
+		"complexity_session_switch_count",
+	} {
+		if !containsColumn(cols, field) {
+			t.Fatalf("listSelectColumns is missing %q; log exports would silently drop session observability", field)
+		}
+	}
+}
+
 func TestBillingProjectionSelectsEveryPricingInput(t *testing.T) {
 	cols := newProjectionTestStore(t).billingSelectColumns()
 	for _, scalar := range billingScalarColumns {

framework/logstore/logstoreparity_test.go

@@ -110,6 +110,7 @@ type parityLogSpec struct {
 	tier         *string
 	mechanism    *string
 	tierScore    *float64
+	session      ComplexitySessionLogFields
 	metadata     *string
 	cacheDebug   string
 	content      string
@@ -121,7 +122,7 @@ type parityLogSpec struct {
 
 func (s parityLogSpec) toLog(base time.Time) *Log {
 	ts := base.Add(-time.Duration(s.offsetSec) * time.Second)
-	return &Log{
+	logEntry := &Log{
 		ID:                    s.id,
 		Timestamp:             ts,
 		Object:                s.object,
@@ -156,6 +157,8 @@ func (s parityLogSpec) toLog(base time.Time) *Log {
 		RateLimitIDs:          s.rateLimitIDs,
 		CreatedAt:             ts,
 	}
+	logEntry.ComplexitySessionLogFields = s.session
+	return logEntry
 }
 
 func paritySpecs() []parityLogSpec {
@@ -166,12 +169,14 @@ func paritySpecs() []parityLogSpec {
 			cost: f64PtrP(0.5), latency: f64PtrP(100), tokens: [3]int{100, 50, 150}, stopReason: strPtrP("stop"),
 			routing: strPtrP("governance,loadbalancing"), metadata: strPtrP(`{"env":"prod"}`),
 			tier: strPtrP("COMPLEX"), mechanism: strPtrP("lexical"), tierScore: f64PtrP(0.55),
+			session:    ComplexitySessionLogFields{ComplexitySessionID: strPtrP("conversation-1"), ComplexitySessionMode: strPtrP("pinned"), ComplexitySessionTierSource: strPtrP("memoised"), ComplexitySessionSwitchCount: intPtrP(0)},
 			cacheDebug: `{"hit_type":"direct"}`, content: "alpha bravo hello", parentID: strPtrP("sess1")},
 		{id: "p2", offsetSec: 90, object: "chat.completion", provider: "openai", model: "gpt-4o", status: "success",
 			vkID: strPtrP("vk1"), vkName: strPtrP("VK One"), teamID: strPtrP("t1"), userID: strPtrP("u2"),
 			cost: f64PtrP(1.25), latency: f64PtrP(250), tokens: [3]int{200, 100, 300}, stopReason: strPtrP("length"),
 			routing: strPtrP("governance"), metadata: strPtrP(`{"env":"dev"}`),
 			tier: strPtrP("SIMPLE"), mechanism: strPtrP("lexical"), tierScore: f64PtrP(0.08),
+			session:    ComplexitySessionLogFields{ComplexitySessionID: strPtrP("conversation-2"), ComplexitySessionMode: strPtrP("cache_aware"), ComplexitySessionTierSource: strPtrP("held"), ComplexitySessionSwitchCount: intPtrP(2)},
 			cacheDebug: `{"hit_type":"semantic"}`, content: "charlie delta", parentID: strPtrP("sess1")},
 		{id: "p3", offsetSec: 80, object: "chat.completion", provider: "openai", model: "gpt-4o-mini", status: "error",
 			vkID: strPtrP("vk2"), vkName: strPtrP("VK Two"), teamID: strPtrP("t2"), userID: strPtrP("u2"),
@@ -415,8 +420,10 @@ func logProjection(l *Log) map[string]any {
 		"prompt_tokens": l.PromptTokens, "completion_tokens": l.CompletionTokens,
 		"total_tokens": l.TotalTokens, "stop_reason": l.StopReason,
 		"complexity_tier": l.ComplexityTier, "complexity_mechanism": l.ComplexityMechanism,
-		"complexity_score": l.ComplexityScore,
-		"content_summary":  l.ContentSummary,
+		"complexity_score":      l.ComplexityScore,
+		"complexity_session_id": l.ComplexitySessionID, "complexity_session_mode": l.ComplexitySessionMode,
+		"complexity_session_tier_source": l.ComplexitySessionTierSource, "complexity_session_switch_count": l.ComplexitySessionSwitchCount,
+		"content_summary": l.ContentSummary,
 	}
 }
 
@@ -536,33 +543,36 @@ func TestLogStoreParity(t *testing.T) {
 	// --- Phase: reads ---
 
 	searchCases := map[string]SearchFilters{
-		"all":                   window,
-		"providers":             {Providers: []string{"openai"}},
-		"models":                {Models: []string{"gpt-4o", "claude-3"}},
-		"status":                {Status: []string{"success"}},
-		"stop_reasons":          {StopReasons: []string{"stop"}},
-		"complexity_tiers":      {ComplexityTiers: []string{"COMPLEX", "MEDIUM"}},
-		"complexity_mechanisms": {ComplexityMechanisms: []string{"lexical"}},
-		"mechanism_skipped":     {ComplexityMechanisms: []string{"skipped"}},
-		"objects":               {Objects: []string{"embedding"}},
-		"aliases":               {Aliases: []string{"a1"}},
-		"selected_keys":         {SelectedKeyIDs: []string{"sk1"}},
-		"virtual_keys":          {VirtualKeyIDs: []string{"vk1"}},
-		"teams":                 {TeamIDs: []string{"t1", "t3"}},
-		"customers":             {CustomerIDs: []string{"c1"}},
-		"users":                 {UserIDs: []string{"u1"}},
-		"business_units":        {BusinessUnitIDs: []string{"b1"}},
-		"routing_engines":       {RoutingEngineUsed: []string{"loadbalancing", "routing-rule"}},
-		"time_range":            {StartTime: timePtrP(base.Add(-75 * time.Second)), EndTime: timePtrP(base.Add(-25 * time.Second))},
-		"latency_range":         {MinLatency: f64PtrP(80), MaxLatency: f64PtrP(260)},
-		"token_range":           {MinTokens: intPtrP(100), MaxTokens: intPtrP(600)},
-		"cost_range":            {MinCost: f64PtrP(1.0), MaxCost: f64PtrP(2.6)},
-		"missing_cost":          {MissingCostOnly: true},
-		"cache_direct":          {CacheHitTypes: []string{"direct"}},
-		"cache_semantic":        {CacheHitTypes: []string{"semantic"}},
-		"metadata":              {MetadataFilters: map[string]string{"env": "prod"}},
-		"content_search":        {ContentSearch: "charlie"},
-		"parent_request":        {ParentRequestID: "sess1"},
+		"all":                             window,
+		"providers":                       {Providers: []string{"openai"}},
+		"models":                          {Models: []string{"gpt-4o", "claude-3"}},
+		"status":                          {Status: []string{"success"}},
+		"stop_reasons":                    {StopReasons: []string{"stop"}},
+		"complexity_tiers":                {ComplexityTiers: []string{"COMPLEX", "MEDIUM"}},
+		"complexity_mechanisms":           {ComplexityMechanisms: []string{"lexical"}},
+		"complexity_session_id":           {ComplexitySessionID: "conversation-1"},
+		"mechanism_skipped":               {ComplexityMechanisms: []string{"skipped"}},
+		"complexity_session_modes":        {ComplexitySessionModes: []string{"cache_aware"}},
+		"complexity_session_tier_sources": {ComplexitySessionTierSources: []string{"memoised"}},
+		"objects":                         {Objects: []string{"embedding"}},
+		"aliases":                         {Aliases: []string{"a1"}},
+		"selected_keys":                   {SelectedKeyIDs: []string{"sk1"}},
+		"virtual_keys":                    {VirtualKeyIDs: []string{"vk1"}},
+		"teams":                           {TeamIDs: []string{"t1", "t3"}},
+		"customers":                       {CustomerIDs: []string{"c1"}},
+		"users":                           {UserIDs: []string{"u1"}},
+		"business_units":                  {BusinessUnitIDs: []string{"b1"}},
+		"routing_engines":                 {RoutingEngineUsed: []string{"loadbalancing", "routing-rule"}},
+		"time_range":                      {StartTime: timePtrP(base.Add(-75 * time.Second)), EndTime: timePtrP(base.Add(-25 * time.Second))},
+		"latency_range":                   {MinLatency: f64PtrP(80), MaxLatency: f64PtrP(260)},
+		"token_range":                     {MinTokens: intPtrP(100), MaxTokens: intPtrP(600)},
+		"cost_range":                      {MinCost: f64PtrP(1.0), MaxCost: f64PtrP(2.6)},
+		"missing_cost":                    {MissingCostOnly: true},
+		"cache_direct":                    {CacheHitTypes: []string{"direct"}},
+		"cache_semantic":                  {CacheHitTypes: []string{"semantic"}},
+		"metadata":                        {MetadataFilters: map[string]string{"env": "prod"}},
+		"content_search":                  {ContentSearch: "charlie"},
+		"parent_request":                  {ParentRequestID: "sess1"},
 	}
 	for name, filters := range searchCases {
 		t.Run("SearchLogs/"+name, func(t *testing.T) {

framework/logstore/matviews.go

@@ -948,6 +948,8 @@ func canUseMatViewFilters(f SearchFilters) bool {
 		len(f.StopReasons) == 0 &&
 		len(f.ComplexityTiers) == 0 &&
 		len(f.ComplexityMechanisms) == 0 &&
+		len(f.ComplexitySessionModes) == 0 &&
+		len(f.ComplexitySessionTierSources) == 0 &&
 		f.MinLatency == nil && f.MaxLatency == nil &&
 		f.MinTokens == nil && f.MaxTokens == nil &&
 		f.MinCost == nil && f.MaxCost == nil &&

framework/logstore/migrations.go

@@ -296,6 +296,7 @@ var logstoreMigrationSteps = []migrationStep{
 	{IDs: []string{"mcp_tool_logs_add_endpoint_columns"}, run: migrationAddEndpointColumnsToMCPToolLogs},
 	{IDs: []string{"mcp_tool_logs_add_plugin_logs_column"}, run: migrationAddMCPPluginLogsColumn},
 	{IDs: []string{"logs_add_complexity_routing_columns"}, run: migrationAddComplexityRoutingColumns},
+	{IDs: []string{"logs_add_complexity_session_columns"}, run: migrationAddComplexitySessionColumns},
 }
 
 // areThereAnyPendingMigrations returns true if there are any pending migrations to be applied.
@@ -2717,6 +2718,16 @@ var performanceIndexes = []performanceIndexDef{
 		name:  "idx_logs_complexity_mechanism",
 		sql:   "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_logs_complexity_mechanism ON logs(complexity_mechanism) WHERE complexity_mechanism IS NOT NULL",
 	},
+	{
+		table: "logs",
+		name:  "idx_logs_complexity_session_mode",
+		sql:   "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_logs_complexity_session_mode ON logs(complexity_session_mode) WHERE complexity_session_mode IS NOT NULL",
+	},
+	{
+		table: "logs",
+		name:  "idx_logs_complexity_session_tier_source",
+		sql:   "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_logs_complexity_session_tier_source ON logs(complexity_session_tier_source) WHERE complexity_session_tier_source IS NOT NULL",
+	},
 	{
 		table: "mcp_tool_logs",
 		name:  "idx_mcp_logs_user_id",
@@ -3669,6 +3680,65 @@ func migrationAddComplexityRoutingColumns(ctx context.Context, db *gorm.DB, logg
 	return nil
 }
 
+// migrationAddComplexitySessionColumns adds nullable request-log fields for
+// session-aware complexity decisions. SQLite creates the session ID index in
+// this migration; PostgreSQL builds its partial index concurrently through the
+// existing background index reconciler.
+func migrationAddComplexitySessionColumns(ctx context.Context, db *gorm.DB, logger schemas.Logger) error {
+	migrationName := "logs_add_complexity_session_columns"
+	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)
+			for _, field := range []string{
+				"complexity_session_id",
+				"complexity_session_mode",
+				"complexity_session_tier_source",
+				"complexity_session_switch_count",
+			} {
+				if err := addColumnIfNotExists(tx, logger, &Log{}, field); err != nil {
+					return err
+				}
+			}
+			mg := tx.Migrator()
+			if tx.Dialector.Name() != "postgres" && !mg.HasIndex(&Log{}, "idx_logs_complexity_session_id") {
+				if err := mg.CreateIndex(&Log{}, "idx_logs_complexity_session_id"); err != nil {
+					return fmt.Errorf("create complexity_session_id index: %w", err)
+				}
+			}
+			return nil
+		},
+		Rollback: func(tx *gorm.DB) error {
+			tx = tx.WithContext(ctx)
+			mg := tx.Migrator()
+			if mg.HasIndex(&Log{}, "idx_logs_complexity_session_id") {
+				if err := mg.DropIndex(&Log{}, "idx_logs_complexity_session_id"); err != nil {
+					return err
+				}
+			}
+			for _, field := range []string{
+				"complexity_session_switch_count",
+				"complexity_session_tier_source",
+				"complexity_session_mode",
+				"complexity_session_id",
+			} {
+				if err := dropColumnIfExists(tx, logger, &Log{}, field); err != nil {
+					return err
+				}
+			}
+			return nil
+		},
+	}})
+	if err := m.Migrate(); err != nil {
+		return fmt.Errorf("error while adding complexity session columns: %s", err.Error())
+	}
+	return nil
+}
+
 // migrationAddRedactionMappingColumn adds the redaction_mapping column to the
 // logs table. It stores the reversible redaction mapping (encrypted when an
 // encryption key is configured) so that reveal data shares the log row's

framework/logstore/migrations_test.go

@@ -49,6 +49,32 @@ func TestMigrationAddMCPPluginLogsColumn(t *testing.T) {
 	assert.Equal(t, int64(1), count)
 }
 
+// TestMigrationAddComplexitySessionColumns verifies that session observability
+// is additive, idempotent, and leaves existing request logs untouched.
+func TestMigrationAddComplexitySessionColumns(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "migrations.db")), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
+	require.NoError(t, err)
+	require.NoError(t, db.Exec("CREATE TABLE logs (id TEXT PRIMARY KEY)").Error)
+	require.NoError(t, db.Exec("INSERT INTO logs (id) VALUES (?)", "existing-log").Error)
+
+	ctx := context.Background()
+	require.NoError(t, migrationAddComplexitySessionColumns(ctx, db, testLogger{}))
+	for _, field := range []string{
+		"ComplexitySessionID",
+		"ComplexitySessionMode",
+		"ComplexitySessionTierSource",
+		"ComplexitySessionSwitchCount",
+	} {
+		assert.True(t, db.Migrator().HasColumn(&Log{}, field), "missing migrated field %s", field)
+	}
+	require.True(t, db.Migrator().HasIndex(&Log{}, "idx_logs_complexity_session_id"))
+	require.NoError(t, migrationAddComplexitySessionColumns(ctx, db, testLogger{}))
+
+	var count int64
+	require.NoError(t, db.Table("logs").Where("id = ?", "existing-log").Count(&count).Error)
+	assert.Equal(t, int64(1), count)
+}
+
 // pgTestSchema is this package's dedicated Postgres schema. Test packages
 // (configstore, configstore/tables, logstore) run in parallel against the same
 // database, so each one works in its own schema to avoid clobbering the

framework/logstore/multi_team_filter_test.go

@@ -89,4 +89,6 @@ func TestCanUseMatViewFilters_ExcludesTeamBU(t *testing.T) {
 	assert.False(t, canUseMatViewFilters(SearchFilters{BusinessUnitIDs: []string{"bu1"}}), "BU filter must force the raw path")
 	assert.False(t, canUseMatViewFilters(SearchFilters{CustomerIDs: []string{"c1"}}), "customer filter must force the raw path")
 	assert.False(t, canUseMatViewFilters(SearchFilters{ParentRequestID: "req-1"}), "parent request filter has no matview dimension and must force the raw path")
+	assert.False(t, canUseMatViewFilters(SearchFilters{ComplexitySessionModes: []string{"pinned"}}), "session mode is not materialized and must force the raw path")
+	assert.False(t, canUseMatViewFilters(SearchFilters{ComplexitySessionTierSources: []string{"held"}}), "session tier source is not materialized and must force the raw path")
 }

framework/logstore/rdb.go

@@ -238,6 +238,12 @@ func (s *RDBLogStore) applyFilters(baseQuery *gorm.DB, filters SearchFilters) *g
 	if len(filters.ComplexityMechanisms) > 0 {
 		baseQuery = baseQuery.Where("complexity_mechanism IN ?", filters.ComplexityMechanisms)
 	}
+	if len(filters.ComplexitySessionModes) > 0 {
+		baseQuery = baseQuery.Where("complexity_session_mode IN ?", filters.ComplexitySessionModes)
+	}
+	if len(filters.ComplexitySessionTierSources) > 0 {
+		baseQuery = baseQuery.Where("complexity_session_tier_source IN ?", filters.ComplexitySessionTierSources)
+	}
 	if len(filters.Objects) > 0 {
 		baseQuery = baseQuery.Where("object_type IN ?", filters.Objects)
 	}
@@ -1171,6 +1177,7 @@ func (s *RDBLogStore) listSelectColumns() string {
 		"virtual_key_id", "virtual_key_name",
 		"routing_engines_used", "routing_rule_id", "routing_rule_name",
 		"complexity_tier", "complexity_mechanism",
+		"complexity_session_id", "complexity_session_mode", "complexity_session_tier_source", "complexity_session_switch_count",
 		"user_id", "user_name", "team_id", "team_name", "customer_id", "customer_name",
 		"business_unit_id", "business_unit_name",
 		"team_ids", "team_names", "customer_ids", "customer_names", "business_unit_ids", "business_unit_names",