diff --git a/core/schemas/bifrost.go b/core/schemas/bifrost.go
index 24f81f72fef..f3151e06534 100644
--- a/core/schemas/bifrost.go
+++ b/core/schemas/bifrost.go
@@ -203,29 +203,33 @@ const (
MCPContextKeyIncludeClients BifrostContextKey = "mcp-include-clients" // Context key for whitelist client filtering
MCPContextKeyIncludeTools BifrostContextKey = "mcp-include-tools" // Context key for whitelist tool filtering (Note: toolName should be in "clientName-toolName" format for individual tools, or "clientName-*" for wildcard)
- BifrostContextKeySelectedKeyID BifrostContextKey = "bifrost-selected-key-id" // string (to store the selected key ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeySelectedKeyName BifrostContextKey = "bifrost-selected-key-name" // string (to store the selected key name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceVirtualKeyID BifrostContextKey = "bifrost-governance-virtual-key-id" // string (to store the virtual key ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceVirtualKeyName BifrostContextKey = "bifrost-governance-virtual-key-name" // string (to store the virtual key name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceTeamID BifrostContextKey = "bifrost-governance-team-id" // string (to store the team ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceTeamName BifrostContextKey = "bifrost-governance-team-name" // string (to store the team name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceCustomerID BifrostContextKey = "bifrost-governance-customer-id" // string (to store the customer ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceCustomerName BifrostContextKey = "bifrost-governance-customer-name" // string (to store the customer name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceBusinessUnitID BifrostContextKey = "bifrost-governance-business-unit-id" // string (to store the business unit ID (set by enterprise governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceBusinessUnitName BifrostContextKey = "bifrost-governance-business-unit-name" // string (to store the business unit name (set by enterprise governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceRoutingRuleID BifrostContextKey = "bifrost-governance-routing-rule-id" // string (to store the routing rule ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceRoutingRuleName BifrostContextKey = "bifrost-governance-routing-rule-name" // string (to store the routing rule name (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))
- BifrostContextKeySelectedPromptID BifrostContextKey = "bifrost-selected-prompt-id" // string (id of the selected prompt (set by prompts plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyGovernanceIncludeOnlyKeys BifrostContextKey = "bf-governance-include-only-keys" // []string (to store the include-only key IDs for provider config routing (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
- BifrostContextKeyNumberOfRetries BifrostContextKey = "bifrost-number-of-retries" // int (to store the number of retries (set by bifrost - DO NOT SET THIS MANUALLY))
- BifrostContextKeyFallbackIndex BifrostContextKey = "bifrost-fallback-index" // int (to store the fallback index (set by bifrost - DO NOT SET THIS MANUALLY)) 0 for primary, 1 for first fallback, etc.
- BifrostContextKeyStreamEndIndicator BifrostContextKey = "bifrost-stream-end-indicator" // bool (set by bifrost - DO NOT SET THIS MANUALLY))
- BifrostContextKeyStreamIdleTimeout BifrostContextKey = "bifrost-stream-idle-timeout" // time.Duration (per-chunk idle timeout for streaming)
- BifrostContextKeySkipKeySelection BifrostContextKey = "bifrost-skip-key-selection" // bool (will pass an empty key to the provider)
- BifrostContextKeyExtraHeaders BifrostContextKey = "bifrost-extra-headers" // map[string][]string
- BifrostContextKeyURLPath BifrostContextKey = "bifrost-extra-url-path" // string
+ BifrostContextKeySelectedKeyID BifrostContextKey = "bifrost-selected-key-id" // string (to store the selected key ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeySelectedKeyName BifrostContextKey = "bifrost-selected-key-name" // string (to store the selected key name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceVirtualKeyID BifrostContextKey = "bifrost-governance-virtual-key-id" // string (to store the virtual key ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceVirtualKeyName BifrostContextKey = "bifrost-governance-virtual-key-name" // string (to store the virtual key name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceTeamID BifrostContextKey = "bifrost-governance-team-id" // string (to store the team ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceTeamName BifrostContextKey = "bifrost-governance-team-name" // string (to store the team name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceCustomerID BifrostContextKey = "bifrost-governance-customer-id" // string (to store the customer ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceCustomerName BifrostContextKey = "bifrost-governance-customer-name" // string (to store the customer name (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceBusinessUnitID BifrostContextKey = "bifrost-governance-business-unit-id" // string (to store the business unit ID (set by enterprise governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceBusinessUnitName BifrostContextKey = "bifrost-governance-business-unit-name" // string (to store the business unit name (set by enterprise governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceTeamIDs BifrostContextKey = "bifrost-governance-team-ids" // []string (all teams a user/AP request belongs to; set by enterprise governance plugin - DO NOT SET THIS MANUALLY)
+ BifrostContextKeyGovernanceTeamNames BifrostContextKey = "bifrost-governance-team-names" // []string (display names, aligned with team-ids; set by enterprise governance plugin - DO NOT SET THIS MANUALLY)
+ BifrostContextKeyGovernanceBusinessUnitIDs BifrostContextKey = "bifrost-governance-business-unit-ids" // []string (distinct BUs across the user's teams; set by enterprise governance plugin - DO NOT SET THIS MANUALLY)
+ BifrostContextKeyGovernanceBusinessUnitNames BifrostContextKey = "bifrost-governance-business-unit-names" // []string (display names, aligned with business-unit-ids; set by enterprise governance plugin - DO NOT SET THIS MANUALLY)
+ BifrostContextKeyGovernanceRoutingRuleID BifrostContextKey = "bifrost-governance-routing-rule-id" // string (to store the routing rule ID (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceRoutingRuleName BifrostContextKey = "bifrost-governance-routing-rule-name" // string (to store the routing rule name (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))
+ BifrostContextKeySelectedPromptID BifrostContextKey = "bifrost-selected-prompt-id" // string (id of the selected prompt (set by prompts plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyGovernanceIncludeOnlyKeys BifrostContextKey = "bf-governance-include-only-keys" // []string (to store the include-only key IDs for provider config routing (set by bifrost governance plugin - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyNumberOfRetries BifrostContextKey = "bifrost-number-of-retries" // int (to store the number of retries (set by bifrost - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyFallbackIndex BifrostContextKey = "bifrost-fallback-index" // int (to store the fallback index (set by bifrost - DO NOT SET THIS MANUALLY)) 0 for primary, 1 for first fallback, etc.
+ BifrostContextKeyStreamEndIndicator BifrostContextKey = "bifrost-stream-end-indicator" // bool (set by bifrost - DO NOT SET THIS MANUALLY))
+ BifrostContextKeyStreamIdleTimeout BifrostContextKey = "bifrost-stream-idle-timeout" // time.Duration (per-chunk idle timeout for streaming)
+ BifrostContextKeySkipKeySelection BifrostContextKey = "bifrost-skip-key-selection" // bool (will pass an empty key to the provider)
+ BifrostContextKeyExtraHeaders BifrostContextKey = "bifrost-extra-headers" // map[string][]string
+ BifrostContextKeyURLPath BifrostContextKey = "bifrost-extra-url-path" // string
BifrostContextKeyUseRawRequestBody BifrostContextKey = "bifrost-use-raw-request-body"
BifrostContextKeyChangeRequestType BifrostContextKey = "bifrost-change-request-type" // RequestType (set by plugins to trigger request type conversion in core, e.g. text->chat or chat->responses)
BifrostContextKeySendBackRawRequest BifrostContextKey = "bifrost-send-back-raw-request" // bool (per-request override — read by bifrost.go, never overwritten)
@@ -1433,7 +1437,7 @@ type BifrostResponseExtraFields struct {
ConvertedRequestType RequestType `json:"converted_request_type,omitempty"`
DroppedCompatPluginParams []string `json:"dropped_compat_plugin_params,omitempty"` // params dropped by the compat plugin based on model catalog
ProviderResponseHeaders map[string]string `json:"provider_response_headers,omitempty"` // HTTP response headers from the provider (filtered to exclude transport-level headers)
- PassthroughPath string `json:"passthrough_path,omitempty"` // Stripped provider path for passthrough requests, e.g. "/v1/chat/completions"
+ PassthroughPath string `json:"passthrough_path,omitempty"` // Stripped provider path for passthrough requests, e.g. "/v1/chat/completions"
}
type BifrostMCPResponseExtraFields struct {
@@ -1670,15 +1674,15 @@ func (e *ErrorField) UnmarshalJSON(data []byte) error {
// BifrostErrorExtraFields contains additional fields in an error response.
type BifrostErrorExtraFields struct {
- Provider ModelProvider `json:"provider,omitempty"`
- OriginalModelRequested string `json:"original_model_requested,omitempty"`
- ResolvedModelUsed string `json:"resolved_model_used,omitempty"`
- RequestType RequestType `json:"request_type,omitempty"`
- MCPRequestType MCPRequestType `json:"mcp_request_type,omitempty"`
- RawRequest interface{} `json:"raw_request,omitempty"`
- RawResponse interface{} `json:"raw_response,omitempty"`
- ConvertedRequestType RequestType `json:"converted_request_type,omitempty"`
- DroppedCompatPluginParams []string `json:"dropped_compat_plugin_params,omitempty"`
- KeyStatuses []KeyStatus `json:"key_statuses,omitempty"`
- MCPAuthRequired *MCPAuthRequiredError `json:"mcp_auth_required,omitempty"` // Set when a per-user MCP tool requires the caller to complete an inline auth flow (OAuth or headers)
+ Provider ModelProvider `json:"provider,omitempty"`
+ OriginalModelRequested string `json:"original_model_requested,omitempty"`
+ ResolvedModelUsed string `json:"resolved_model_used,omitempty"`
+ RequestType RequestType `json:"request_type,omitempty"`
+ MCPRequestType MCPRequestType `json:"mcp_request_type,omitempty"`
+ RawRequest interface{} `json:"raw_request,omitempty"`
+ RawResponse interface{} `json:"raw_response,omitempty"`
+ ConvertedRequestType RequestType `json:"converted_request_type,omitempty"`
+ DroppedCompatPluginParams []string `json:"dropped_compat_plugin_params,omitempty"`
+ KeyStatuses []KeyStatus `json:"key_statuses,omitempty"`
+ MCPAuthRequired *MCPAuthRequiredError `json:"mcp_auth_required,omitempty"` // Set when a per-user MCP tool requires the caller to complete an inline auth flow (OAuth or headers)
}
diff --git a/framework/logstore/matviews.go b/framework/logstore/matviews.go
index d341c209ed6..44b11aa84fb 100644
--- a/framework/logstore/matviews.go
+++ b/framework/logstore/matviews.go
@@ -116,12 +116,38 @@ const filterDataMatViewWindow = "30 days"
// - requiredColumns: resolved column aliases used by repairMatViewShapes
// to detect drifted matviews. Declared explicitly (not parsed from
// selectExpr) so SQL fragments like COALESCE(...) cannot poison the check.
+// - bodyOverride: when set, replaces the whole `SELECT DISTINCT ... FROM
+// logs WHERE ...` body (selectExpr/whereExpr are ignored). Used by the
+// multi-valued team / business-unit views, which must union the scalar
+// column with the JSON-array column rather than read a single column.
type filterMatViewDef struct {
name string
selectExpr string
whereExpr string
uniqueIdx string
requiredColumns []string
+ bodyOverride string
+}
+
+// multiValueFilterMatViewBody builds the SELECT body for a filter matview whose
+// dimension is single-valued on the scalar column (old / pre-migration rows and
+// the VK-team path) and multi-valued on the JSON-array column (the enterprise
+// user/AP path). It reuses teamOrBUFanoutFrom — the same fan-out the ranking /
+// histogram readers use — so a team / business unit that only ever appears in
+// the JSON array still surfaces in the filter dropdown. The fanned-out
+// dim_id/dim_name become the dropdown id/name; the visibility columns
+// (user_id, team_id, virtual_key_id) come from the original log row (exposed via
+// l.* by the fan-out subquery) so DAC scope still applies. idCol is the scalar
+// id column ("team_id" / "business_unit_id").
+func multiValueFilterMatViewBody(idCol string) string {
+ from, _ := teamOrBUFanoutFrom(idCol)
+ return fmt.Sprintf(
+ "SELECT DISTINCT dim_id AS id, dim_name AS name, "+
+ "COALESCE(user_id, '') AS user_id, COALESCE(team_id, '') AS team_id, "+
+ "COALESCE(virtual_key_id, '') AS virtual_key_id "+
+ "FROM %s WHERE timestamp >= NOW() - INTERVAL '%s' AND dim_id != '' AND dim_name != ''",
+ from, filterDataMatViewWindow,
+ )
}
// scopeProjection is the per-row visibility columns appended to every
@@ -202,12 +228,12 @@ var filterMatViews = []filterMatViewDef{
},
{
name: "mv_filter_teams",
- // team_id is exposed as "id" for the dropdown and also as the scope
- // column for uniform DAC predicates.
- selectExpr: "team_id AS id, team_name AS name, " +
- "COALESCE(user_id, '') AS user_id, COALESCE(team_id, '') AS team_id, " +
- "COALESCE(virtual_key_id, '') AS virtual_key_id",
- whereExpr: "team_id IS NOT NULL AND team_id != '' AND team_name IS NOT NULL AND team_name != ''",
+ // A request can belong to one team (scalar team_id) or many (JSON-array
+ // team_ids, enterprise user/AP path). bodyOverride unions both so every
+ // team shows in the dropdown, not just the scalar primary. team_id is
+ // exposed as the dropdown "id" and the original row's team_id is kept as
+ // the scope column for uniform DAC predicates.
+ bodyOverride: multiValueFilterMatViewBody("team_id"),
uniqueIdx: "id, name, " + scopeIdxColumns,
requiredColumns: append([]string{"id", "name"}, scopeRequiredColumns...),
},
@@ -219,16 +245,17 @@ var filterMatViews = []filterMatViewDef{
requiredColumns: append([]string{"id", "name"}, scopeRequiredColumns...),
},
{
- name: "mv_filter_users",
- selectExpr: "user_id AS id, user_name AS name, " + scopeProjection,
- whereExpr: "user_id IS NOT NULL AND user_id != '' AND user_name IS NOT NULL AND user_name != ''",
+ name: "mv_filter_users",
+ selectExpr: "user_id AS id, user_name AS name, " + scopeProjection,
+ whereExpr: "user_id IS NOT NULL AND user_id != '' AND user_name IS NOT NULL AND user_name != ''",
uniqueIdx: "id, name, " + scopeIdxColumns,
requiredColumns: append([]string{"id", "name"}, scopeRequiredColumns...),
},
{
- name: "mv_filter_business_units",
- selectExpr: "business_unit_id AS id, business_unit_name AS name, " + scopeProjection,
- whereExpr: "business_unit_id IS NOT NULL AND business_unit_id != '' AND business_unit_name IS NOT NULL AND business_unit_name != ''",
+ name: "mv_filter_business_units",
+ // Same scalar-or-JSON-array union as mv_filter_teams: a request can carry
+ // one business unit (scalar) or many (JSON-array business_unit_ids).
+ bodyOverride: multiValueFilterMatViewBody("business_unit_id"),
uniqueIdx: "id, name, " + scopeIdxColumns,
requiredColumns: append([]string{"id", "name"}, scopeRequiredColumns...),
},
@@ -242,11 +269,14 @@ var filterMatViewKeyPairColumns = map[[2]string]string{
{"routing_rule_id", "routing_rule_name"}: "mv_filter_routing_rules",
{"team_id", "team_name"}: "mv_filter_teams",
{"customer_id", "customer_name"}: "mv_filter_customers",
- {"user_id", "user_name"}: "mv_filter_users",
+ {"user_id", "user_name"}: "mv_filter_users",
{"business_unit_id", "business_unit_name"}: "mv_filter_business_units",
}
func filterMatViewDDL(v filterMatViewDef) string {
+ if v.bodyOverride != "" {
+ return fmt.Sprintf("CREATE MATERIALIZED VIEW IF NOT EXISTS %s AS %s", v.name, v.bodyOverride)
+ }
return fmt.Sprintf(
"CREATE MATERIALIZED VIEW IF NOT EXISTS %s AS SELECT DISTINCT %s FROM logs WHERE timestamp >= NOW() - INTERVAL '%s' AND (%s)",
v.name, v.selectExpr, filterDataMatViewWindow, v.whereExpr,
@@ -736,7 +766,9 @@ func canUseMatViewFilters(f SearchFilters) bool {
f.MinTokens == nil && f.MaxTokens == nil &&
f.MinCost == nil && f.MaxCost == nil &&
!f.MissingCostOnly &&
- len(f.CacheHitTypes) == 0
+ len(f.CacheHitTypes) == 0 &&
+ len(f.TeamIDs) == 0 &&
+ len(f.BusinessUnitIDs) == 0
}
// canUseMatView checks both that materialized views are ready (created and
@@ -1738,10 +1770,10 @@ func (s *RDBLogStore) getDimensionRankingsFromMatView(ctx context.Context, filte
}
type row struct {
- ID string `gorm:"column:id"`
- Total int64 `gorm:"column:total"`
- TotalTkns int64 `gorm:"column:total_tkns"`
- TotalCost float64 `gorm:"column:total_cost"`
+ ID string `gorm:"column:id"`
+ Total int64 `gorm:"column:total"`
+ TotalTkns int64 `gorm:"column:total_tkns"`
+ TotalCost float64 `gorm:"column:total_cost"`
}
var results []row
diff --git a/framework/logstore/migrations.go b/framework/logstore/migrations.go
index a6fffe5e682..905fedcaaeb 100644
--- a/framework/logstore/migrations.go
+++ b/framework/logstore/migrations.go
@@ -336,6 +336,15 @@ func triggerMigrations(ctx context.Context, db *gorm.DB) error {
if err := migrationRecreateFilterUsersMatView(ctx, db); err != nil {
return err
}
+ if err := migrationAddMultiTeamBusinessUnitColumns(ctx, db); err != nil {
+ return err
+ }
+ if err := migrationAddMultiTeamBusinessUnitGINIndexes(ctx, db); err != nil {
+ return err
+ }
+ if err := migrationRecreateFilterTeamBUMatViews(ctx, db); err != nil {
+ return err
+ }
// migrationSplitFilterDataMatView is intentionally NOT invoked in this
// release. Dropping mv_logs_filterdata while old replicas are still
// serving /api/logs/filterdata from it would surface "relation does not
@@ -2005,6 +2014,38 @@ func migrationAddMetadataGINIndex(ctx context.Context, db *gorm.DB) error {
return nil
}
+// migrationAddMultiTeamBusinessUnitGINIndexes registers the GIN indexes backing
+// multi-team / multi-BU log filtering. Like migrationAddMetadataGINIndex, the
+// build itself is deferred to ensureMultiTeamBusinessUnitGINIndexes (post-startup,
+// background, CONCURRENTLY); this migration exists only to provide a rollback that
+// drops the indexes. Postgres-only.
+func migrationAddMultiTeamBusinessUnitGINIndexes(ctx context.Context, db *gorm.DB) error {
+ opts := *migrator.DefaultOptions
+ opts.UseTransaction = false
+ m := migrator.New(db, &opts, []*migrator.Migration{{
+ ID: "logs_add_multi_team_bu_gin_indexes_v1",
+ Migrate: func(tx *gorm.DB) error {
+ return nil
+ },
+ Rollback: func(tx *gorm.DB) error {
+ tx = tx.WithContext(ctx)
+ if tx.Dialector.Name() == "postgres" {
+ if err := tx.Exec("DROP INDEX IF EXISTS idx_logs_team_ids_gin").Error; err != nil {
+ return fmt.Errorf("failed to drop team_ids GIN index: %w", err)
+ }
+ if err := tx.Exec("DROP INDEX IF EXISTS idx_logs_business_unit_ids_gin").Error; err != nil {
+ return fmt.Errorf("failed to drop business_unit_ids GIN index: %w", err)
+ }
+ }
+ return nil
+ },
+ }})
+ if err := m.Migrate(); err != nil {
+ return fmt.Errorf("error while registering multi team/BU GIN indexes: %s", err.Error())
+ }
+ return nil
+}
+
// ensureMetadataGINIndex checks whether idx_logs_metadata_gin exists and is valid.
// If the index is missing or was left in an INVALID state by a previously interrupted
// CREATE INDEX CONCURRENTLY, it drops the remnant and rebuilds the index synchronously.
@@ -2088,6 +2129,59 @@ func cleanupInvalidLogMetadata(ctx context.Context, conn *sql.Conn) error {
`)
}
+// ensureArrayGINIndex builds a partial jsonb_path_ops GIN index on a JSON-array
+// text column (e.g. team_ids) so `column::jsonb @> '[...]'` containment filters
+// are indexed. Mirrors ensureMetadataGINIndex's lifecycle: tolerates an INVALID
+// remnant from an interrupted build and builds CONCURRENTLY so writers are not
+// blocked. No data cleanup is needed — the log writer only ever stores a valid
+// JSON array or NULL in these columns. indexName/column are internal constants
+// (not user input), so identifier interpolation is safe. Postgres-only.
+func ensureArrayGINIndex(ctx context.Context, conn *sql.Conn, indexName, column string) error {
+ var indexValid bool
+ if err := conn.QueryRowContext(ctx, `
+ SELECT COALESCE(bool_and(pi.indisvalid), false)
+ FROM pg_class pc
+ JOIN pg_index pi ON pi.indrelid = pc.oid
+ JOIN pg_class ic ON ic.oid = pi.indexrelid
+ WHERE pc.relname = 'logs' AND ic.relname = $1
+ `, indexName).Scan(&indexValid); err != nil {
+ return fmt.Errorf("failed to query GIN index validity for %s: %w", indexName, err)
+ }
+ if indexValid {
+ return nil
+ }
+
+ // Drop any INVALID remnant left by a prior interrupted CONCURRENTLY build.
+ if _, err := conn.ExecContext(ctx, "DROP INDEX CONCURRENTLY IF EXISTS "+indexName); err != nil {
+ return fmt.Errorf("failed to drop invalid GIN index %s: %w", indexName, err)
+ }
+
+ // Non-fatal tuning to speed up the build.
+ _, _ = conn.ExecContext(ctx, "SET maintenance_work_mem = '512MB'")
+ _, _ = conn.ExecContext(ctx, "SET max_parallel_maintenance_workers = 4")
+
+ // jsonb_path_ops supports @> (containment) and is ~3x smaller than the default
+ // opclass. The partial predicate matches the IS JSON ARRAY guard the filter
+ // query adds (rdb.go), so the planner uses this index.
+ stmt := fmt.Sprintf(
+ "CREATE INDEX CONCURRENTLY IF NOT EXISTS %s ON logs USING gin ((%s::jsonb) jsonb_path_ops) WHERE %s IS NOT NULL AND %s IS JSON ARRAY",
+ indexName, column, column, column,
+ )
+ if _, err := conn.ExecContext(ctx, stmt); err != nil {
+ return fmt.Errorf("failed to create GIN index %s: %w", indexName, err)
+ }
+ return nil
+}
+
+// ensureMultiTeamBusinessUnitGINIndexes builds the GIN indexes backing multi-team
+// and multi-BU log filtering (team_ids / business_unit_ids).
+func ensureMultiTeamBusinessUnitGINIndexes(ctx context.Context, conn *sql.Conn) error {
+ if err := ensureArrayGINIndex(ctx, conn, "idx_logs_team_ids_gin", "team_ids"); err != nil {
+ return err
+ }
+ return ensureArrayGINIndex(ctx, conn, "idx_logs_business_unit_ids_gin", "business_unit_ids")
+}
+
// migrationAddDashboardEnhancements adds cached_read_tokens column to logs table.
// The expensive backfill, covering index rebuild, and MCP index creation are deferred
// to ensureDashboardEnhancements (called post-startup in a background goroutine) so
@@ -2837,6 +2931,50 @@ func migrationAddGovernanceContextColumns(ctx context.Context, db *gorm.DB) erro
return nil
}
+// migrationAddMultiTeamBusinessUnitColumns adds the JSON-array columns capturing
+// the full deduped set of teams / business units a request belongs to (enterprise
+// user/AP path). The scalar team_id/business_unit_id remain the primary; these
+// power display, multi-team filtering (jsonb @> + GIN), and fan-out aggregation.
+func migrationAddMultiTeamBusinessUnitColumns(ctx context.Context, db *gorm.DB) error {
+ opts := *migrator.DefaultOptions
+ opts.UseTransaction = true
+
+ columns := []string{"team_ids", "team_names", "business_unit_ids", "business_unit_names"}
+
+ m := migrator.New(db, &opts, []*migrator.Migration{{
+ ID: "logs_add_multi_team_business_unit_columns",
+ Migrate: func(tx *gorm.DB) error {
+ tx = tx.WithContext(ctx)
+ mig := tx.Migrator()
+ for _, col := range columns {
+ if !mig.HasColumn(&Log{}, col) {
+ if err := mig.AddColumn(&Log{}, col); err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+ },
+ Rollback: func(tx *gorm.DB) error {
+ tx = tx.WithContext(ctx)
+ mig := tx.Migrator()
+ for _, col := range columns {
+ if mig.HasColumn(&Log{}, col) {
+ if err := mig.DropColumn(&Log{}, col); err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+ },
+ }})
+ err := m.Migrate()
+ if err != nil {
+ return fmt.Errorf("error while adding multi team/business-unit columns: %s", err.Error())
+ }
+ return nil
+}
+
// migrationRecreateMatViewsWithGovernanceColumns drops and recreates materialized views
// so they include the new governance context columns (user_id, team_id, customer_id, business_unit_id).
// The actual rebuild is deferred to ensureMatViews, which runs after startup on
@@ -3333,3 +3471,38 @@ func migrationRecreateFilterUsersMatView(ctx context.Context, db *gorm.DB) error
}
return nil
}
+
+// migrationRecreateFilterTeamBUMatViews drops mv_filter_teams and
+// mv_filter_business_units so ensureMatViews recreates them with the multi-value
+// body (scalar column UNION the JSON-array column). Required because
+// repairMatViewShapes only detects drift by column presence, and the column
+// shape (id, name, user_id, team_id, virtual_key_id) is unchanged — only the
+// SELECT body changed — so the views would otherwise keep their old scalar-only
+// definition. Recreated views keep identical columns, so old replicas reading
+// them during a rolling deploy are unaffected (no legacyMatViewNames dance).
+func migrationRecreateFilterTeamBUMatViews(ctx context.Context, db *gorm.DB) error {
+ if db.Dialector.Name() != "postgres" {
+ return nil
+ }
+ opts := *migrator.DefaultOptions
+ opts.UseTransaction = true
+ m := migrator.New(db, &opts, []*migrator.Migration{{
+ ID: "logs_recreate_filter_team_bu_matviews_multivalue",
+ Migrate: func(tx *gorm.DB) error {
+ tx = tx.WithContext(ctx)
+ for _, view := range []string{"mv_filter_teams", "mv_filter_business_units"} {
+ if err := tx.Exec("DROP MATERIALIZED VIEW IF EXISTS " + view + " CASCADE").Error; err != nil {
+ return fmt.Errorf("failed to drop %s: %w", view, err)
+ }
+ }
+ return nil
+ },
+ Rollback: func(tx *gorm.DB) error {
+ return nil
+ },
+ }})
+ if err := m.Migrate(); err != nil {
+ return fmt.Errorf("error while recreating filter team/business-unit matviews: %s", err.Error())
+ }
+ return nil
+}
diff --git a/framework/logstore/multi_team_filter_test.go b/framework/logstore/multi_team_filter_test.go
new file mode 100644
index 00000000000..8a43582cd15
--- /dev/null
+++ b/framework/logstore/multi_team_filter_test.go
@@ -0,0 +1,82 @@
+package logstore
+
+import (
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// TestMultiValueDimensionFilterSQL verifies the predicate that matches logs by a
+// dimension that is scalar on the primary column and multi-valued on the JSON
+// array column: it must OR the scalar IN with guarded array containment per id,
+// and bind args in order (the id slice first, then one JSON fragment per id).
+func TestMultiValueDimensionFilterSQL(t *testing.T) {
+ sql, args := multiValueDimensionFilterSQL("team_id", "team_ids", []string{"t1", "t2"})
+
+ // Outer parens so it ANDs as one group with other filters.
+ assert.True(t, strings.HasPrefix(sql, "(") && strings.HasSuffix(sql, ")"), "must be parenthesised: %s", sql)
+ // Scalar primary path (covers VK-team / pre-migration / NULL-array rows).
+ assert.Contains(t, sql, "team_id IN ?")
+ // Partial-index guard so the planner uses the jsonb_path_ops GIN.
+ assert.Contains(t, sql, "team_ids IS NOT NULL AND team_ids IS JSON ARRAY")
+ // One containment test per requested id.
+ assert.Equal(t, 2, strings.Count(sql, "team_ids::jsonb @> ?::jsonb"))
+
+ // Args: the id slice first (for `IN ?`), then a JSON array fragment per id.
+ require.Len(t, args, 3)
+ assert.Equal(t, []string{"t1", "t2"}, args[0])
+ assert.Equal(t, `["t1"]`, args[1])
+ assert.Equal(t, `["t2"]`, args[2])
+
+ // Works for the BU columns too.
+ buSQL, buArgs := multiValueDimensionFilterSQL("business_unit_id", "business_unit_ids", []string{"bu1"})
+ assert.Contains(t, buSQL, "business_unit_id IN ?")
+ assert.Contains(t, buSQL, "business_unit_ids::jsonb @> ?::jsonb")
+ require.Len(t, buArgs, 2)
+ assert.Equal(t, `["bu1"]`, buArgs[1])
+}
+
+// TestTeamOrBUFanoutFrom verifies the fan-out FROM subquery: it must unnest the
+// array columns (id+name aligned by ordinality) for array rows and fall back to
+// the scalar id/name for non-array rows, with mutually exclusive branches, and
+// be aliased AS logs so DAC scope + filters still resolve.
+func TestTeamOrBUFanoutFrom(t *testing.T) {
+ teamSQL, ok := teamOrBUFanoutFrom("team_id")
+ require.True(t, ok)
+ assert.Contains(t, teamSQL, "jsonb_array_elements_text(l.team_ids::jsonb) WITH ORDINALITY")
+ assert.Contains(t, teamSQL, "jsonb_array_elements_text(l.team_names::jsonb) WITH ORDINALITY")
+ assert.Contains(t, teamSQL, "ON n.ord = t.ord", "names aligned with ids by ordinality")
+ // array branch guard + mutually-exclusive scalar fallback branch
+ assert.Contains(t, teamSQL, "WHERE l.team_ids IS NOT NULL AND l.team_ids IS JSON ARRAY")
+ assert.Contains(t, teamSQL, "SELECT l.team_id, COALESCE(l.team_name, '')")
+ assert.Contains(t, teamSQL, "WHERE l.team_ids IS NULL OR l.team_ids IS NOT JSON ARRAY")
+ assert.Contains(t, teamSQL, "UNION ALL")
+ assert.Contains(t, teamSQL, ") AS logs", "aliased AS logs so scope/filters resolve")
+ assert.Contains(t, teamSQL, "fan.dim_id AS dim_id")
+ assert.Contains(t, teamSQL, "fan.dim_name AS dim_name")
+
+ buSQL, ok := teamOrBUFanoutFrom("business_unit_id")
+ require.True(t, ok)
+ assert.Contains(t, buSQL, "business_unit_ids::jsonb")
+ assert.Contains(t, buSQL, "SELECT l.business_unit_id, COALESCE(l.business_unit_name, '')")
+
+ // Non-fan-out dimensions return false (caller uses the normal scalar path).
+ _, ok = teamOrBUFanoutFrom("user_id")
+ assert.False(t, ok)
+ _, ok = teamOrBUFanoutFrom("provider")
+ assert.False(t, ok)
+}
+
+// TestCanUseMatViewFilters_ExcludesTeamBU verifies that a team or business-unit
+// filter disqualifies the matview path: mv_logs_hourly only has the scalar
+// primary, so these must fall through to the raw (array-or-scalar) path to stay
+// complete. Other filters (e.g. provider) remain matview-eligible.
+func TestCanUseMatViewFilters_ExcludesTeamBU(t *testing.T) {
+ assert.True(t, canUseMatViewFilters(SearchFilters{}), "empty filters → matview eligible")
+ assert.True(t, canUseMatViewFilters(SearchFilters{Providers: []string{"openai"}}), "provider filter stays matview-eligible")
+
+ assert.False(t, canUseMatViewFilters(SearchFilters{TeamIDs: []string{"t1"}}), "team filter must force the raw path")
+ assert.False(t, canUseMatViewFilters(SearchFilters{BusinessUnitIDs: []string{"bu1"}}), "BU filter must force the raw path")
+}
diff --git a/framework/logstore/multi_team_matview_test.go b/framework/logstore/multi_team_matview_test.go
new file mode 100644
index 00000000000..73d633caa6d
--- /dev/null
+++ b/framework/logstore/multi_team_matview_test.go
@@ -0,0 +1,120 @@
+package logstore
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/maximhq/bifrost/framework/queryscope"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "gorm.io/gorm"
+)
+
+// insertTeamBULog inserts a single log row carrying the scalar and/or JSON-array
+// team / business-unit columns. Empty-string args are stored as NULL so the
+// matview's scalar-vs-array branch selection (team_ids IS JSON ARRAY) is
+// exercised faithfully — a row with no array column must fall back to the scalar.
+func insertTeamBULog(t *testing.T, db *gorm.DB, ts time.Time,
+ userID, teamID, teamName, teamIDs, teamNames, buID, buName, buIDs, buNames string) {
+ t.Helper()
+ nz := func(s string) any {
+ if s == "" {
+ return nil
+ }
+ return s
+ }
+ err := db.Exec(`
+ INSERT INTO logs (id, timestamp, object_type, provider, model, status,
+ user_id, team_id, team_name, team_ids, team_names,
+ business_unit_id, business_unit_name, business_unit_ids, business_unit_names,
+ created_at, latency, cost, prompt_tokens, completion_tokens, total_tokens)
+ VALUES (?, ?, 'chat_completion', 'openai', 'gpt-4', 'success',
+ ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 100, 0.01, 10, 5, 15)
+ `, uuid.New().String(), ts, nz(userID), nz(teamID), nz(teamName), nz(teamIDs), nz(teamNames),
+ nz(buID), nz(buName), nz(buIDs), nz(buNames), ts).Error
+ require.NoError(t, err, "failed to insert team/BU test log")
+}
+
+func keyPairsByID(pairs []KeyPairResult) map[string]string {
+ byID := make(map[string]string, len(pairs))
+ for _, p := range pairs {
+ byID[p.ID] = p.Name
+ }
+ return byID
+}
+
+// TestFilterTeamMatView_CollectsScalarAndArray is the regression for the reported
+// bug: the team filter dropdown read mv_filter_teams, which only projected the
+// scalar team_id/team_name. Teams that exist only in the JSON-array team_ids
+// (enterprise user/AP path) were missing. The recreated view must surface both.
+func TestFilterTeamMatView_CollectsScalarAndArray(t *testing.T) {
+ store, db := setupPerfTestDB(t)
+ store.matViewsReady.Store(true) // force the matview read path
+ ctx := context.Background()
+ now := time.Now().UTC()
+
+ // Old / VK-team log: scalar team only, no JSON array.
+ insertTeamBULog(t, db, now, "u-1", "t-scalar", "Scalar Team", "", "", "", "", "", "")
+ // Enterprise log: teams only in the JSON array, scalar team NULL.
+ insertTeamBULog(t, db, now, "u-1", "", "", `["t-arr1","t-arr2"]`, `["Array One","Array Two"]`, "", "", "", "")
+ refreshTestMatViews(t, db)
+
+ pairs, err := store.GetDistinctKeyPairs(ctx, "team_id", "team_name", 1000, "")
+ require.NoError(t, err)
+ byID := keyPairsByID(pairs)
+
+ assert.Equal(t, "Scalar Team", byID["t-scalar"], "scalar-only team must still appear (backward compatibility)")
+ assert.Equal(t, "Array One", byID["t-arr1"], "array-only team must now appear")
+ assert.Equal(t, "Array Two", byID["t-arr2"], "second array team must appear, name aligned by ordinality")
+}
+
+// TestFilterBusinessUnitMatView_CollectsScalarAndArray mirrors the team test for
+// the business-unit dropdown (mv_filter_business_units), which had the identical
+// scalar-only bug.
+func TestFilterBusinessUnitMatView_CollectsScalarAndArray(t *testing.T) {
+ store, db := setupPerfTestDB(t)
+ store.matViewsReady.Store(true)
+ ctx := context.Background()
+ now := time.Now().UTC()
+
+ insertTeamBULog(t, db, now, "u-1", "", "", "", "", "bu-scalar", "Scalar BU", "", "")
+ insertTeamBULog(t, db, now, "u-1", "", "", "", "", "", "", `["bu-arr1"]`, `["BU One"]`)
+ refreshTestMatViews(t, db)
+
+ pairs, err := store.GetDistinctKeyPairs(ctx, "business_unit_id", "business_unit_name", 1000, "")
+ require.NoError(t, err)
+ byID := keyPairsByID(pairs)
+
+ assert.Equal(t, "Scalar BU", byID["bu-scalar"], "scalar-only business unit must still appear")
+ assert.Equal(t, "BU One", byID["bu-arr1"], "array-only business unit must now appear")
+}
+
+// TestFilterTeamMatView_DACScopeAppliesAfterFanout proves the visibility columns
+// (user_id) survive the fan-out: each fanned-out team row carries its source
+// log's user_id, so a QueryScope still both resolves (no "column does not exist")
+// and filters — an array team owned by another user must not leak.
+func TestFilterTeamMatView_DACScopeAppliesAfterFanout(t *testing.T) {
+ store, db := setupPerfTestDB(t)
+ store.matViewsReady.Store(true)
+ now := time.Now().UTC()
+
+ // Array team owned by u-secret; scalar team visible to u-visible.
+ insertTeamBULog(t, db, now, "u-secret", "", "", `["t-secret"]`, `["Secret Team"]`, "", "", "", "")
+ insertTeamBULog(t, db, now, "u-visible", "t-public", "Public Team", "", "", "", "", "", "")
+ refreshTestMatViews(t, db)
+
+ scope := queryscope.QueryScope(func(db *gorm.DB) *gorm.DB {
+ return db.Where("user_id = ?", "u-visible")
+ })
+ ctx := queryscope.WithQueryScope(context.Background(), scope)
+
+ pairs, err := store.GetDistinctKeyPairs(ctx, "team_id", "team_name", 1000, "")
+ require.NoError(t, err, "scope WHERE must resolve against the fanned-out matview columns")
+ byID := keyPairsByID(pairs)
+
+ assert.Equal(t, "Public Team", byID["t-public"], "team visible to the scoped user must appear")
+ _, leaked := byID["t-secret"]
+ assert.False(t, leaked, "array team owned by another user must be filtered out by DAC scope")
+}
diff --git a/framework/logstore/postgres.go b/framework/logstore/postgres.go
index b908765e976..7f1dd332bdc 100644
--- a/framework/logstore/postgres.go
+++ b/framework/logstore/postgres.go
@@ -197,6 +197,12 @@ func newPostgresLogStore(ctx context.Context, config *PostgresConfig, logger sch
logger.Info("logstore: metadata GIN index is ready")
}
+ if err := ensureMultiTeamBusinessUnitGINIndexes(context.Background(), lock.conn); err != nil {
+ logger.Warn(fmt.Sprintf("logstore: team/business-unit GIN index build failed: %s (filtering will still work without the index)", err))
+ } else {
+ logger.Info("logstore: team/business-unit GIN indexes are ready")
+ }
+
if err := ensureDashboardEnhancements(context.Background(), lock.conn); err != nil {
logger.Warn(fmt.Sprintf("logstore: dashboard enhancements failed: %s (dashboard will still work with partial data)", err))
} else {
diff --git a/framework/logstore/rdb.go b/framework/logstore/rdb.go
index ce10cbcf4e7..d2c5bd3f8ad 100644
--- a/framework/logstore/rdb.go
+++ b/framework/logstore/rdb.go
@@ -85,6 +85,62 @@ func (s *RDBLogStore) ScopedDB(ctx context.Context) *gorm.DB {
return db
}
+// multiValueDimensionFilterSQL builds a Postgres predicate matching logs by a
+// dimension that is single-valued on the scalar column (the primary, set by the
+// VK path / pre-migration rows) and multi-valued on the JSON-array column (the
+// full set, set by the enterprise user/AP path). It ORs the scalar `IN` (btree
+// index) with array containment per id (partial jsonb_path_ops GIN index). The
+// `IS NOT NULL AND IS JSON ARRAY` guard matches the partial index predicate so
+// the planner uses the GIN. Returns the parenthesised SQL and its args.
+func multiValueDimensionFilterSQL(scalarCol, arrayCol string, ids []string) (string, []interface{}) {
+ arrConds := make([]string, len(ids))
+ args := []interface{}{ids}
+ for i, id := range ids {
+ arrConds[i] = arrayCol + "::jsonb @> ?::jsonb"
+ frag, _ := sonic.Marshal([]string{id})
+ args = append(args, string(frag))
+ }
+ sql := fmt.Sprintf("(%s IN ? OR (%s IS NOT NULL AND %s IS JSON ARRAY AND (%s)))",
+ scalarCol, arrayCol, arrayCol, strings.Join(arrConds, " OR "))
+ return sql, args
+}
+
+// teamOrBUFanoutFrom returns a Postgres FROM subquery (aliased AS logs) that fans
+// each log row out to one row per associated team / business unit, exposing
+// derived `dim_id` and `dim_name` columns alongside all original log columns
+// (l.*) so DAC scope and filters still resolve. Rows with the JSON-array column
+// set are unnested (id+name aligned by ordinality); rows without it (pre-upgrade
+// or VK-team logs) fall back to the scalar id/name — so historical logs keep
+// contributing. The two branches are mutually exclusive, so no row is counted
+// twice for the same dimension value. Returns ("", false) for non-fan-out
+// dimensions. idCol is the scalar id column ("team_id" / "business_unit_id"),
+// which both the ranking and histogram dimensions resolve to. No bind args: all
+// identifiers are internal constants.
+func teamOrBUFanoutFrom(idCol string) (string, bool) {
+ var arrIDs, arrNames, scalarName string
+ switch idCol {
+ case "team_id":
+ arrIDs, arrNames, scalarName = "team_ids", "team_names", "team_name"
+ case "business_unit_id":
+ arrIDs, arrNames, scalarName = "business_unit_ids", "business_unit_names", "business_unit_name"
+ default:
+ return "", false
+ }
+ return fmt.Sprintf(`(
+ SELECT l.*, fan.dim_id AS dim_id, fan.dim_name AS dim_name
+ FROM logs l
+ CROSS JOIN LATERAL (
+ SELECT t.value AS dim_id, COALESCE(n.value, '') AS dim_name
+ FROM jsonb_array_elements_text(l.%[1]s::jsonb) WITH ORDINALITY AS t(value, ord)
+ LEFT JOIN jsonb_array_elements_text(l.%[2]s::jsonb) WITH ORDINALITY AS n(value, ord) ON n.ord = t.ord
+ WHERE l.%[1]s IS NOT NULL AND l.%[1]s IS JSON ARRAY
+ UNION ALL
+ SELECT l.%[3]s, COALESCE(l.%[4]s, '')
+ WHERE l.%[1]s IS NULL OR l.%[1]s IS NOT JSON ARRAY
+ ) AS fan
+) AS logs`, arrIDs, arrNames, idCol, scalarName), true
+}
+
// applyFilters applies search filters to a GORM query. Callers are
// responsible for starting from ScopedDB(ctx) when row visibility
// should be respected; this helper only adds the per-call filter
@@ -121,7 +177,12 @@ func (s *RDBLogStore) applyFilters(baseQuery *gorm.DB, filters SearchFilters) *g
baseQuery = baseQuery.Where("routing_rule_id IN ?", filters.RoutingRuleIDs)
}
if len(filters.TeamIDs) > 0 {
- baseQuery = baseQuery.Where("team_id IN ?", filters.TeamIDs)
+ if s.db.Dialector.Name() == "postgres" {
+ sql, args := multiValueDimensionFilterSQL("team_id", "team_ids", filters.TeamIDs)
+ baseQuery = baseQuery.Where(sql, args...)
+ } else {
+ baseQuery = baseQuery.Where("team_id IN ?", filters.TeamIDs)
+ }
}
if len(filters.CustomerIDs) > 0 {
baseQuery = baseQuery.Where("customer_id IN ?", filters.CustomerIDs)
@@ -130,7 +191,12 @@ func (s *RDBLogStore) applyFilters(baseQuery *gorm.DB, filters SearchFilters) *g
baseQuery = baseQuery.Where("user_id IN ?", filters.UserIDs)
}
if len(filters.BusinessUnitIDs) > 0 {
- baseQuery = baseQuery.Where("business_unit_id IN ?", filters.BusinessUnitIDs)
+ if s.db.Dialector.Name() == "postgres" {
+ sql, args := multiValueDimensionFilterSQL("business_unit_id", "business_unit_ids", filters.BusinessUnitIDs)
+ baseQuery = baseQuery.Where(sql, args...)
+ } else {
+ baseQuery = baseQuery.Where("business_unit_id IN ?", filters.BusinessUnitIDs)
+ }
}
if len(filters.RoutingEngineUsed) > 0 {
// Query routing engines (comma-separated values) - find logs containing ANY of the specified engines
@@ -2003,7 +2069,25 @@ func (s *RDBLogStore) GetDimensionRankings(ctx context.Context, filters SearchFi
return nil, fmt.Errorf("invalid ranking dimension: %s", dimension)
}
- if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) {
+ // Multi-valued team / business-unit dimensions fan out over the JSON array
+ // (with scalar fallback for old / VK-team logs) so a request credits every
+ // team/BU it touches. Postgres-only; this forces the live path — the
+ // matview-accelerated equivalent is deferred to the partitioning work.
+ fanoutFrom := ""
+ if s.db.Dialector.Name() == "postgres" {
+ if f, isFanout := teamOrBUFanoutFrom(idCol); isFanout {
+ fanoutFrom = f
+ idCol, nameCol = "dim_id", "dim_name"
+ }
+ }
+ baseTable := func(q *gorm.DB) *gorm.DB {
+ if fanoutFrom != "" {
+ return q.Table(fanoutFrom)
+ }
+ return q.Model(&Log{})
+ }
+
+ if fanoutFrom == "" && s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) {
return s.getDimensionRankingsFromMatView(ctx, filters, dimension)
}
@@ -2022,7 +2106,7 @@ func (s *RDBLogStore) GetDimensionRankings(ctx context.Context, filters SearchFi
COALESCE(SUM(cost), 0) as total_cost
`, idCol, nameExpr)
- currentQuery := s.ScopedDB(ctx).Model(&Log{})
+ currentQuery := baseTable(s.ScopedDB(ctx))
currentQuery = s.applyFilters(currentQuery, filters)
currentQuery = currentQuery.Where("status IN ?", []string{"success", "error"})
currentQuery = currentQuery.Where(fmt.Sprintf("%s IS NOT NULL AND %s != ''", idCol, idCol))
@@ -2061,7 +2145,7 @@ func (s *RDBLogStore) GetDimensionRankings(ctx context.Context, filters SearchFi
prevFilters.StartTime = &prevStart
prevFilters.EndTime = &prevEnd
- prevQuery := s.ScopedDB(ctx).Model(&Log{})
+ prevQuery := baseTable(s.ScopedDB(ctx))
prevQuery = s.applyFilters(prevQuery, prevFilters)
prevQuery = prevQuery.Where("status IN ?", []string{"success", "error"})
prevQuery = prevQuery.Where(fmt.Sprintf("%s IS NOT NULL AND %s != ''", idCol, idCol))
@@ -2663,12 +2747,29 @@ func (s *RDBLogStore) GetDimensionCostHistogram(ctx context.Context, filters Sea
if bucketSizeSeconds <= 0 {
bucketSizeSeconds = 3600
}
- if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 {
- return s.getDimensionCostHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension)
- }
dimCol := string(dimension)
dialect := s.db.Dialector.Name()
- baseQuery := s.ScopedDB(ctx).Model(&Log{})
+ // Team / business-unit dimensions fan out over the JSON array (scalar
+ // fallback for old / VK-team logs). Postgres-only; forces the live path.
+ // NOTE: under fan-out the per-bucket *total* cost is the attributed total
+ // (≥ real, since a shared request counts toward each of its teams/BUs); the
+ // per-dimension breakdown is exact. Surface it as "attributed" in the UI.
+ fanoutFrom := ""
+ if dialect == "postgres" {
+ if f, isFanout := teamOrBUFanoutFrom(dimCol); isFanout {
+ fanoutFrom = f
+ dimCol = "dim_id"
+ }
+ }
+ if fanoutFrom == "" && dialect == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 {
+ return s.getDimensionCostHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension)
+ }
+ baseQuery := s.ScopedDB(ctx)
+ if fanoutFrom != "" {
+ baseQuery = baseQuery.Table(fanoutFrom)
+ } else {
+ baseQuery = baseQuery.Model(&Log{})
+ }
baseQuery = s.applyFilters(baseQuery, filters)
baseQuery = baseQuery.Where("status IN ?", []string{"success", "error"})
baseQuery = baseQuery.Where("cost IS NOT NULL AND cost > 0")
@@ -2758,12 +2859,26 @@ func (s *RDBLogStore) GetDimensionTokenHistogram(ctx context.Context, filters Se
if bucketSizeSeconds <= 0 {
bucketSizeSeconds = 3600
}
- if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 {
- return s.getDimensionTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension)
- }
dimCol := string(dimension)
dialect := s.db.Dialector.Name()
- baseQuery := s.ScopedDB(ctx).Model(&Log{})
+ // Team / business-unit dimensions fan out over the JSON array (scalar
+ // fallback for old / VK-team logs). Postgres-only; forces the live path.
+ fanoutFrom := ""
+ if dialect == "postgres" {
+ if f, isFanout := teamOrBUFanoutFrom(dimCol); isFanout {
+ fanoutFrom = f
+ dimCol = "dim_id"
+ }
+ }
+ if fanoutFrom == "" && dialect == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 {
+ return s.getDimensionTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension)
+ }
+ baseQuery := s.ScopedDB(ctx)
+ if fanoutFrom != "" {
+ baseQuery = baseQuery.Table(fanoutFrom)
+ } else {
+ baseQuery = baseQuery.Model(&Log{})
+ }
baseQuery = s.applyFilters(baseQuery, filters)
baseQuery = baseQuery.Where("status IN ?", []string{"success", "error"})
diff --git a/framework/logstore/rdb_postgres_perf_test.go b/framework/logstore/rdb_postgres_perf_test.go
index 68674dcd3b9..ed2a27190e3 100644
--- a/framework/logstore/rdb_postgres_perf_test.go
+++ b/framework/logstore/rdb_postgres_perf_test.go
@@ -183,8 +183,14 @@ func insertPerfMCPLog(t *testing.T, db *gorm.DB, opts mcpLogOpts) {
// refreshTestMatViews refreshes materialized views after inserting test data.
// This is needed because matviews are populated at creation time and don't
// automatically reflect new inserts until explicitly refreshed.
+//
+// The refresh gate is reset first: it's a package-level singleton, so without a
+// reset a prior test leaves it initialized and the eventually-consistent
+// pg_stat_user_tables counter (which lags fresh INSERTs by a few seconds) can
+// make refreshMatViews short-circuit, leaving the matview stale for this test.
func refreshTestMatViews(t *testing.T, db *gorm.DB) {
t.Helper()
+ resetTestMatViewRefreshGate()
ctx := context.Background()
err := refreshMatViews(ctx, db)
require.NoError(t, err, "Failed to refresh materialized views")
diff --git a/framework/logstore/tables.go b/framework/logstore/tables.go
index 0fa010be5a0..f5a1704f295 100644
--- a/framework/logstore/tables.go
+++ b/framework/logstore/tables.go
@@ -154,6 +154,10 @@ type Log struct {
CustomerName *string `gorm:"type:varchar(255)" json:"customer_name"`
BusinessUnitID *string `gorm:"type:varchar(255);index:idx_logs_business_unit_id" json:"business_unit_id"`
BusinessUnitName *string `gorm:"type:varchar(255)" json:"business_unit_name"`
+ TeamIDs *string `gorm:"type:text" json:"-"`
+ TeamNames *string `gorm:"type:text" json:"-"`
+ BusinessUnitIDs *string `gorm:"type:text" json:"-"`
+ BusinessUnitNames *string `gorm:"type:text" json:"-"`
InputHistory string `gorm:"type:text" json:"-"` // JSON serialized []schemas.ChatMessage
ResponsesInputHistory string `gorm:"type:text" json:"-"` // JSON serialized []schemas.ResponsesMessage
OutputMessage string `gorm:"type:text" json:"-"` // JSON serialized *schemas.ChatMessage
@@ -249,6 +253,10 @@ type Log struct {
AttemptTrailParsed []schemas.KeyAttemptRecord `gorm:"-" json:"attempt_trail,omitempty"`
BudgetIDsParsed []string `gorm:"-" json:"budget_ids,omitempty"`
RateLimitIDsParsed []string `gorm:"-" json:"rate_limit_ids,omitempty"`
+ TeamIDsParsed []string `gorm:"-" json:"team_ids,omitempty"`
+ TeamNamesParsed []string `gorm:"-" json:"team_names,omitempty"`
+ BusinessUnitIDsParsed []string `gorm:"-" json:"business_unit_ids,omitempty"`
+ BusinessUnitNamesParsed []string `gorm:"-" json:"business_unit_names,omitempty"`
// Populated in handlers after find using the virtual key id and key id
VirtualKey *tables.TableVirtualKey `gorm:"-" json:"virtual_key,omitempty"` // redacted
@@ -577,6 +585,39 @@ func (l *Log) SerializeFields() error {
}
}
+ if len(l.TeamIDsParsed) > 0 {
+ if data, err := sonic.Marshal(l.TeamIDsParsed); err != nil {
+ return err
+ } else {
+ s := string(data)
+ l.TeamIDs = &s
+ }
+ }
+ if len(l.TeamNamesParsed) > 0 {
+ if data, err := sonic.Marshal(l.TeamNamesParsed); err != nil {
+ return err
+ } else {
+ s := string(data)
+ l.TeamNames = &s
+ }
+ }
+ if len(l.BusinessUnitIDsParsed) > 0 {
+ if data, err := sonic.Marshal(l.BusinessUnitIDsParsed); err != nil {
+ return err
+ } else {
+ s := string(data)
+ l.BusinessUnitIDs = &s
+ }
+ }
+ if len(l.BusinessUnitNamesParsed) > 0 {
+ if data, err := sonic.Marshal(l.BusinessUnitNamesParsed); err != nil {
+ return err
+ } else {
+ s := string(data)
+ l.BusinessUnitNames = &s
+ }
+ }
+
// Build content summary for search.
// Skip if already set (e.g., by the hybrid log store which builds input-only summaries).
if l.ContentSummary == "" {
@@ -813,6 +854,27 @@ func (l *Log) DeserializeFields() error {
}
}
+ if l.TeamIDs != nil && *l.TeamIDs != "" {
+ if err := sonic.Unmarshal([]byte(*l.TeamIDs), &l.TeamIDsParsed); err != nil {
+ l.TeamIDsParsed = nil
+ }
+ }
+ if l.TeamNames != nil && *l.TeamNames != "" {
+ if err := sonic.Unmarshal([]byte(*l.TeamNames), &l.TeamNamesParsed); err != nil {
+ l.TeamNamesParsed = nil
+ }
+ }
+ if l.BusinessUnitIDs != nil && *l.BusinessUnitIDs != "" {
+ if err := sonic.Unmarshal([]byte(*l.BusinessUnitIDs), &l.BusinessUnitIDsParsed); err != nil {
+ l.BusinessUnitIDsParsed = nil
+ }
+ }
+ if l.BusinessUnitNames != nil && *l.BusinessUnitNames != "" {
+ if err := sonic.Unmarshal([]byte(*l.BusinessUnitNames), &l.BusinessUnitNamesParsed); err != nil {
+ l.BusinessUnitNamesParsed = nil
+ }
+ }
+
if l.RoutingEnginesUsedStr != nil && *l.RoutingEnginesUsedStr != "" {
// Parse comma-separated routing engines
l.RoutingEnginesUsed = strings.Split(*l.RoutingEnginesUsedStr, ",")
diff --git a/plugins/governance/store.go b/plugins/governance/store.go
index 6bc14ecf47e..78a84c4e999 100644
--- a/plugins/governance/store.go
+++ b/plugins/governance/store.go
@@ -160,6 +160,10 @@ type GovernanceStore interface {
// Team level CheckUserBudget
CheckTeamBudget(ctx context.Context, teamID string, request *EvaluationRequest, baselines map[string]float64) (Decision, error)
CheckTeamRateLimit(ctx context.Context, teamID string, request *EvaluationRequest, tokensBaselines map[string]int64, requestsBaselines map[string]int64) (Decision, error)
+ // Team-level live budget/rate-limit collectors (resolved from the hot maps);
+ // used by the enterprise user→team→business-unit hierarchy collector.
+ CollectTeamBudgets(ctx context.Context, teamID string) []*configstoreTables.TableBudget
+ CollectTeamRateLimits(ctx context.Context, teamID string) []*configstoreTables.TableRateLimit
// Customer-level governance checks
CheckCustomerBudget(ctx context.Context, customerID string, request *EvaluationRequest, baselines map[string]float64) (Decision, error)
CheckCustomerRateLimit(ctx context.Context, customerID string, request *EvaluationRequest, tokensBaselines map[string]int64, requestsBaselines map[string]int64) (Decision, error)
@@ -1267,6 +1271,59 @@ func (gs *LocalGovernanceStore) CheckTeamRateLimit(ctx context.Context, teamID s
return gs.CheckRateLimit(ctx, entityWiseRateLimits, tokensBaselines, requestsBaselines)
}
+// CollectTeamBudgets returns the live budget objects configured for a team,
+// resolved by ID from the hot budgets map (so usage counters and recent edits
+// are reflected). Mirrors the read pattern in CheckTeamBudget. Returns nil when
+// the team is unknown or has no budgets. Exported so the enterprise layer can
+// fold team budgets into a user→team→business-unit hierarchy collector the same
+// way collectBudgetsFromHierarchy folds them into the VK hierarchy.
+func (gs *LocalGovernanceStore) CollectTeamBudgets(ctx context.Context, teamID string) []*configstoreTables.TableBudget {
+ if teamID == "" {
+ return nil
+ }
+ teamValue, exists := gs.teams.Load(teamID)
+ if !exists || teamValue == nil {
+ return nil
+ }
+ team, ok := teamValue.(*configstoreTables.TableTeam)
+ if !ok || team == nil || len(team.Budgets) == 0 {
+ return nil
+ }
+ list := make([]*configstoreTables.TableBudget, 0, len(team.Budgets))
+ for _, b := range team.Budgets {
+ if hot := gs.LoadBudget(ctx, b.ID); hot != nil {
+ list = append(list, hot)
+ }
+ }
+ if len(list) == 0 {
+ return nil
+ }
+ return list
+}
+
+// CollectTeamRateLimits returns the live rate-limit object configured for a team
+// (at most one), resolved by ID from the hot rate-limits map. Mirrors the read
+// pattern in CheckTeamRateLimit. Returns nil when the team is unknown or has no
+// rate limit. Exported for the enterprise user-hierarchy collector.
+func (gs *LocalGovernanceStore) CollectTeamRateLimits(ctx context.Context, teamID string) []*configstoreTables.TableRateLimit {
+ if teamID == "" {
+ return nil
+ }
+ teamValue, exists := gs.teams.Load(teamID)
+ if !exists || teamValue == nil {
+ return nil
+ }
+ team, ok := teamValue.(*configstoreTables.TableTeam)
+ if !ok || team == nil || team.RateLimitID == nil {
+ return nil
+ }
+ rl := gs.LoadRateLimit(ctx, *team.RateLimitID)
+ if rl == nil {
+ return nil
+ }
+ return []*configstoreTables.TableRateLimit{rl}
+}
+
// CheckCustomerBudget checks customer-level budget and returns evaluation result if violated
func (gs *LocalGovernanceStore) CheckCustomerBudget(ctx context.Context, customerID string, request *EvaluationRequest, baselines map[string]float64) (Decision, error) {
if customerID == "" {
diff --git a/plugins/logging/main.go b/plugins/logging/main.go
index 6b4e93b4a63..3b18e45d7cc 100644
--- a/plugins/logging/main.go
+++ b/plugins/logging/main.go
@@ -919,6 +919,18 @@ func (p *LoggerPlugin) PostLLMHook(ctx *schemas.BifrostContext, result *schemas.
if rateLimitIDs, ok := ctx.Value(schemas.BifrostContextKeyGovernanceRateLimitIDs).([]string); ok && len(rateLimitIDs) > 0 {
entry.RateLimitIDsParsed = rateLimitIDs
}
+ if teamIDs, ok := ctx.Value(schemas.BifrostContextKeyGovernanceTeamIDs).([]string); ok && len(teamIDs) > 0 {
+ entry.TeamIDsParsed = teamIDs
+ }
+ if teamNames, ok := ctx.Value(schemas.BifrostContextKeyGovernanceTeamNames).([]string); ok && len(teamNames) > 0 {
+ entry.TeamNamesParsed = teamNames
+ }
+ if buIDs, ok := ctx.Value(schemas.BifrostContextKeyGovernanceBusinessUnitIDs).([]string); ok && len(buIDs) > 0 {
+ entry.BusinessUnitIDsParsed = buIDs
+ }
+ if buNames, ok := ctx.Value(schemas.BifrostContextKeyGovernanceBusinessUnitNames).([]string); ok && len(buNames) > 0 {
+ entry.BusinessUnitNamesParsed = buNames
+ }
entry.MetadataParsed = pending.InitialData.Metadata
entry.MetadataParsed = mergeRealtimeMetadata(entry.MetadataParsed, ctx)
entry.RoutingEngineLogs = routingEngineLogs
diff --git a/ui/app/workspace/dashboard/components/dimensionRankingsTab.tsx b/ui/app/workspace/dashboard/components/dimensionRankingsTab.tsx
index 6ced24e0764..9bf990a6278 100644
--- a/ui/app/workspace/dashboard/components/dimensionRankingsTab.tsx
+++ b/ui/app/workspace/dashboard/components/dimensionRankingsTab.tsx
@@ -19,6 +19,7 @@ interface DimensionRankingsTabProps {
loading: boolean;
dimensionLabel: string;
testIdPrefix: string;
+ attributed?: boolean;
}
function TopDimensionTooltip({ active, payload }: any) {
@@ -38,11 +39,13 @@ function TopDimensionChart({
loading,
dimensionLabel,
testIdPrefix,
+ attributed,
}: {
data: DimensionRankingsResponse | null;
loading: boolean;
dimensionLabel: string;
testIdPrefix: string;
+ attributed?: boolean;
}) {
const { chartData, grandTotal, rankedItems } = useMemo(() => {
if (!data?.rankings?.length) return { chartData: [], grandTotal: null, rankedItems: [] };
@@ -73,20 +76,27 @@ function TopDimensionChart({
loading={loading}
testId={`${testIdPrefix}-top-chart`}
className="z-[1] h-full"
- totalLabel="Total Requests"
+ totalLabel={attributed ? "Total Requests (attributed)" : "Total Requests"}
total={grandTotal !== null ? : undefined}
- totalTooltip={grandTotal !== null ? grandTotal.toLocaleString("en-US") : undefined}
+ totalTooltip={
+ grandTotal === null ? undefined : attributed ? (
+
+
{grandTotal.toLocaleString("en-US")} requests
+
+ Attributed — a request counts toward each {dimensionLabel.toLowerCase()} it belongs to, so this can exceed the actual request
+ count.
+