From 4e624a25c2ad223f9bd0e7bb5ec78df592e0e713 Mon Sep 17 00:00:00 2001 From: Maycon Santos Date: Sat, 25 Jul 2026 19:31:55 +0000 Subject: [PATCH] [e2e] validate stored totals and cache costs against published rates The provider matrix now sends a long-form prompt, recomputes each access-log row's expected cost_usd and cache_cost_usd from hardcoded published per-1k rates and the row's persisted token buckets, and finishes with a raw-SQL audit of every usage row straight in the management sqlite store. Bedrock defaults to Sonnet 4.6 and the workflow gains a bedrock_model dispatch input plus a push trigger on this branch while it is under review. --- .github/workflows/agent-network-e2e.yml | 13 ++ e2e/agentnetwork/chat_test.go | 183 +++++++++++++++++++++++- e2e/harness/client.go | 10 +- e2e/harness/combined.go | 23 +++ 4 files changed, 217 insertions(+), 12 deletions(-) diff --git a/.github/workflows/agent-network-e2e.yml b/.github/workflows/agent-network-e2e.yml index bf4868871..90deaf9ac 100644 --- a/.github/workflows/agent-network-e2e.yml +++ b/.github/workflows/agent-network-e2e.yml @@ -5,6 +5,17 @@ on: schedule: - cron: "0 3 * * *" workflow_dispatch: + inputs: + bedrock_model: + description: >- + Bedrock inference-profile id to drive the matrix with, exactly as + AWS issues it. Leave empty for the Sonnet 4.6 default. + required: false + default: "" + # Auto-run on pushes to the cache-accounting branch until its PR opens. + push: + branches: + - "anet-cache-cost-accounting" concurrency: group: ${{ github.workflow }}-${{ github.ref }} @@ -62,6 +73,8 @@ jobs: CLOUDFLARE_TOKEN: ${{ secrets.E2E_CLOUDFLARE_TOKEN }} AWS_BEARER_TOKEN_BEDROCK: ${{ secrets.E2E_AWS_BEARER_TOKEN_BEDROCK }} AWS_REGION: ${{ secrets.E2E_AWS_REGION }} + # Bedrock model override: dispatch input wins, then the repo variable, else the test default. + AWS_BEDROCK_MODEL: ${{ inputs.bedrock_model || vars.E2E_AWS_BEDROCK_MODEL }} # Vertex (Anthropic-on-Vertex): SA + project required; region defaults # to "global", model to a pinned claude snapshot. GOOGLE_VERTEX_SA_BASE64: ${{ secrets.E2E_GOOGLE_VERTEX_SA_BASE64 }} diff --git a/e2e/agentnetwork/chat_test.go b/e2e/agentnetwork/chat_test.go index 90c3766ec..d5582205f 100644 --- a/e2e/agentnetwork/chat_test.go +++ b/e2e/agentnetwork/chat_test.go @@ -9,12 +9,169 @@ import ( "testing" "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" "github.com/netbirdio/netbird/e2e/harness" "github.com/netbirdio/netbird/shared/management/http/api" ) +// per1k is a model's published USD rates per 1k tokens. read is the prompt-cache read rate +// (OpenAI: the cached-input discount rate); write is the cache-creation rate where one exists. +type per1k struct{ in, out, read, write float64 } + +// publishedPer1k hardcodes the vendors' PUBLISHED rates for the models the live matrix can drive, +// keyed by the normalized model id the proxy stamps. Deliberately independent of the proxy's +// pricing table so a wrong embedded rate or a broken normalization fails the run. +var publishedPer1k = map[string]per1k{ + "gpt-4o-mini": {0.00015, 0.0006, 0.000075, 0}, + "gpt-4o": {0.0025, 0.01, 0.00125, 0}, + "claude-haiku-4-5": {0.001, 0.005, 0.0001, 0.00125}, + "claude-sonnet-4-5": {0.003, 0.015, 0.0003, 0.00375}, + "claude-sonnet-4-6": {0.003, 0.015, 0.0003, 0.00375}, + "kimi-k3": {0.003, 0.015, 0.0003, 0.003}, // no published write rate: bills at the input rate + "anthropic.claude-haiku-4-5": {0.001, 0.005, 0.0001, 0.00125}, + "anthropic.claude-sonnet-4-5": {0.003, 0.015, 0.0003, 0.00375}, + "anthropic.claude-sonnet-4-6": {0.003, 0.015, 0.0003, 0.00375}, +} + +// rawCostVerificationSQL is the operator-facing double-check, run straight against the management +// sqlite store: recompute each usage row's expected total and cache cost from its own persisted +// token buckets and hardcoded published rates. OpenAI counts cached tokens as a subset of input; +// Anthropic-shape providers count cache buckets additively. +const rawCostVerificationSQL = ` +WITH rates(model, in_rate, out_rate, read_rate, write_rate) AS ( + VALUES + ('gpt-4o-mini', 0.00015, 0.0006, 0.000075, 0.0), + ('gpt-4o', 0.0025, 0.01, 0.00125, 0.0), + ('claude-haiku-4-5', 0.001, 0.005, 0.0001, 0.00125), + ('claude-sonnet-4-5', 0.003, 0.015, 0.0003, 0.00375), + ('claude-sonnet-4-6', 0.003, 0.015, 0.0003, 0.00375), + ('kimi-k3', 0.003, 0.015, 0.0003, 0.003), + ('anthropic.claude-haiku-4-5', 0.001, 0.005, 0.0001, 0.00125), + ('anthropic.claude-sonnet-4-5', 0.003, 0.015, 0.0003, 0.00375), + ('anthropic.claude-sonnet-4-6', 0.003, 0.015, 0.0003, 0.00375) +) +SELECT + u.provider, + u.model, + u.input_tokens, + u.output_tokens, + u.cached_input_tokens, + u.cache_creation_tokens, + u.cost_usd, + u.cache_cost_usd, + CASE WHEN u.provider = 'openai' THEN + (u.input_tokens - MIN(u.cached_input_tokens, u.input_tokens))*r.in_rate/1000.0 + + MIN(u.cached_input_tokens, u.input_tokens)*r.read_rate/1000.0 + + u.output_tokens*r.out_rate/1000.0 + ELSE + u.input_tokens*r.in_rate/1000.0 + u.cached_input_tokens*r.read_rate/1000.0 + + u.cache_creation_tokens*r.write_rate/1000.0 + u.output_tokens*r.out_rate/1000.0 + END AS expected_total, + CASE WHEN u.provider = 'openai' THEN + MIN(u.cached_input_tokens, u.input_tokens)*r.read_rate/1000.0 + ELSE + u.cached_input_tokens*r.read_rate/1000.0 + u.cache_creation_tokens*r.write_rate/1000.0 + END AS expected_cache +FROM agent_network_request_usage u +JOIN rates r ON r.model = u.model +ORDER BY u.timestamp` + +// verifyUsageRowsSQL re-checks every persisted usage row directly in the management sqlite store, +// bypassing the API path — the same audit an operator can run on a production store.db. +func verifyUsageRowsSQL(t *testing.T, srv *harness.Combined) { + t.Helper() + + dbPath, err := srv.SnapshotStoreDB(t.TempDir()) + require.NoError(t, err, "snapshot management sqlite store") + + db, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{}) + require.NoError(t, err, "open store snapshot") + sqlDB, err := db.DB() + require.NoError(t, err) + defer func() { _ = sqlDB.Close() }() + + rows, err := db.Raw(rawCostVerificationSQL).Rows() + require.NoError(t, err, "run raw cost verification query") + defer func() { _ = rows.Close() }() + + verified := 0 + for rows.Next() { + var provider, model string + var inTok, outTok, readTok, writeTok int64 + var cost, cacheCost, wantTotal, wantCache float64 + require.NoError(t, rows.Scan(&provider, &model, &inTok, &outTok, &readTok, &writeTok, &cost, &cacheCost, &wantTotal, &wantCache), "scan usage row") + t.Logf("[sql] %s/%s: in=%d out=%d cache_read=%d cache_write=%d stored=$%.6f/$%.6f expected=$%.6f/$%.6f", + provider, model, inTok, outTok, readTok, writeTok, cost, cacheCost, wantTotal, wantCache) + assert.InDeltaf(t, wantTotal, cost, 1e-6, "stored cost_usd for %s/%s must match the published-rate recompute", provider, model) + assert.InDeltaf(t, wantCache, cacheCost, 1e-6, "stored cache_cost_usd for %s/%s must match the published-rate recompute", provider, model) + verified++ + } + require.NoError(t, rows.Err(), "iterate usage rows") + require.Positive(t, verified, "raw SQL check must cover at least one usage row") + t.Logf("[sql] verified %d usage rows in store.db against published rates", verified) + + gwRows, err := db.Raw(`SELECT model, cost_usd FROM agent_network_request_usage WHERE model LIKE '%/%'`).Rows() + require.NoError(t, err, "query gateway-prefixed usage rows") + defer func() { _ = gwRows.Close() }() + for gwRows.Next() { + var model string + var cost float64 + require.NoError(t, gwRows.Scan(&model, &cost), "scan gateway usage row") + t.Logf("[sql] gateway %s: stored=$%.6f (must be 0 — deliberately unpriced)", model, cost) + assert.Zerof(t, cost, "gateway-prefixed model %q must store cost 0, never a guessed rate", model) + } + require.NoError(t, gwRows.Err(), "iterate gateway usage rows") +} + +// validateAccessLogCost recomputes a live access-log row's expected total and cache cost from the +// published per-1k rates and the row's persisted token buckets, and asserts both stored values. +// Gateway-prefixed model ids the proxy deliberately does not price must store cost 0. +func validateAccessLogCost(t *testing.T, pc providerCase, row api.AgentNetworkAccessLog) { + t.Helper() + model := catalogModel(pc) + provider := "" + if row.Provider != nil { + provider = *row.Provider + } + t.Logf("[cost] %s: provider=%s model=%s in=%d out=%d total=%d cache_read=%d cache_write=%d cost=$%.6f cache_cost=$%.6f", + pc.name, provider, model, row.InputTokens, row.OutputTokens, row.TotalTokens, + row.CachedInputTokens, row.CacheCreationTokens, row.CostUsd, row.CacheCostUsd) + + rates, known := publishedPer1k[model] + if !known { + if strings.Contains(model, "/") { + assert.Zerof(t, row.CostUsd, "gateway-prefixed model %q is not priced so the cost meter must skip (cost 0)", model) + return + } + t.Logf("[cost] %s: no published rate on file for model %q (env-overridden?); skipping cost validation", pc.name, model) + return + } + + // input_tokens may legitimately be 0: Moonshot/Kimi reports fully cached prompts under the cache + // buckets only. Output and total must always be present on a priced row. + require.Positive(t, row.OutputTokens, "priced row must carry output tokens") + require.Positive(t, row.TotalTokens, "priced row must carry total tokens") + + var wantTotal, wantCache float64 + if provider == "openai" { + cached := min(row.CachedInputTokens, row.InputTokens) // cached is a subset of input + wantCache = float64(cached) / 1000 * rates.read + wantTotal = float64(row.InputTokens-cached)/1000*rates.in + wantCache + float64(row.OutputTokens)/1000*rates.out + } else { + // Anthropic / Bedrock shape: cache buckets are additive to input_tokens. + wantCache = float64(row.CachedInputTokens)/1000*rates.read + float64(row.CacheCreationTokens)/1000*rates.write + wantTotal = float64(row.InputTokens)/1000*rates.in + wantCache + float64(row.OutputTokens)/1000*rates.out + } + + t.Logf("[cost] %s: expecting total=$%.6f cache=$%.6f from published rates", pc.name, wantTotal, wantCache) + assert.InDeltaf(t, wantTotal, row.CostUsd, 1e-6, "stored cost_usd for %s (%s)", pc.name, model) + assert.InDeltaf(t, wantCache, row.CacheCostUsd, 1e-6, "stored cache_cost_usd for %s (%s)", pc.name, model) +} + // providerCase is one entry in the live provider matrix. The same scenario runs // for every available provider; availability is keyed off env vars so the suite // covers whatever credentials are present (source ~/.llm-keys locally / set the @@ -116,12 +273,12 @@ func availableProviders() []providerCase { if region == "" { region = "eu-central-1" } - // A valid Bedrock inference-profile id (region prefix + date + version), - // overridable per account. `global.` profiles can be invoked from any - // region; set AWS_BEDROCK_MODEL to match the enabled profile for the token. + // A valid Bedrock inference-profile id, overridable per account (AWS_BEDROCK_MODEL, also the + // workflow's bedrock_model dispatch input). `global.` profiles work from any region. Defaults to + // Sonnet 4.6, whose id convention dropped the -YYYYMMDD-v1:0 suffix that Haiku 4.5 still carries. model := os.Getenv("AWS_BEDROCK_MODEL") if model == "" { - model = "global.anthropic.claude-haiku-4-5-20251001-v1:0" + model = "global.anthropic.claude-sonnet-4-6" } ps = append(ps, providerCase{name: "bedrock", catalogID: "bedrock_api", upstream: "https://bedrock-runtime." + region + ".amazonaws.com", apiKey: k, model: model, kind: harness.WireBedrock}) } @@ -257,6 +414,10 @@ func TestProvidersMatrix(t *testing.T) { // session id and confirm the marker propagated end-to-end. sessionID := "e2e-session-" + pc.name + // A long-form prompt so completions carry realistic token counts for cost validation; + // max_tokens in the harness bodies (2048) lets the full answer through. + const matrixPrompt = "explain GitHub workflow in 1000 words" + // Retry briefly to absorb tunnel/DNS jitter on the first call. var code int var body string @@ -267,11 +428,11 @@ func TestProvidersMatrix(t *testing.T) { var cerr error switch pc.kind { case harness.WireVertex: - c, b, cerr = cl.Vertex(ctx, settings.Endpoint, proxyIP, pc.project, pc.region, pc.model, "Reply with exactly: pong", sessionID) + c, b, cerr = cl.Vertex(ctx, settings.Endpoint, proxyIP, pc.project, pc.region, pc.model, matrixPrompt, sessionID) case harness.WireBedrock: - c, b, cerr = cl.Bedrock(ctx, settings.Endpoint, proxyIP, pc.model, "Reply with exactly: pong", sessionID) + c, b, cerr = cl.Bedrock(ctx, settings.Endpoint, proxyIP, pc.model, matrixPrompt, sessionID) default: - c, b, cerr = cl.ChatPrefixed(ctx, settings.Endpoint, proxyIP, pc.pathPrefix, pc.kind, pc.model, "Reply with exactly: pong", sessionID) + c, b, cerr = cl.ChatPrefixed(ctx, settings.Endpoint, proxyIP, pc.pathPrefix, pc.kind, pc.model, matrixPrompt, sessionID) } if cerr == nil { code, body = c, b @@ -290,6 +451,7 @@ func TestProvidersMatrix(t *testing.T) { // The session id sent as x-session-id must round-trip into the // access-log row for this provider. + var row api.AgentNetworkAccessLog require.Eventually(t, func() bool { logs, lerr := srv.ListAccessLogs(ctx) if lerr != nil { @@ -297,11 +459,15 @@ func TestProvidersMatrix(t *testing.T) { } for _, r := range logs.Data { if r.SessionId != nil && *r.SessionId == sessionID { + row = r return true } } return false }, 30*time.Second, 2*time.Second, "session id %q must be recorded in an access-log row for %s", sessionID, pc.name) + + // Stored total and cache cost must match the published rates applied to the row's buckets. + validateAccessLogCost(t, pc, row) }) } @@ -322,4 +488,7 @@ func TestProvidersMatrix(t *testing.T) { } return false }, 60*time.Second, 3*time.Second, "consumption must be recorded with positive token counts after live traffic") + + // Final raw-SQL audit: bypass the API and re-verify every persisted usage row in the store. + verifyUsageRowsSQL(t, srv) } diff --git a/e2e/harness/client.go b/e2e/harness/client.go index 2ffcf653d..f53d0ea64 100644 --- a/e2e/harness/client.go +++ b/e2e/harness/client.go @@ -256,7 +256,7 @@ func (cl *Client) ChatPrefixed(ctx context.Context, endpoint, proxyIP, pathPrefi case WireMessages: path = "/v1/messages" headers = []string{"anthropic-version: 2023-06-01"} - body = fmt.Sprintf(`{"model":%q,"max_tokens":64,"messages":[{"role":"user","content":%q}]}`, model, prompt) + body = fmt.Sprintf(`{"model":%q,"max_tokens":2048,"messages":[{"role":"user","content":%q}]}`, model, prompt) default: path = "/v1/chat/completions" body = fmt.Sprintf(`{"model":%q,"messages":[{"role":"user","content":%q}]}`, model, prompt) @@ -271,7 +271,7 @@ func (cl *Client) ChatPrefixed(ctx context.Context, endpoint, proxyIP, pathPrefi // is sent as the universal x-session-id header the proxy records. func (cl *Client) Vertex(ctx context.Context, endpoint, proxyIP, project, region, model, prompt, sessionID string) (int, string, error) { path := fmt.Sprintf("/v1/projects/%s/locations/%s/publishers/anthropic/models/%s:rawPredict", project, region, model) - body := fmt.Sprintf(`{"anthropic_version":"vertex-2023-10-16","max_tokens":64,"messages":[{"role":"user","content":%q}]}`, prompt) + body := fmt.Sprintf(`{"anthropic_version":"vertex-2023-10-16","max_tokens":2048,"messages":[{"role":"user","content":%q}]}`, prompt) return cl.post(ctx, endpoint, proxyIP, path, body, withSessionID(nil, sessionID)) } @@ -282,7 +282,7 @@ func (cl *Client) Vertex(ctx context.Context, endpoint, proxyIP, project, region // header the proxy records. func (cl *Client) Bedrock(ctx context.Context, endpoint, proxyIP, model, prompt, sessionID string) (int, string, error) { path := "/model/" + model + "/invoke" - body := fmt.Sprintf(`{"anthropic_version":"bedrock-2023-05-31","max_tokens":64,"messages":[{"role":"user","content":%q}]}`, prompt) + body := fmt.Sprintf(`{"anthropic_version":"bedrock-2023-05-31","max_tokens":2048,"messages":[{"role":"user","content":%q}]}`, prompt) return cl.post(ctx, endpoint, proxyIP, path, body, withSessionID(nil, sessionID)) } @@ -341,7 +341,7 @@ func (cl *Client) Terminate(ctx context.Context) error { return cl.container.Terminate(ctx) } -// containerLogs reads up to 256 KiB of a container's logs for diagnostics. +// containerLogs reads up to 4 MiB of a container's logs for diagnostics — enough for a whole provider-matrix run. func containerLogs(ctx context.Context, c testcontainers.Container) string { if c == nil { return "" @@ -351,6 +351,6 @@ func containerLogs(ctx context.Context, c testcontainers.Container) string { return fmt.Sprintf("", err) } defer r.Close() - b, _ := io.ReadAll(io.LimitReader(r, 256<<10)) + b, _ := io.ReadAll(io.LimitReader(r, 4<<20)) return string(b) } diff --git a/e2e/harness/combined.go b/e2e/harness/combined.go index a6f43a139..5723100ca 100644 --- a/e2e/harness/combined.go +++ b/e2e/harness/combined.go @@ -221,6 +221,29 @@ func (c *Combined) CreateProxyTokenCLI(ctx context.Context, name string) (string return "", fmt.Errorf("token not found in CLI output: %s", string(out)) } +// SnapshotStoreDB copies the management sqlite store (with WAL/SHM sidecars) out of the bind-mounted +// data dir into dstDir and returns the copy's path; reading a copy avoids locking against live writes. +func (c *Combined) SnapshotStoreDB(dstDir string) (string, error) { + src := filepath.Join(c.workDir, "data", "store.db") + if _, err := os.Stat(src); err != nil { + return "", fmt.Errorf("management store not found at %s: %w", src, err) + } + dst := filepath.Join(dstDir, "store.db") + for _, suffix := range []string{"", "-wal", "-shm"} { + data, err := os.ReadFile(src + suffix) + if err != nil { + if os.IsNotExist(err) && suffix != "" { + continue // sidecar only exists in WAL mode + } + return "", fmt.Errorf("read %s: %w", src+suffix, err) + } + if err := os.WriteFile(dst+suffix, data, 0o600); err != nil { + return "", fmt.Errorf("write %s: %w", dst+suffix, err) + } + } + return dst, nil +} + // Logs returns the combined server container logs, for diagnostics. func (c *Combined) Logs(ctx context.Context) string { return containerLogs(ctx, c.container)