diff --git a/apps/manager-server/internal/repository/modelprice/repository.go b/apps/manager-server/internal/repository/modelprice/repository.go index bafbe7de4..cf6feb0fe 100644 --- a/apps/manager-server/internal/repository/modelprice/repository.go +++ b/apps/manager-server/internal/repository/modelprice/repository.go @@ -10,7 +10,7 @@ import ( "time" "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/model" - sqliterepo "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/sqlite" + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/usagepricing" ) var ErrStructureChangeAfterRawDeletion = errors.New("model price structure cannot change after archived raw usage has been deleted") @@ -251,13 +251,12 @@ func (r *repository) ReplaceAll(ctx context.Context, prices map[string]model.Mod beforeRevision := model.ModelPriceStructureRevision(beforePrices) afterRevision := model.ModelPriceStructureRevision(normalizedPrices) if beforeRevision != afterRevision { - hasDeletedRaw, err := sqliterepo.HistoricalRawDeletionExists(tx) - if err != nil { + if err := usagepricing.VerifyRetainedPricingRebuildSourceTx(ctx, tx); err != nil { + if errors.Is(err, usagepricing.ErrRetainedPricingHistoryIncomplete) { + return fmt.Errorf("%w: %v", ErrStructureChangeAfterRawDeletion, err) + } return err } - if hasDeletedRaw { - return ErrStructureChangeAfterRawDeletion - } } if _, err := tx.ExecContext(ctx, `delete from model_price_service_tiers`); err != nil { @@ -463,13 +462,12 @@ func (r *repository) UpsertSynced(ctx context.Context, prices map[string]model.M beforeRevision := model.ModelPriceStructureRevision(beforePrices) afterRevision := model.ModelPriceStructureRevision(afterPrices) if beforeRevision != afterRevision { - hasDeletedRaw, err := sqliterepo.HistoricalRawDeletionExists(tx) - if err != nil { + if err := usagepricing.VerifyRetainedPricingRebuildSourceTx(ctx, tx); err != nil { + if errors.Is(err, usagepricing.ErrRetainedPricingHistoryIncomplete) { + return model.ModelPriceSyncResult{}, fmt.Errorf("%w: %v", ErrStructureChangeAfterRawDeletion, err) + } return model.ModelPriceSyncResult{}, err } - if hasDeletedRaw { - return model.ModelPriceSyncResult{}, ErrStructureChangeAfterRawDeletion - } } sort.Strings(result.Preserved) if err := tx.Commit(); err != nil { diff --git a/apps/manager-server/internal/repository/modelprice/repository_test.go b/apps/manager-server/internal/repository/modelprice/repository_test.go index 9b331aee8..c3de6ff90 100644 --- a/apps/manager-server/internal/repository/modelprice/repository_test.go +++ b/apps/manager-server/internal/repository/modelprice/repository_test.go @@ -10,6 +10,7 @@ import ( "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/model" "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/modelprice" sqliterepo "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/sqlite" + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/usagemonitoring" ) func openTestDB(t *testing.T) (*sql.DB, modelprice.Repository) { @@ -43,6 +44,36 @@ func markRawDeleted(t *testing.T, db *sql.DB) { insertArchiveRef(t, db, 2000) } +func markRawDeletedWithRetainedProjection(t *testing.T, db *sql.DB) { + t.Helper() + ctx := context.Background() + const eventHash = "hash-retained" + if _, err := db.Exec(`insert into usage_events( + id, event_hash, timestamp_ms, timestamp, model, created_at_ms + ) values(1, ?, 1000, '1970-01-01T00:00:01Z', 'model-a', 1000)`, eventHash); err != nil { + t.Fatalf("insert retained raw event: %v", err) + } + projection := usagemonitoring.New(db) + result, err := projection.CatchUpProjection(ctx, 10, 1500) + if err != nil { + t.Fatalf("build retained projection: %v", err) + } + if result.CoverageEventID != 1 || result.Pending { + t.Fatalf("unexpected retained projection result: %#v", result) + } + if _, err := db.Exec(`pragma foreign_keys = off`); err != nil { + t.Fatalf("disable foreign keys: %v", err) + } + if _, err := db.Exec(`insert into usage_archive_event_refs( + event_hash, run_id, segment_sequence, raw_event_id, timestamp_ms, archived_at_ms, raw_deleted_at_ms + ) values(?, 'run-retained', 1, 1, 1000, 1600, 1700)`, eventHash); err != nil { + t.Fatalf("insert retained archive ref: %v", err) + } + if _, err := db.Exec(`delete from usage_events where id = 1`); err != nil { + t.Fatalf("delete retained raw event: %v", err) + } +} + // Test P1-1：没有 raw deletion，structure change 允许 func TestModelPriceStructureChangeAllowedWithoutRawDeletion(t *testing.T) { ctx := context.Background() @@ -514,3 +545,134 @@ func TestModelPriceUpsertSyncedManualPriceAndNewModelRejectedWithRawDeletion(t * t.Fatalf("model-b should not exist after rollback") } } + + +func TestModelPriceStructureChangeAllowedAfterRawDeletionWithRetainedProjection(t *testing.T) { + ctx := context.Background() + db, repo := openTestDB(t) + + if err := repo.ReplaceAll(ctx, map[string]model.ModelPrice{ + "model-a": {Prompt: 1.0, Completion: 2.0}, + }); err != nil { + t.Fatalf("initial ReplaceAll: %v", err) + } + markRawDeletedWithRetainedProjection(t, db) + + if err := repo.ReplaceAll(ctx, map[string]model.ModelPrice{ + "model-a": {Prompt: 1.0, Completion: 2.0}, + "model-b": {Prompt: 3.0, Completion: 4.0}, + }); err != nil { + t.Fatalf("structure change with retained projection: %v", err) + } + + persisted, err := repo.LoadAll(ctx) + if err != nil { + t.Fatalf("LoadAll: %v", err) + } + if _, ok := persisted["model-b"]; !ok { + t.Fatalf("expected model-b to be persisted") + } +} + +func TestModelPriceUpsertSyncedStructureChangeAllowedAfterRawDeletionWithRetainedProjection(t *testing.T) { + ctx := context.Background() + db, repo := openTestDB(t) + + if err := repo.ReplaceAll(ctx, map[string]model.ModelPrice{ + "model-a": {Prompt: 1.0, Completion: 2.0}, + }); err != nil { + t.Fatalf("initial ReplaceAll: %v", err) + } + markRawDeletedWithRetainedProjection(t, db) + + result, err := repo.UpsertSynced(ctx, map[string]model.ModelPrice{ + "model-b": {Prompt: 3.0, Completion: 4.0, Source: "sync"}, + }) + if err != nil { + t.Fatalf("UpsertSynced with retained projection: %v", err) + } + if result.Imported != 1 { + t.Fatalf("unexpected sync result: %+v", result) + } + persisted, err := repo.LoadAll(ctx) + if err != nil { + t.Fatalf("LoadAll: %v", err) + } + if _, ok := persisted["model-b"]; !ok { + t.Fatalf("expected model-b to be persisted") + } +} + + +func TestModelPriceStructureChangePreservesRetainedVerifierSystemError(t *testing.T) { + ctx := context.Background() + db, repo := openTestDB(t) + + if err := repo.ReplaceAll(ctx, map[string]model.ModelPrice{ + "model-a": {Prompt: 1.0, Completion: 2.0}, + }); err != nil { + t.Fatalf("initial ReplaceAll: %v", err) + } + markRawDeleted(t, db) + if _, err := db.Exec(`drop table usage_monitoring_rollup_state`); err != nil { + t.Fatalf("drop monitoring rollup state: %v", err) + } + + err := repo.ReplaceAll(ctx, map[string]model.ModelPrice{ + "model-a": {Prompt: 1.0, Completion: 2.0}, + "model-b": {Prompt: 3.0, Completion: 4.0}, + }) + if err == nil { + t.Fatal("expected retained verifier system error") + } + if errors.Is(err, modelprice.ErrStructureChangeAfterRawDeletion) { + t.Fatalf("system error was misclassified as structure conflict: %v", err) + } + + persisted, loadErr := repo.LoadAll(ctx) + if loadErr != nil { + t.Fatalf("LoadAll: %v", loadErr) + } + if len(persisted) != 1 { + t.Fatalf("expected mutation rollback, got %d models", len(persisted)) + } + if _, ok := persisted["model-b"]; ok { + t.Fatal("model-b should not exist after verifier system error") + } +} + +func TestModelPriceUpsertSyncedPreservesRetainedVerifierSystemError(t *testing.T) { + ctx := context.Background() + db, repo := openTestDB(t) + + if err := repo.ReplaceAll(ctx, map[string]model.ModelPrice{ + "model-a": {Prompt: 1.0, Completion: 2.0}, + }); err != nil { + t.Fatalf("initial ReplaceAll: %v", err) + } + markRawDeleted(t, db) + if _, err := db.Exec(`drop table usage_monitoring_rollup_state`); err != nil { + t.Fatalf("drop monitoring rollup state: %v", err) + } + + result, err := repo.UpsertSynced(ctx, map[string]model.ModelPrice{ + "model-b": {Prompt: 3.0, Completion: 4.0, Source: "sync"}, + }) + if err == nil { + t.Fatalf("expected retained verifier system error, got result=%+v", result) + } + if errors.Is(err, modelprice.ErrStructureChangeAfterRawDeletion) { + t.Fatalf("system error was misclassified as structure conflict: %v", err) + } + + persisted, loadErr := repo.LoadAll(ctx) + if loadErr != nil { + t.Fatalf("LoadAll: %v", loadErr) + } + if len(persisted) != 1 { + t.Fatalf("expected sync rollback, got %d models", len(persisted)) + } + if _, ok := persisted["model-b"]; ok { + t.Fatal("model-b should not exist after verifier system error") + } +} diff --git a/apps/manager-server/internal/repository/usagepricing/projection_read.go b/apps/manager-server/internal/repository/usagepricing/projection_read.go index fd25e47ec..b1c62bbee 100644 --- a/apps/manager-server/internal/repository/usagepricing/projection_read.go +++ b/apps/manager-server/internal/repository/usagepricing/projection_read.go @@ -11,6 +11,16 @@ import ( "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/usageidentity" ) +var retainedPricingColumns = []string{ + "timestamp_ms", "model", "requested_model", "resolved_model", "service_tier", "failed", + "input_tokens", "output_tokens", "reasoning_tokens", "cached_tokens", "cache_tokens", + "cache_read_tokens", "cache_creation_tokens", "normalized_total_input_tokens", "total_tokens", "latency_ms", + "provider", "auth_index", "source", "source_hash", "account_snapshot", "auth_label_snapshot", + "auth_file_snapshot", "auth_provider_snapshot", "auth_project_id_snapshot", "auth_account_id_snapshot", +} + +var ErrRetainedPricingHistoryIncomplete = errors.New("retained pricing history is incomplete") + // LoadHourlyRowsFromEventsTx bypasses a deficient pricing cache without writing // to it. A compatible retained projection supplies archived events; raw events // after its watermark supply the tail, with no overlapping IDs. @@ -35,51 +45,170 @@ func (r *repository) LoadHourlyRowsFromEventsTx(ctx context.Context, tx *sql.Tx, } // LoadAccountRowsFromEventsTx uses the same retained event source and exact -// pricing bands as hourly recovery. The caller must verify core coverage. +// pricing bands as hourly recovery. The retained projection arm is scoped by +// account_key before entering the pricing CTE so degraded account-history reads +// can use the projection's account index instead of scanning retained history. +// The caller must verify core coverage. func (r *repository) LoadAccountRowsFromEventsTx(ctx context.Context, tx *sql.Tx, accountKeys []string) ([]AccountRow, error) { keys := normalizeValues(accountKeys) if len(keys) == 0 { return []AccountRow{}, nil } - source, err := retainedEventSourceTx(ctx, tx) + source, sourceArgs, err := retainedAccountEventSourceTx(ctx, tx, keys) if err != nil { return nil, err } grouped := map[accountKey]*AccountRow{} - if err := mergeAccountRowsFromSource(ctx, tx, 0, keys, grouped, source); err != nil { + if err := mergeAccountRowsFromSourceArgs(ctx, tx, 0, keys, grouped, source, sourceArgs); err != nil { return nil, err } return sortedAccountRows(grouped), nil } -func retainedEventSourceTx(ctx context.Context, tx *sql.Tx) (string, error) { - source := "usage_events" +// VerifyRetainedPricingRebuildSourceTx proves that every raw event removed by +// the supported archive flow is still available from the compatible monitoring +// projection. Structural model-price mutations call this before commit so an +// unrecoverable pricing revision can never become active. +func VerifyRetainedPricingRebuildSourceTx(ctx context.Context, tx *sql.Tx) error { + _, _, err := retainedPricingRebuildSourceTx(ctx, tx) + return err +} + +// retainedPricingRebuildSourceTx returns the exact event source and target ID +// used only for full pricing rebuilds. Live raw rows are authoritative when +// present; the projection contributes only events whose raw rows were deleted. +// This keeps the rebuild source duplicate-free while preserving original event +// IDs for the existing checkpoint state machine. +func retainedPricingRebuildSourceTx(ctx context.Context, tx *sql.Tx) (string, int64, error) { + var hasDeletedRaw bool + if err := tx.QueryRowContext(ctx, `select exists ( + select 1 from usage_archive_event_refs where raw_deleted_at_ms is not null + )`).Scan(&hasDeletedRaw); err != nil { + return "", 0, err + } + if !hasDeletedRaw { + latestID, err := latestEventID(ctx, tx) + return "usage_events", latestID, err + } + var schemaVersion int var revision, status string var coverageID int64 - err := tx.QueryRowContext(ctx, `select schema_version, structure_revision, status, coverage_event_id + if err := tx.QueryRowContext(ctx, `select schema_version, structure_revision, status, coverage_event_id from usage_monitoring_rollup_state where rollup_name = 'projection_v1'`).Scan( &schemaVersion, &revision, &status, &coverageID, + ); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return "", 0, fmt.Errorf("%w: retained usage projection state is unavailable", ErrRetainedPricingHistoryIncomplete) + } + return "", 0, err + } + if schemaVersion != 1 || + revision != usageidentity.MonitoringProjectionStructureRevision() || + status == "clearing" { + return "", 0, fmt.Errorf("%w: retained usage projection is incompatible or rebuilding", ErrRetainedPricingHistoryIncomplete) + } + + var incomplete bool + if err := tx.QueryRowContext(ctx, `select exists ( + select 1 + from usage_archive_event_refs archived + left join usage_monitoring_event_projection_v1 p + on p.event_id = archived.raw_event_id + where archived.raw_deleted_at_ms is not null + and not exists ( + select 1 from usage_events live where live.event_hash = archived.event_hash + ) + and (archived.raw_event_id > ? or p.event_id is null) + )`, coverageID).Scan(&incomplete); err != nil { + return "", 0, err + } + if incomplete { + return "", 0, fmt.Errorf("%w: retained usage projection is missing a deleted pricing event", ErrRetainedPricingHistoryIncomplete) + } + + var latestID int64 + if err := tx.QueryRowContext(ctx, `select coalesce(max(event_id), 0) from ( + select id as event_id from usage_events + union all + select archived.raw_event_id as event_id + from usage_archive_event_refs archived + where archived.raw_deleted_at_ms is not null + and not exists ( + select 1 from usage_events live where live.event_hash = archived.event_hash + ) + )`).Scan(&latestID); err != nil { + return "", 0, err + } + + source := fmt.Sprintf(`(select e.id, e.%s from usage_events e + union all + select p.event_id as id, p.%s + from usage_archive_event_refs archived + join %s p on p.event_id = archived.raw_event_id + where archived.raw_deleted_at_ms is not null + and not exists ( + select 1 from usage_events live where live.event_hash = archived.event_hash + ))`, + strings.Join(retainedPricingColumns, ", e."), + strings.Join(retainedPricingColumns, ", p."), + usageprojection.EventTable, ) - if err != nil && !errors.Is(err, sql.ErrNoRows) { + return source, latestID, nil +} + +func retainedEventSourceTx(ctx context.Context, tx *sql.Tx) (string, error) { + coverageID, compatible, err := retainedProjectionCoverageTx(ctx, tx) + if err != nil { return "", err } - if err == nil && schemaVersion == 1 && coverageID > 0 && - revision == usageidentity.MonitoringProjectionStructureRevision() && status != "clearing" { - // These are the inputs used by the existing pricing-band query. Do not - // synthesize request details or trust an obsolete projection identity. - columns := []string{ - "timestamp_ms", "model", "requested_model", "resolved_model", "service_tier", "failed", - "input_tokens", "output_tokens", "reasoning_tokens", "cached_tokens", "cache_tokens", - "cache_read_tokens", "cache_creation_tokens", "normalized_total_input_tokens", "total_tokens", "latency_ms", - "provider", "auth_index", "source", "source_hash", "account_snapshot", "auth_label_snapshot", - "auth_file_snapshot", "auth_provider_snapshot", "auth_project_id_snapshot", "auth_account_id_snapshot", + if !compatible || coverageID <= 0 { + return "usage_events", nil + } + return fmt.Sprintf(`(select p.event_id as id, p.%s from %s p where p.event_id <= %d + union all select e.id, e.%s from usage_events e where e.id > %d)`, + strings.Join(retainedPricingColumns, ", p."), usageprojection.EventTable, coverageID, + strings.Join(retainedPricingColumns, ", e."), coverageID, + ), nil +} + +func retainedAccountEventSourceTx(ctx context.Context, tx *sql.Tx, accountKeys []string) (string, []any, error) { + coverageID, compatible, err := retainedProjectionCoverageTx(ctx, tx) + if err != nil { + return "", nil, err + } + if !compatible || coverageID <= 0 { + return "usage_events", nil, nil + } + placeholders := strings.TrimRight(strings.Repeat("?,", len(accountKeys)), ",") + args := make([]any, 0, len(accountKeys)) + for _, key := range accountKeys { + args = append(args, key) + } + return fmt.Sprintf(`(select p.event_id as id, p.%s from %s p + where p.account_key in (%s) and p.event_id <= %d + union all select e.id, e.%s from usage_events e where e.id > %d)`, + strings.Join(retainedPricingColumns, ", p."), usageprojection.EventTable, placeholders, coverageID, + strings.Join(retainedPricingColumns, ", e."), coverageID, + ), args, nil +} + +func retainedProjectionCoverageTx(ctx context.Context, tx *sql.Tx) (int64, bool, error) { + var schemaVersion int + var revision, status string + var coverageID int64 + err := tx.QueryRowContext(ctx, `select schema_version, structure_revision, status, coverage_event_id + from usage_monitoring_rollup_state where rollup_name = 'projection_v1'`).Scan( + &schemaVersion, &revision, &status, &coverageID, + ) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return 0, false, nil } - source = fmt.Sprintf(`(select p.event_id as id, p.%s from %s p where p.event_id <= %d - union all select e.id, e.%s from usage_events e where e.id > %d)`, - strings.Join(columns, ", p."), usageprojection.EventTable, coverageID, - strings.Join(columns, ", e."), coverageID, - ) + return 0, false, err } - return source, nil + compatible := schemaVersion == 1 && + revision == usageidentity.MonitoringProjectionStructureRevision() && + status != "clearing" + return coverageID, compatible, nil } diff --git a/apps/manager-server/internal/repository/usagepricing/projection_read_test.go b/apps/manager-server/internal/repository/usagepricing/projection_read_test.go new file mode 100644 index 000000000..e4d7078fa --- /dev/null +++ b/apps/manager-server/internal/repository/usagepricing/projection_read_test.go @@ -0,0 +1,91 @@ +package usagepricing + +import ( + "context" + "database/sql" + "strings" + "testing" + + "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/usageidentity" + _ "modernc.org/sqlite" +) + +func TestRetainedAccountEventSourceUsesProjectionAccountIndex(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + columns := make([]string, 0, len(retainedPricingColumns)+2) + columns = append(columns, "event_id integer primary key", "account_key text not null") + for _, column := range retainedPricingColumns { + columns = append(columns, column+" text") + } + if _, err := db.Exec("create table usage_monitoring_event_projection_v1 (" + strings.Join(columns, ",") + ")"); err != nil { + t.Fatal(err) + } + rawColumns := make([]string, 0, len(retainedPricingColumns)+1) + rawColumns = append(rawColumns, "id integer primary key") + for _, column := range retainedPricingColumns { + rawColumns = append(rawColumns, column+" text") + } + if _, err := db.Exec("create table usage_events (" + strings.Join(rawColumns, ",") + ")"); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`create index idx_usage_monitoring_event_projection_account_window + on usage_monitoring_event_projection_v1(account_key, timestamp_ms, event_id)`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`create table usage_monitoring_rollup_state ( + rollup_name text primary key, + schema_version integer not null, + structure_revision text not null, + status text not null, + coverage_event_id integer not null + )`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`insert into usage_monitoring_rollup_state( + rollup_name, schema_version, structure_revision, status, coverage_event_id + ) values('projection_v1', 1, ?, 'ready', 100)`, + usageidentity.MonitoringProjectionStructureRevision()); err != nil { + t.Fatal(err) + } + + ctx := context.Background() + tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + source, args, err := retainedAccountEventSourceTx(ctx, tx, []string{"account-a", "account-b"}) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(source, "p.account_key in (?,?)") { + t.Fatalf("retained account source is not account-scoped: %s", source) + } + + rows, err := tx.QueryContext(ctx, "explain query plan select id from "+source+" retained", args...) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var details []string + for rows.Next() { + var id, parent, unused int + var detail string + if err := rows.Scan(&id, &parent, &unused, &detail); err != nil { + t.Fatal(err) + } + details = append(details, detail) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + plan := strings.Join(details, "\n") + if !strings.Contains(plan, "idx_usage_monitoring_event_projection_account_window") { + t.Fatalf("retained account projection plan does not use account index: %s", plan) + } +} diff --git a/apps/manager-server/internal/repository/usagepricing/repository.go b/apps/manager-server/internal/repository/usagepricing/repository.go index df81a917e..8de5232d7 100644 --- a/apps/manager-server/internal/repository/usagepricing/repository.go +++ b/apps/manager-server/internal/repository/usagepricing/repository.go @@ -184,23 +184,21 @@ func (r *repository) CatchUp(ctx context.Context, limit int, nowMS int64) (Catch if err != nil { return CatchUpResult{}, err } + rebuildMode := state.StructureRevision != revision || + state.Status == "clearing" || state.Status == "rebuilding" || state.Status == "pending" latestID, err := latestEventID(ctx, tx) if err != nil { return CatchUpResult{}, err } - rebuilt := (state.Status == "pending" || state.Status == "rebuilding" || state.Status == "clearing") && - state.CoverageEventID < state.TargetEventID - if state.StructureRevision != revision || state.Status == "clearing" || state.Status == "rebuilding" || state.Status == "pending" { - var hasDeletedRaw bool - if err := tx.QueryRowContext(ctx, `select exists ( - select 1 from usage_archive_event_refs where raw_deleted_at_ms is not null - )`).Scan(&hasDeletedRaw); err != nil { - return CatchUpResult{}, err - } - if hasDeletedRaw { - return CatchUpResult{}, errors.New("cannot rebuild pricing rollups from incomplete raw usage history") + eventSource := "usage_events" + if rebuildMode { + eventSource, latestID, err = retainedPricingRebuildSourceTx(ctx, tx) + if err != nil { + return CatchUpResult{}, fmt.Errorf("cannot rebuild pricing rollups from retained usage history: %w", err) } } + rebuilt := (state.Status == "pending" || state.Status == "rebuilding" || state.Status == "clearing") && + state.CoverageEventID < state.TargetEventID if state.StructureRevision != revision { if err := resetForRevision(ctx, tx, revision, latestID, nowMS); err != nil { return CatchUpResult{}, err @@ -252,7 +250,7 @@ func (r *repository) CatchUp(ctx context.Context, limit int, nowMS int64) (Catch return CatchUpResult{}, err } } - ids, err := eventIDsThrough(ctx, tx, state.BackfillLastEventID, targetEventID, limit) + ids, err := eventIDsThrough(ctx, tx, eventSource, state.BackfillLastEventID, targetEventID, limit) if err != nil { return CatchUpResult{}, err } @@ -296,13 +294,13 @@ func (r *repository) CatchUp(ctx context.Context, limit int, nowMS int64) (Catch } lastEventID := ids[len(ids)-1] - if err := upsertHourlyBatch(ctx, tx, revision, state.BackfillLastEventID, lastEventID, nowMS); err != nil { + if err := upsertHourlyBatch(ctx, tx, eventSource, revision, state.BackfillLastEventID, lastEventID, nowMS); err != nil { return CatchUpResult{}, err } - if err := upsertAccountBatch(ctx, tx, revision, state.BackfillLastEventID, lastEventID, nowMS); err != nil { + if err := upsertAccountBatch(ctx, tx, eventSource, revision, state.BackfillLastEventID, lastEventID, nowMS); err != nil { return CatchUpResult{}, err } - minBucket, maxBucket, err := batchBucketRange(ctx, tx, state.BackfillLastEventID, lastEventID) + minBucket, maxBucket, err := batchBucketRange(ctx, tx, eventSource, state.BackfillLastEventID, lastEventID) if err != nil { return CatchUpResult{}, err } @@ -509,11 +507,14 @@ func latestEventID(ctx context.Context, tx *sql.Tx) (int64, error) { return id, nil } -func eventIDsThrough(ctx context.Context, tx *sql.Tx, lastEventID, targetEventID int64, limit int) ([]int64, error) { +func eventIDsThrough(ctx context.Context, tx *sql.Tx, source string, lastEventID, targetEventID int64, limit int) ([]int64, error) { if targetEventID <= lastEventID { return []int64{}, nil } - rows, err := tx.QueryContext(ctx, `select id from usage_events where id > ? and id <= ? order by id limit ?`, lastEventID, targetEventID, limit) + rows, err := tx.QueryContext(ctx, fmt.Sprintf( + `select id from %s where id > ? and id <= ? order by id limit ?`, + source, + ), lastEventID, targetEventID, limit) if err != nil { return nil, err } @@ -529,12 +530,12 @@ func eventIDsThrough(ctx context.Context, tx *sql.Tx, lastEventID, targetEventID return ids, rows.Err() } -func batchBucketRange(ctx context.Context, tx *sql.Tx, afterID, throughID int64) (sql.NullInt64, sql.NullInt64, error) { +func batchBucketRange(ctx context.Context, tx *sql.Tx, source string, afterID, throughID int64) (sql.NullInt64, sql.NullInt64, error) { var minBucket, maxBucket sql.NullInt64 err := tx.QueryRowContext(ctx, fmt.Sprintf(`select min(timestamp_ms - (timestamp_ms %% %d)), max(timestamp_ms - (timestamp_ms %% %d)) - from usage_events where id > ? and id <= ?`, hourMS, hourMS), afterID, throughID).Scan(&minBucket, &maxBucket) + from %s where id > ? and id <= ?`, hourMS, hourMS, source), afterID, throughID).Scan(&minBucket, &maxBucket) return minBucket, maxBucket, err } @@ -588,8 +589,8 @@ func bandedEventsFromSourceCTE(whereClause, source string) string { )`, requestedModelExpression, analyticsModelExpression, analyticsModelExpression, accountKeyExpression, source, whereClause, model.ModelPriceBaseContextThreshold) } -func upsertHourlyBatch(ctx context.Context, tx *sql.Tx, revision string, afterID, throughID, nowMS int64) error { - query := bandedEventsCTE("e.id > ? and e.id <= ?") + fmt.Sprintf(` +func upsertHourlyBatch(ctx context.Context, tx *sql.Tx, source, revision string, afterID, throughID, nowMS int64) error { + query := bandedEventsFromSourceCTE("e.id > ? and e.id <= ?", source) + fmt.Sprintf(` insert into usage_pricing_hourly_rollups_v1 ( structure_revision, bucket_ms, model, billing_model, pricing_model, service_tier, context_threshold_tokens, failed, calls, @@ -659,8 +660,8 @@ func upsertHourlyBatch(ctx context.Context, tx *sql.Tx, revision string, afterID return err } -func upsertAccountBatch(ctx context.Context, tx *sql.Tx, revision string, afterID, throughID, nowMS int64) error { - query := bandedEventsCTE("e.id > ? and e.id <= ?") + fmt.Sprintf(` +func upsertAccountBatch(ctx context.Context, tx *sql.Tx, source, revision string, afterID, throughID, nowMS int64) error { + query := bandedEventsFromSourceCTE("e.id > ? and e.id <= ?", source) + fmt.Sprintf(` insert into usage_pricing_account_rollups_v1 ( structure_revision, account_key, account_snapshot, auth_label_snapshot, auth_provider_snapshot, auth_index, source, source_hash, model, @@ -1154,6 +1155,18 @@ func mergeAccountRowsFromSource( accountKeys []string, grouped map[accountKey]*AccountRow, source string, +) error { + return mergeAccountRowsFromSourceArgs(ctx, tx, afterID, accountKeys, grouped, source, nil) +} + +func mergeAccountRowsFromSourceArgs( + ctx context.Context, + tx *sql.Tx, + afterID int64, + accountKeys []string, + grouped map[accountKey]*AccountRow, + source string, + sourceArgs []any, ) error { placeholders := strings.TrimRight(strings.Repeat("?,", len(accountKeys)), ",") query := bandedEventsFromSourceCTE("e.id > ?", source) + fmt.Sprintf(` @@ -1194,7 +1207,8 @@ func mergeAccountRowsFromSource( usage.LongContextInputTokenThreshold, placeholders, ) - args := make([]any, 0, len(accountKeys)+1) + args := make([]any, 0, len(sourceArgs)+len(accountKeys)+1) + args = append(args, sourceArgs...) args = append(args, afterID) for _, key := range accountKeys { args = append(args, key) diff --git a/apps/manager-server/internal/repository/usagepricing/repository_test.go b/apps/manager-server/internal/repository/usagepricing/repository_test.go index cce4b620d..5d746ab75 100644 --- a/apps/manager-server/internal/repository/usagepricing/repository_test.go +++ b/apps/manager-server/internal/repository/usagepricing/repository_test.go @@ -9,6 +9,7 @@ import ( "testing" "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/model" + sqliterepo "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/repository/sqlite" "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/store" "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/testutil" "github.com/seakee/cpa-manager-plus/apps/manager-server/internal/usage" @@ -78,6 +79,110 @@ func TestPricingRollupBandsStrictThresholdsAndMergesRawDelta(t *testing.T) { } } +func TestPricingRollupRebuildsFromRetainedProjectionAfterRawDeletion(t *testing.T) { + ctx := context.Background() + cfg := testutil.NewConfig(t) + st := testutil.NewStore(t, cfg) + price := store.ModelPrice{ + Prompt: 1, + ContextTiers: []store.ModelPriceContextTier{{ + ThresholdTokens: 100, + Prompt: 2, + PromptConfigured: true, + }}, + } + if err := st.SaveModelPrices(ctx, map[string]store.ModelPrice{"resolved-model": price}); err != nil { + t.Fatalf("save initial prices: %v", err) + } + events := []usage.Event{ + pricingEvent("retained-live", 3_600_001, 150), + pricingEvent("retained-deleted-highest", 3_600_002, 250), + } + if _, err := st.UsageEvents.InsertBatch(ctx, events); err != nil { + t.Fatalf("insert retained rebuild events: %v", err) + } + projection, err := st.CatchUpUsageMonitoringProjection(ctx, 10, 9_000) + if err != nil { + t.Fatalf("build monitoring projection: %v", err) + } + if projection.CoverageEventID != 2 || projection.Pending { + t.Fatalf("projection result = %#v", projection) + } + if _, err := st.CatchUpUsagePricing(ctx, 10, 10_000); err != nil { + t.Fatalf("build initial pricing rollup: %v", err) + } + + db, err := sqliterepo.Open(cfg.DBPath) + if err != nil { + t.Fatalf("open raw sqlite handle: %v", err) + } + defer db.Close() + var deletedID, deletedTimestamp int64 + if err := db.QueryRowContext(ctx, `select id, timestamp_ms from usage_events where event_hash = ?`, + events[1].EventHash).Scan(&deletedID, &deletedTimestamp); err != nil { + t.Fatalf("resolve deleted event: %v", err) + } + if _, err := db.ExecContext(ctx, `pragma foreign_keys = off`); err != nil { + t.Fatalf("disable foreign keys: %v", err) + } + if _, err := db.ExecContext(ctx, `insert into usage_archive_event_refs( + event_hash, run_id, segment_sequence, raw_event_id, timestamp_ms, archived_at_ms, raw_deleted_at_ms + ) values(?, 'retained-pricing-run', 1, ?, ?, 11000, 12000)`, + events[1].EventHash, deletedID, deletedTimestamp); err != nil { + t.Fatalf("insert retained archive ref: %v", err) + } + if _, err := db.ExecContext(ctx, `delete from usage_events where id = ?`, deletedID); err != nil { + t.Fatalf("delete highest raw event: %v", err) + } + + price.ContextTiers[0].ThresholdTokens = 200 + if err := st.SaveModelPrices(ctx, map[string]store.ModelPrice{"resolved-model": price}); err != nil { + t.Fatalf("save structural price change with retained history: %v", err) + } + result, err := st.CatchUpUsagePricing(ctx, 10, 13_000) + if err != nil { + t.Fatalf("rebuild pricing from retained history: %v", err) + } + if !result.Rebuilt || result.Pending || result.TargetEventID != deletedID || result.CoverageEventID != deletedID { + t.Fatalf("retained rebuild result = %#v, deletedID=%d", result, deletedID) + } + + rows, _, available, err := st.UsagePricingHourlyRows(ctx, store.UsagePricingHourlyFilter{ + FromMS: 3_600_000, + ToMS: 7_200_000, + IncludeFailed: true, + }) + if err != nil || !available { + t.Fatalf("load retained pricing rows: available=%v err=%v", available, err) + } + var calls int64 + byThreshold := map[int64]int64{} + for _, row := range rows { + calls += row.Calls + byThreshold[row.ContextThresholdTokens] += row.Calls + } + if calls != 2 { + t.Fatalf("retained rebuild calls=%d rows=%#v", calls, rows) + } + if byThreshold[model.ModelPriceBaseContextThreshold] != 1 || byThreshold[200] != 1 { + t.Fatalf("retained rebuild threshold rows=%#v", byThreshold) + } + + accountRows, _, accountAvailable, err := st.UsagePricingAccountRows(ctx, []string{ + pricingAccountKey("team-a.json", "auth-team-a"), + }) + if err != nil || !accountAvailable { + t.Fatalf("load retained account rows: available=%v err=%v", accountAvailable, err) + } + var accountCalls int64 + for _, row := range accountRows { + accountCalls += row.Calls + } + if accountCalls != 2 { + t.Fatalf("retained account calls=%d rows=%#v", accountCalls, accountRows) + } +} + func TestPricingRollupDoesNotClassifyIncrementalBacklogAsRebuild(t *testing.T) { ctx := context.Background() cfg := testutil.NewConfig(t) diff --git a/apps/manager-server/internal/service/monitoring/pricing_coverage_test.go b/apps/manager-server/internal/service/monitoring/pricing_coverage_test.go index 6874d1dd8..1b20f4fcd 100644 --- a/apps/manager-server/internal/service/monitoring/pricing_coverage_test.go +++ b/apps/manager-server/internal/service/monitoring/pricing_coverage_test.go @@ -133,35 +133,40 @@ func TestAnalyticsPricingRecoveryPreservesFiltersAndCollapsedBuckets(t *testing. } } -func TestPricingCatchUpPreservesArchivedHistoryDuringRebuild(t *testing.T) { +func TestPricingCatchUpRebuildsArchivedHistoryFromRetainedProjection(t *testing.T) { for _, test := range []struct { name string sql string }{ {"revision change", `update usage_pricing_rollup_state set structure_revision = 'obsolete'`}, {"resumed clearing", `update usage_pricing_rollup_state set status = 'clearing', coverage_event_id = 0, backfill_last_event_id = 0`}, - {"resumed rebuilding", `update usage_pricing_rollup_state set status = 'rebuilding', coverage_event_id = 0, backfill_last_event_id = 0`}, + {"resumed rebuilding", `delete from usage_pricing_hourly_rollups_v1; + delete from usage_pricing_account_rollups_v1; + update usage_pricing_rollup_state set status = 'rebuilding', coverage_event_id = 0, backfill_last_event_id = 0`}, } { t.Run(test.name, func(t *testing.T) { db, sqlDB, _, _ := pricingCoverageFixture(t) ctx := context.Background() + before := pricingCoverageCounts(t, sqlDB) if _, err := sqlDB.ExecContext(ctx, test.sql); err != nil { t.Fatal(err) } - before := pricingCoverageCounts(t, sqlDB) - stateBefore, err := db.UsagePricingState(ctx) + result, err := db.CatchUpUsagePricing(ctx, 100, time.Now().UnixMilli()) if err != nil { - t.Fatal(err) + t.Fatalf("rebuild from retained projection: %v", err) } - if _, err := db.CatchUpUsagePricing(ctx, 100, time.Now().UnixMilli()); err == nil { - t.Fatal("rebuild from deleted raw history unexpectedly succeeded") + if !result.Rebuilt || result.Pending || result.CoverageEventID < result.TargetEventID { + t.Fatalf("unexpected retained rebuild result: %#v", result) } stateAfter, err := db.UsagePricingState(ctx) if err != nil { t.Fatal(err) } - if after := pricingCoverageCounts(t, sqlDB); after != before || !reflect.DeepEqual(stateAfter, stateBefore) { - t.Fatalf("rebuild changed retained history: before=%v after=%v states=%#v / %#v", before, after, stateBefore, stateAfter) + if stateAfter.Status != "ready" || stateAfter.CoverageEventID < stateAfter.TargetEventID { + t.Fatalf("retained rebuild state = %#v", stateAfter) + } + if after := pricingCoverageCounts(t, sqlDB); after != before { + t.Fatalf("retained rebuild changed historical coverage: before=%v after=%v", before, after) } }) } @@ -190,7 +195,10 @@ func TestAccountHistoryRecoversIncompleteArchivedPricing(t *testing.T) { t.Fatalf("baseline history calls = %d, want 8", calls) } wantCost := analytics.Summary.TotalCost + analytics.SummaryComparison.TotalCost - if delta := want.Items[0].TotalCost + want.Items[1].TotalCost - wantCost; delta < -0.000001 || delta > 0.000001 { + if want.Items[0].TotalCost == nil || want.Items[1].TotalCost == nil { + t.Fatalf("baseline account costs unavailable: %#v", want.Items) + } + if delta := *want.Items[0].TotalCost + *want.Items[1].TotalCost - wantCost; delta < -0.000001 || delta > 0.000001 { t.Fatalf("baseline account costs differ from hourly costs: %v", delta) } if _, err := sqlDB.ExecContext(ctx, test.sql); err != nil { @@ -208,16 +216,107 @@ func TestAccountHistoryRecoversIncompleteArchivedPricing(t *testing.T) { } } -func TestAccountHistoryIncompleteArchivedPricingFailsClosed(t *testing.T) { +func TestAccountHistoryDegradesWhenArchivedPricingCannotBeRecovered(t *testing.T) { + for _, test := range []struct { + name string + catchUp bool + sql string + }{ + {name: "read", sql: `delete from usage_pricing_account_rollups_v1; + delete from usage_monitoring_event_projection_v1 where event_id not in (select id from usage_events)`}, + {name: "catch up", catchUp: true, sql: `delete from usage_pricing_account_rollups_v1; + delete from usage_monitoring_event_projection_v1 where event_id not in (select id from usage_events); + update usage_pricing_rollup_state set structure_revision = 'obsolete'`}, + } { + t.Run(test.name, func(t *testing.T) { + db, sqlDB, _, _ := pricingCoverageFixture(t) + ctx := context.Background() + request := pricingCoverageAccountRequest() + want, err := New(db).AccountHistory(ctx, request) + if err != nil || len(want.Items) != 2 { + t.Fatalf("baseline account history = %#v, %v", want, err) + } + if _, err := sqlDB.ExecContext(ctx, test.sql); err != nil { + t.Fatal(err) + } + request.CatchUp = test.catchUp + got, err := New(db).AccountHistory(ctx, request) + if err != nil || len(got.Items) != len(want.Items) { + t.Fatalf("degraded account history = %#v, %v", got, err) + } + for index := range got.Items { + item, baseline := got.Items[index], want.Items[index] + if !item.Matched || item.TotalCost != nil { + t.Fatalf("degraded item availability = %#v", item) + } + if item.TotalRequests != baseline.TotalRequests || item.SuccessCalls != baseline.SuccessCalls || + item.FailureCalls != baseline.FailureCalls || item.TotalTokens != baseline.TotalTokens || + !reflect.DeepEqual(item.SuccessRate, baseline.SuccessRate) || + !reflect.DeepEqual(item.FirstSeenMS, baseline.FirstSeenMS) || + !reflect.DeepEqual(item.LastSeenMS, baseline.LastSeenMS) { + t.Fatalf("degraded core history differs: got=%#v want=%#v", item, baseline) + } + if !reflect.DeepEqual(item.LatestRequest, baseline.LatestRequest) || + !reflect.DeepEqual(item.RecentRequests, baseline.RecentRequests) { + t.Fatalf("degraded recent requests differ: got=%#v want=%#v", item, baseline) + } + } + }) + } +} + +func TestAccountHistoryIsolatesUnrecoverablePricingByAccount(t *testing.T) { + db, sqlDB, _, _ := pricingCoverageFixture(t) + ctx := context.Background() + request := pricingCoverageAccountRequest() + want, err := New(db).AccountHistory(ctx, request) + if err != nil || len(want.Items) != 2 || want.Items[0].TotalCost == nil || want.Items[1].TotalCost == nil { + t.Fatalf("baseline account history = %#v, %v", want.Items, err) + } + + if _, err := sqlDB.ExecContext(ctx, `delete from usage_pricing_account_rollups_v1 where auth_index = 'a'; + delete from usage_monitoring_event_projection_v1 + where auth_index = 'a' and event_id not in (select id from usage_events)`); err != nil { + t.Fatal(err) + } + got, err := New(db).AccountHistory(ctx, request) + if err != nil || len(got.Items) != len(want.Items) { + t.Fatalf("isolated account history = %#v, %v", got.Items, err) + } + + wantA := want.Items[0] + wantA.TotalCost = nil + if !reflect.DeepEqual(got.Items[0], wantA) { + t.Fatalf("incomplete account was not isolated: got=%#v want=%#v", got.Items[0], wantA) + } + if !reflect.DeepEqual(got.Items[1], want.Items[1]) { + t.Fatalf("complete account pricing was degraded: got=%#v want=%#v", got.Items[1], want.Items[1]) + } +} + +func TestAccountHistoryDoesNotHidePricingRecoveryQueryErrors(t *testing.T) { + db, sqlDB, _, _ := pricingCoverageFixture(t) + ctx := context.Background() + if _, err := sqlDB.ExecContext(ctx, `delete from usage_pricing_account_rollups_v1 where auth_index = 'a'; + alter table usage_monitoring_event_projection_v1 + rename column normalized_total_input_tokens to unavailable_tokens`); err != nil { + t.Fatal(err) + } + _, err := New(db).AccountHistory(ctx, pricingCoverageAccountRequest()) + if err == nil || errors.Is(err, store.ErrUsagePricingCoverageIncomplete) { + t.Fatalf("pricing recovery query error was hidden: %v", err) + } +} + +func TestAccountHistoryDoesNotHideCoreHistoryReadErrors(t *testing.T) { db, sqlDB, _, _ := pricingCoverageFixture(t) ctx := context.Background() - if _, err := sqlDB.ExecContext(ctx, `delete from usage_pricing_account_rollups_v1; - delete from usage_monitoring_event_projection_v1 where event_id not in (select id from usage_events)`); err != nil { + if _, err := sqlDB.ExecContext(ctx, `alter table usage_account_model_rollups rename column total_tokens to unavailable_tokens`); err != nil { t.Fatal(err) } - got, err := New(db).AccountHistory(ctx, pricingCoverageAccountRequest()) - if !errors.Is(err, store.ErrUsagePricingCoverageIncomplete) || len(got.Items) != 0 { - t.Fatalf("incomplete account history returned success: %#v error=%v", got.Items, err) + _, err := New(db).AccountHistory(ctx, pricingCoverageAccountRequest()) + if err == nil || errors.Is(err, store.ErrUsagePricingCoverageIncomplete) { + t.Fatalf("core history read error was hidden: %v", err) } } diff --git a/apps/manager-server/internal/service/monitoring/service.go b/apps/manager-server/internal/service/monitoring/service.go index 91b62ef1c..2433040ae 100644 --- a/apps/manager-server/internal/service/monitoring/service.go +++ b/apps/manager-server/internal/service/monitoring/service.go @@ -289,7 +289,7 @@ type AccountHistoryItem struct { SuccessCalls int64 `json:"success_calls"` FailureCalls int64 `json:"failure_calls"` TotalTokens int64 `json:"total_tokens"` - TotalCost float64 `json:"total_cost"` + TotalCost *float64 `json:"total_cost"` SuccessRate *float64 `json:"success_rate"` FirstSeenMS *int64 `json:"first_seen_ms"` LastSeenMS *int64 `json:"last_seen_ms"` @@ -1773,7 +1773,8 @@ func (s *Service) accountHistory(ctx context.Context, req AccountHistoryRequest) return AccountHistoryResponse{}, err } processed = result.Processed - if _, err := s.store.CatchUpUsagePricing(ctx, accountHistoryCatchUpLimit, generatedAtMS); err != nil { + if _, err := s.store.CatchUpUsagePricing(ctx, accountHistoryCatchUpLimit, generatedAtMS); err != nil && + !errors.Is(err, store.ErrUsagePricingCoverageIncomplete) { return AccountHistoryResponse{}, err } } @@ -1838,15 +1839,20 @@ func (s *Service) accountHistory(ctx context.Context, req AccountHistoryRequest) if err != nil { return nil, err } - prices := pricingSnapshot.Prices - if pricingSnapshot.Available { - return buildPricingAccountHistoryTotals(pricingSnapshot.Rows, prices), nil + totals := buildPricingAccountHistoryTotals(pricingSnapshot.Rows, pricingSnapshot.Prices) + if len(pricingSnapshot.PricingIncompleteAccountKeys) == 0 { + return totals, nil } - rows, err := s.store.AccountHistoryRollupRows(ctx, readKeys) - if err != nil { - return nil, err + unpricedRows := make([]store.AccountHistoryRollupRow, 0, len(pricingSnapshot.CoreRows)) + for _, row := range pricingSnapshot.CoreRows { + if _, incomplete := pricingSnapshot.PricingIncompleteAccountKeys[row.AccountKey]; incomplete { + unpricedRows = append(unpricedRows, row) + } + } + for key, total := range buildUnpricedAccountHistoryTotals(unpricedRows) { + totals[key] = total } - return buildAccountHistoryTotals(rows, prices), nil + return totals, nil } totals, err := loadTotals(keys) if err != nil { @@ -1935,7 +1941,7 @@ func (s *Service) accountHistory(ctx context.Context, req AccountHistoryRequest) SuccessCalls: total.successCalls, FailureCalls: total.failureCalls, TotalTokens: total.totalTokens, - TotalCost: total.cost, + TotalCost: accountHistoryCostPointer(total), SuccessRate: successRate, FirstSeenMS: nullableMSPointer(total.firstSeenMS), LastSeenMS: nullableMSPointer(total.lastSeenMS), @@ -4072,13 +4078,14 @@ func buildHeaderSnapshots(items []store.HeaderSnapshot) []HeaderSnapshot { } type accountHistoryTotal struct { - requests int64 - successCalls int64 - failureCalls int64 - totalTokens int64 - cost float64 - firstSeenMS int64 - lastSeenMS int64 + requests int64 + successCalls int64 + failureCalls int64 + totalTokens int64 + cost float64 + costAvailable bool + firstSeenMS int64 + lastSeenMS int64 } func accountHistoryTargetKey(target AccountHistoryTarget) (string, bool) { @@ -4181,7 +4188,7 @@ func accountLatestRequestFromStore(request store.LatestAccountRequest) *AccountL } } -func buildAccountHistoryTotals(rows []store.AccountHistoryRollupRow, prices map[string]store.ModelPrice) map[string]*accountHistoryTotal { +func buildUnpricedAccountHistoryTotals(rows []store.AccountHistoryRollupRow) map[string]*accountHistoryTotal { totals := map[string]*accountHistoryTotal{} for _, row := range rows { total := totals[row.AccountKey] @@ -4193,23 +4200,6 @@ func buildAccountHistoryTotals(rows []store.AccountHistoryRollupRow, prices map[ total.successCalls += row.SuccessCalls total.failureCalls += row.FailureCalls total.totalTokens += row.TotalTokens - total.cost += pricing.CostForModelCandidatesWithServiceTier( - []string{row.BillingModel, row.Model}, - row.ServiceTier, - pricing.ModelTokens{ - InputTokens: row.InputTokens, - OutputTokens: row.OutputTokens, - CachedTokens: row.CachedTokens, - CacheReadTokens: row.CacheReadTokens, - CacheCreationTokens: row.CacheCreationTokens, - LongInputTokens: row.LongInputTokens, - LongOutputTokens: row.LongOutputTokens, - LongCachedTokens: row.LongCachedTokens, - LongCacheReadTokens: row.LongCacheReadTokens, - LongCacheCreationTokens: row.LongCacheCreationTokens, - }, - prices, - ) if total.firstSeenMS == 0 || (row.FirstSeenMS > 0 && row.FirstSeenMS < total.firstSeenMS) { total.firstSeenMS = row.FirstSeenMS } @@ -4225,7 +4215,7 @@ func buildPricingAccountHistoryTotals(rows []store.UsagePricingAccountRow, price for _, row := range rows { total := totals[row.AccountKey] if total == nil { - total = &accountHistoryTotal{} + total = &accountHistoryTotal{costAvailable: true} totals[row.AccountKey] = total } total.requests += row.Calls @@ -4278,6 +4268,7 @@ func mergeAliasedAccountHistoryTotals(totals map[string]*accountHistoryTotal, al primary.failureCalls += legacy.failureCalls primary.totalTokens += legacy.totalTokens primary.cost += legacy.cost + primary.costAvailable = primary.costAvailable && legacy.costAvailable if primary.firstSeenMS == 0 || (legacy.firstSeenMS > 0 && legacy.firstSeenMS < primary.firstSeenMS) { primary.firstSeenMS = legacy.firstSeenMS } @@ -4288,6 +4279,14 @@ func mergeAliasedAccountHistoryTotals(totals map[string]*accountHistoryTotal, al } } +func accountHistoryCostPointer(total *accountHistoryTotal) *float64 { + if total == nil || !total.costAvailable { + return nil + } + value := total.cost + return &value +} + func accountWindowUsageTargetKey(target AccountWindowUsageTarget) (string, bool) { if !AccountWindowUsageTargetHasCredentialIdentity(target) { return "", false diff --git a/apps/manager-server/internal/service/monitoring/service_test.go b/apps/manager-server/internal/service/monitoring/service_test.go index 7ecea05cd..9c842cf9c 100644 --- a/apps/manager-server/internal/service/monitoring/service_test.go +++ b/apps/manager-server/internal/service/monitoring/service_test.go @@ -2249,7 +2249,7 @@ func TestAccountHistoryReturnsRollupTotalsAndCost(t *testing.T) { if history.SuccessRate == nil || math.Abs(*history.SuccessRate-0.5) > 0.000001 { t.Fatalf("success rate = %#v", history.SuccessRate) } - if math.Abs(history.TotalCost-2.055) > 0.000001 { + if history.TotalCost == nil || math.Abs(*history.TotalCost-2.055) > 0.000001 { t.Fatalf("total cost = %v", history.TotalCost) } if history.FirstSeenMS == nil || *history.FirstSeenMS != baseMS+1_000 || history.LastSeenMS == nil || *history.LastSeenMS != baseMS+2_000 { @@ -2477,7 +2477,7 @@ func TestAccountHistoryPricesContextTierBands(t *testing.T) { t.Fatalf("history item = %#v", resp.Items) } const wantCost = 4.60002 - if math.Abs(resp.Items[0].TotalCost-wantCost) > 0.000001 { + if resp.Items[0].TotalCost == nil || math.Abs(*resp.Items[0].TotalCost-wantCost) > 0.000001 { t.Fatalf("history cost = %v, want %v", resp.Items[0].TotalCost, wantCost) } } diff --git a/apps/manager-server/internal/store/store.go b/apps/manager-server/internal/store/store.go index a5e1a32a9..3c9e4ff17 100644 --- a/apps/manager-server/internal/store/store.go +++ b/apps/manager-server/internal/store/store.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "database/sql" + "errors" "fmt" "io" "sync" @@ -132,10 +133,12 @@ type UsageHourlyPricingSnapshot struct { } type UsagePricingAccountSnapshot struct { - Rows []UsagePricingAccountRow - State UsagePricingState - Available bool - Prices map[string]ModelPrice + CoreRows []AccountHistoryRollupRow + Rows []UsagePricingAccountRow + State UsagePricingState + Available bool + PricingIncompleteAccountKeys map[string]struct{} + Prices map[string]ModelPrice } type Store struct { @@ -547,7 +550,11 @@ func (s *Store) CatchUpUsagePricing(ctx context.Context, limit int, nowMS int64) if !ready { return UsagePricingCatchUpResult{Pending: true}, nil } - return s.UsagePricing.CatchUp(ctx, limit, nowMS) + result, err := s.UsagePricing.CatchUp(ctx, limit, nowMS) + if err != nil && errors.Is(err, usagepricing.ErrRetainedPricingHistoryIncomplete) { + return result, fmt.Errorf("%w: %w", ErrUsagePricingCoverageIncomplete, err) + } + return result, err } func (s *Store) RecordUsagePricingFailure(ctx context.Context, rollupErr error, nowMS int64) error { @@ -715,17 +722,49 @@ func (s *Store) LoadUsagePricingAccountSnapshot(ctx context.Context, accountKeys if err != nil { return UsagePricingAccountSnapshot{}, err } - rows, state, available, pricingErr := s.UsagePricing.LoadAccountRowsTx(ctx, tx, accountKeys) - if pricingErr != nil || !available || !accountPricingCoverageMatches(coreRows, rows) { - rows, err = s.UsagePricing.LoadAccountRowsFromEventsTx(ctx, tx, accountKeys) - if err != nil { - return UsagePricingAccountSnapshot{}, fmt.Errorf("%w: retained account event query: %w", ErrUsagePricingCoverageIncomplete, err) + rows, state, available, err := s.UsagePricing.LoadAccountRowsTx(ctx, tx, accountKeys) + if err != nil { + return UsagePricingAccountSnapshot{}, err + } + if !available { + rows = nil + } + + incomplete := accountPricingCoverageIncompleteKeys(coreRows, rows) + if len(incomplete) > 0 { + recoveryKeys := make([]string, 0, len(incomplete)) + for key := range incomplete { + recoveryKeys = append(recoveryKeys, key) + } + recoveredRows, recoveryErr := s.UsagePricing.LoadAccountRowsFromEventsTx(ctx, tx, recoveryKeys) + if recoveryErr != nil { + return UsagePricingAccountSnapshot{}, recoveryErr } - if !accountPricingCoverageMatches(coreRows, rows) { - return UsagePricingAccountSnapshot{}, ErrUsagePricingCoverageIncomplete + + recoveryCoreRows := make([]AccountHistoryRollupRow, 0, len(coreRows)) + for _, row := range coreRows { + if _, needsRecovery := incomplete[row.AccountKey]; needsRecovery { + recoveryCoreRows = append(recoveryCoreRows, row) + } + } + stillIncomplete := accountPricingCoverageIncompleteKeys(recoveryCoreRows, recoveredRows) + + completeRows := make([]UsagePricingAccountRow, 0, len(rows)+len(recoveredRows)) + for _, row := range rows { + if _, needsRecovery := incomplete[row.AccountKey]; !needsRecovery { + completeRows = append(completeRows, row) + } + } + for _, row := range recoveredRows { + if _, unresolved := stillIncomplete[row.AccountKey]; !unresolved { + completeRows = append(completeRows, row) + } } - available = true + rows = completeRows + incomplete = stillIncomplete } + available = len(incomplete) == 0 + prices, err := s.ModelPrices.LoadAllTx(ctx, tx) if err != nil { return UsagePricingAccountSnapshot{}, err @@ -734,10 +773,12 @@ func (s *Store) LoadUsagePricingAccountSnapshot(ctx context.Context, accountKeys return UsagePricingAccountSnapshot{}, err } return UsagePricingAccountSnapshot{ - Rows: rows, - State: state, - Available: available, - Prices: prices, + CoreRows: coreRows, + Rows: rows, + State: state, + Available: available, + PricingIncompleteAccountKeys: incomplete, + Prices: prices, }, nil } diff --git a/apps/manager-server/internal/store/usage_pricing_coverage.go b/apps/manager-server/internal/store/usage_pricing_coverage.go index 94b3cf021..209cb744d 100644 --- a/apps/manager-server/internal/store/usage_pricing_coverage.go +++ b/apps/manager-server/internal/store/usage_pricing_coverage.go @@ -85,7 +85,7 @@ func addAccountPricingCoverage(grouped map[accountPricingCoverageKey]accountPric grouped[key] = entry } -func accountPricingCoverageMatches(core []AccountHistoryRollupRow, prices []UsagePricingAccountRow) bool { +func accountPricingCoverageIncompleteKeys(core []AccountHistoryRollupRow, prices []UsagePricingAccountRow) map[string]struct{} { coreTotals := make(map[accountPricingCoverageKey]accountPricingCoverage, len(core)) priceTotals := make(map[accountPricingCoverageKey]accountPricingCoverage, len(prices)) for _, row := range core { @@ -110,13 +110,20 @@ func accountPricingCoverageMatches(core []AccountHistoryRollupRow, prices []Usag row.LongCacheReadTokens, row.LongCacheCreationTokens, row.TotalTokens, }, row.FirstSeenMS, row.LastSeenMS) } - if len(coreTotals) != len(priceTotals) { - return false - } + incomplete := make(map[string]struct{}) for key, expected := range coreTotals { if actual, exists := priceTotals[key]; !exists || actual != expected { - return false + incomplete[key.accountKey] = struct{}{} } } - return true + for key := range priceTotals { + if _, exists := coreTotals[key]; !exists { + incomplete[key.accountKey] = struct{}{} + } + } + return incomplete +} + +func accountPricingCoverageMatches(core []AccountHistoryRollupRow, prices []UsagePricingAccountRow) bool { + return len(accountPricingCoverageIncompleteKeys(core, prices)) == 0 } diff --git a/apps/web/src/features/accounts/AccountsPage.tsx b/apps/web/src/features/accounts/AccountsPage.tsx index 7c3d6678f..962c153ba 100644 --- a/apps/web/src/features/accounts/AccountsPage.tsx +++ b/apps/web/src/features/accounts/AccountsPage.tsx @@ -8820,8 +8820,12 @@ export function AccountsPage() { key: 'cost', icon: , className: styles.accountHistoryMetricCost, - value: matched ? formatCompactUsd(accountHistory.total_cost) : '-', - exact: matched ? formatUsd(accountHistory.total_cost) : '-', + value: + matched && accountHistory.total_cost !== null + ? formatCompactUsd(accountHistory.total_cost) + : '-', + exact: + matched && accountHistory.total_cost !== null ? formatUsd(accountHistory.total_cost) : '-', }, { key: 'success', diff --git a/apps/web/src/features/accounts/components/accountDetail/AccountQuotaTab.tsx b/apps/web/src/features/accounts/components/accountDetail/AccountQuotaTab.tsx index 6ff2f3bab..0698eaaa3 100644 --- a/apps/web/src/features/accounts/components/accountDetail/AccountQuotaTab.tsx +++ b/apps/web/src/features/accounts/components/accountDetail/AccountQuotaTab.tsx @@ -229,7 +229,7 @@ export function AccountQuotaTab({ icon={} tone="amber" label={t('accounts.detail_total_cost')} - value={history ? formatUsd(history.totalCost) : '-'} + value={history?.totalCost != null ? formatUsd(history.totalCost) : '-'} /> } diff --git a/apps/web/src/features/accounts/model/accountDetailViewModel.test.ts b/apps/web/src/features/accounts/model/accountDetailViewModel.test.ts index fcc757d55..15922fa45 100644 --- a/apps/web/src/features/accounts/model/accountDetailViewModel.test.ts +++ b/apps/web/src/features/accounts/model/accountDetailViewModel.test.ts @@ -244,6 +244,19 @@ const makeMonitoringValue = ( }); describe('accountDetailViewModel', () => { + it('preserves unavailable historical cost without dropping core history', () => { + const viewModel = buildAccountDetailViewModel(makeRow(), { + history: makeHistory({ total_cost: null }), + }); + + expect(viewModel.history).toMatchObject({ + matched: true, + totalRequests: 12, + totalTokens: 2400, + totalCost: null, + }); + }); + it('uses the full unified plan label for credential details', () => { const viewModel = buildAccountDetailViewModel( makeRow({ diff --git a/apps/web/src/features/accounts/model/accountDetailViewModel.ts b/apps/web/src/features/accounts/model/accountDetailViewModel.ts index 7e4035ce4..09459512c 100644 --- a/apps/web/src/features/accounts/model/accountDetailViewModel.ts +++ b/apps/web/src/features/accounts/model/accountDetailViewModel.ts @@ -208,7 +208,7 @@ export interface AccountDetailHistorySummary { successCalls: number; failureCalls: number; totalTokens: number; - totalCost: number; + totalCost: number | null; successRate: number | null; firstSeenMs: number | null; lastSeenMs: number | null; diff --git a/apps/web/src/features/accounts/model/accountsPagePresentation.test.ts b/apps/web/src/features/accounts/model/accountsPagePresentation.test.ts index 34194b677..ead014aca 100644 --- a/apps/web/src/features/accounts/model/accountsPagePresentation.test.ts +++ b/apps/web/src/features/accounts/model/accountsPagePresentation.test.ts @@ -57,6 +57,7 @@ describe('accountsPagePresentation', () => { expect(parsePriorityValue('1.2')).toBeNull(); expect(formatHistorySuccessRate(0.975)).toBe('97.5%'); expect(formatMoney(12.34)).toBe('$12.34'); + expect(formatMoney(null)).toBe('-'); expect(quotaStatusLabelKey('exhausted')).toBe('accounts.quota_status_exhausted'); }); diff --git a/apps/web/src/features/accounts/model/accountsPagePresentation.ts b/apps/web/src/features/accounts/model/accountsPagePresentation.ts index 37f605521..0505a9753 100644 --- a/apps/web/src/features/accounts/model/accountsPagePresentation.ts +++ b/apps/web/src/features/accounts/model/accountsPagePresentation.ts @@ -111,7 +111,8 @@ export const getProviderLabel = (provider: string, t: TFunction) => { export const formatPercent = (value: number | null | undefined, digits = 0) => typeof value === 'number' && Number.isFinite(value) ? `${value.toFixed(digits)}%` : '-'; -export const formatMoney = (value: number) => formatUsd(value); +export const formatMoney = (value: number | null | undefined) => + typeof value === 'number' && Number.isFinite(value) ? formatUsd(value) : '-'; export const formatHistoryNumber = (value: number, locale: string) => { const numberValue = Number(value); diff --git a/apps/web/src/features/config/ConfigPage.persistence.test.tsx b/apps/web/src/features/config/ConfigPage.persistence.test.tsx index 54963c756..92e6f2e19 100644 --- a/apps/web/src/features/config/ConfigPage.persistence.test.tsx +++ b/apps/web/src/features/config/ConfigPage.persistence.test.tsx @@ -3,6 +3,8 @@ import type { ReactNode } from 'react'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { ApiKeyMutation } from '@/components/config/ApiKeysCardEditor'; import type { ManagerConfigResponse } from '@/services/api/usageService'; +import type { VisualConfigValues } from '@/types/visualConfig'; +import { parse as parseYaml, stringify as stringifyYaml } from 'yaml'; vi.mock('react-dom', () => ({ createPortal: (children: ReactNode) => children, @@ -32,6 +34,7 @@ const mocks = vi.hoisted(() => ({ capturedApiKeyOperationStart: null as (() => void) | null, capturedApiKeyOperationEnd: null as (() => void) | null, translate: (key: string) => key, + useRealVisualConfig: false, visualState: { apiKeysText: 'sk-old', dirty: false, @@ -54,10 +57,14 @@ vi.mock('@/hooks/useMediaQuery', () => ({ vi.mock('@/components/config/VisualConfigEditor', () => ({ VisualConfigEditor: ({ + values, + onChange, onPersistApiKeyMutation, onApiKeyOperationStart, onApiKeyOperationEnd, }: { + values: VisualConfigValues; + onChange: (values: Partial) => void; onPersistApiKeyMutation: (mutation: ApiKeyMutation) => Promise; onApiKeyOperationStart: () => void; onApiKeyOperationEnd: () => void; @@ -78,6 +85,8 @@ vi.mock('@/components/config/VisualConfigEditor', () => ({ return (
+ {values.apiKeysText} +