admin(aimetrics): /v1/admin/aimetrics AI-metrics read view (Langfuse totals/latency/models + usage + eval scores) + table-canonical SQL tests

clients/admin/admin.go

@@ -103,6 +103,7 @@ func routes(app *zip.App, s *cloud.Service[core.State]) {
 	app.Get("/v1/admin/products", core.Guard(s, products))
 	app.Get("/v1/admin/compute", core.Guard(s, compute))
 	app.Get("/v1/admin/o11y", core.Guard(s, o11y))
+	app.Get("/v1/admin/aimetrics", core.Guard(s, aimetrics))
 	app.Post("/v1/admin/sync", core.Guard(s, syncNow))
 
 	// Product analytics — org-scoped (SuperAdmin: all-orgs; org admin: their own org).

clients/admin/aimetrics.go

@@ -0,0 +1,392 @@
+// Copyright 2023-2026 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package admin
+
+// aimetrics — GET /v1/admin/aimetrics, the GLOBAL fleet-wide AI / training / eval
+// read that powers the operator's AI-metrics board on admin.hanzo.ai. It is the
+// AI-and-eval-focused companion to o11y (o11y.go): where o11y answers "how is the
+// FLEET behaving" (RED metrics, logs, usage), this answers "how are the MODELS and
+// EVALS doing" — LLM generations, per-model spend, and eval-run quality/progress —
+// over the SAME ONE datastore (Datastore), the SAME shared client
+// (aiobject.DatastoreQuery), no second connection.
+//
+// Signals, each from its canonical table in the one datastore:
+//   - LLM generations → langfuse.observations : generations, cost (USD), latency
+//                                                (fleet-wide; honest-empty until the
+//                                                Langfuse ingest lands rows)
+//   - Per-model usage → hanzo.cloud_usage      : requests, tokens, cost per model
+//                                                (the live usage ledger the ai gateway
+//                                                writes — populated today)
+//   - Eval runs       → hanzo.eval_traces      : traces, runs, datasets, models under
+//                                                test, per-trace latency
+//   - Eval progress   → hanzo.eval_scores      : score count, avg score, per-score-name
+//                                                distribution, recent-run averages, and
+//                                                the avg-score-over-time TREND — the
+//                                                training/eval progress signal
+//
+// The eval_traces / eval_scores tables are OWNED and written by the eval telemetry
+// store (clients/eval/telemetry.go) — the SAME warehouse, same db ("hanzo"), same
+// shared aiobject client. admin only READS them here. There is deliberately no
+// "training_progress" table: the router's per-request training events live in the ai
+// OLTP Postgres (object.RoutingEvent), NOT the OLAP warehouse, so the honest
+// warehouse-side progress signal is the eval-score trend, not a routing table.
+//
+// SUPERADMIN ONLY (the core.Guard wrap in admin.go), all-orgs, no org filter — the
+// one place a fleet operator crosses tenants for AI/eval metrics; a non-admin bearer
+// is refused 403 before a single row is read. Fail-closed.
+//
+// Honest by construction, exactly like o11y/compute: no datastore connected → the
+// real empty aggregate, never a fabricated fleet; and every signal degrades
+// INDEPENDENTLY — a table that is absent or a column that differs contributes its
+// zero-value (the enclosing `if err == nil`), never a failure, so the board always
+// renders what the datastore actually holds. admin READS only; it owns and creates
+// NO table. Money from cloud_usage is USD cents, from langfuse is USD; latency is
+// milliseconds; time bounds are POSITIONAL parameters (never interpolated), and the
+// bucket interval is a server-side constant — injection-safe.
+
+import (
+	"strconv"
+	"time"
+
+	aiobject "github.com/hanzoai/ai/object"
+	"github.com/hanzoai/cloud"
+	"github.com/hanzoai/cloud/clients/admin/core"
+	"github.com/zap-proto/zip"
+)
+
+// Fully-qualified datastore tables. admin only READS these — the ai gateway owns
+// hanzo.cloud_usage, Langfuse owns langfuse.observations, and the eval telemetry
+// store (clients/eval) owns hanzo.eval_traces / hanzo.eval_scores.
+const (
+	aimUsageTable  = "hanzo.cloud_usage"
+	aimLangfuseObs = "langfuse.observations"
+	aimEvalTraces  = "hanzo.eval_traces"
+	aimEvalScores  = "hanzo.eval_scores"
+	aimTopN        = 12
+)
+
+// aiMetrics is the whole AI-metrics board payload.
+type aiMetrics struct {
+	Range          string           `json:"range"`
+	Start          string           `json:"start"`
+	End            string           `json:"end"`
+	Langfuse       aimLangfuse      `json:"langfuse"`
+	Usage          aimUsage         `json:"usage"`
+	Evals          aimEvals         `json:"evals"`
+	TopModels      []aimModelStat   `json:"topModels"`      // cloud_usage per-model (populated today)
+	LangfuseModels []aimLfModelStat `json:"langfuseModels"` // langfuse per-model (honest-empty today)
+	ScoreNames     []aimScoreStat   `json:"scoreNames"`     // eval_scores per score-name
+	EvalRuns       []aimRunStat     `json:"evalRuns"`       // recent eval runs (progress)
+	ScoreSeries    []aimScorePoint  `json:"scoreSeries"`    // avg eval score over time (progress trend)
+}
+
+// aimLangfuse is the fleet-wide Langfuse generation rollup (honest-empty today).
+// Cost is USD (Langfuse's native unit); latency is milliseconds (end_time-start_time).
+type aimLangfuse struct {
+	Generations  int64   `json:"generations"`
+	CostUsd      float64 `json:"costUsd"`
+	LatencyMsAvg float64 `json:"latencyMsAvg"`
+	LatencyMsP95 float64 `json:"latencyMsP95"`
+}
+
+// aimUsage is the fleet LLM-usage KPI band from the live cloud_usage ledger.
+type aimUsage struct {
+	Requests         int64 `json:"requests"`
+	Tokens           int64 `json:"tokens"`
+	PromptTokens     int64 `json:"promptTokens"`
+	CompletionTokens int64 `json:"completionTokens"`
+	CostCents        int64 `json:"costCents"`
+	Models           int64 `json:"models"`
+}
+
+// aimEvals is the fleet eval KPI band: the trace half (eval_traces) and the score
+// half (eval_scores). LatencyMsAvg is the mean model-under-test call window.
+type aimEvals struct {
+	Runs         int64   `json:"runs"`
+	Traces       int64   `json:"traces"`
+	Datasets     int64   `json:"datasets"`
+	Models       int64   `json:"models"`
+	LatencyMsAvg float64 `json:"latencyMsAvg"`
+	Scores       int64   `json:"scores"`
+	ScoreNames   int64   `json:"scoreNames"`
+	AvgScore     float64 `json:"avgScore"`
+}
+
+// aimModelStat is one row of the per-model usage leaderboard (cloud_usage).
+type aimModelStat struct {
+	Model     string `json:"model"`
+	Requests  int64  `json:"requests"`
+	Tokens    int64  `json:"tokens"`
+	CostCents int64  `json:"costCents"`
+}
+
+// aimLfModelStat is one row of the per-model Langfuse leaderboard (honest-empty today).
+type aimLfModelStat struct {
+	Model       string  `json:"model"`
+	Generations int64   `json:"generations"`
+	CostUsd     float64 `json:"costUsd"`
+}
+
+// aimScoreStat is one row of the per-score-name eval leaderboard (eval_scores).
+type aimScoreStat struct {
+	Name     string  `json:"name"`
+	Count    int64   `json:"count"`
+	AvgValue float64 `json:"avgValue"`
+	MinValue float64 `json:"minValue"`
+	MaxValue float64 `json:"maxValue"`
+}
+
+// aimRunStat is one recent eval run: its dataset, how many scores it recorded, its
+// mean score, and when it last ran — the run-level eval-progress row.
+type aimRunStat struct {
+	RunName  string  `json:"runName"`
+	Dataset  string  `json:"dataset"`
+	Scores   int64   `json:"scores"`
+	AvgValue float64 `json:"avgValue"`
+	LastTs   string  `json:"lastTs"`
+}
+
+// aimScorePoint is one bucket of the avg-eval-score-over-time trend.
+type aimScorePoint struct {
+	Ts       string  `json:"ts"`
+	AvgValue float64 `json:"avgValue"`
+	Count    int64   `json:"count"`
+}
+
+// aimetrics answers GET /v1/admin/aimetrics. ?range=24h|7d|30d bounds the window
+// (default 30d). SUPERADMIN ONLY (core.Guard). Every signal degrades independently:
+// a table that is absent or errors contributes its zero-value, never a failure — the
+// board always renders what the datastore actually holds.
+func aimetrics(s *cloud.Service[core.State], c *zip.Ctx) error {
+	ctx := c.Context()
+	rangeLabel := o11yRange(c.Query("range"))
+	since := computeSince(rangeLabel)
+	payload := aiMetrics{
+		Range:          rangeLabel,
+		Start:          since.Format(time.RFC3339),
+		End:            time.Now().UTC().Format(time.RFC3339),
+		TopModels:      []aimModelStat{},
+		LangfuseModels: []aimLfModelStat{},
+		ScoreNames:     []aimScoreStat{},
+		EvalRuns:       []aimRunStat{},
+		ScoreSeries:    []aimScorePoint{},
+	}
+
+	// Honest-empty when the warehouse is not connected: the board renders its zero
+	// state, never a fabricated fleet.
+	if !aiobject.DatastoreEnabled() {
+		return core.OK(c, payload)
+	}
+
+	sinceTS := chTS(since) // DateTime literal — cloud_usage.timestamp, langfuse.start_time, eval_*.ts
+	interval := o11yBucket(rangeLabel)
+
+	// ── Langfuse generations (fleet) — honest-empty until ingest lands rows ──
+	if rows, err := aiobject.DatastoreQuery(ctx, aimLangfuseTotalsSQL(), sinceTS); err == nil {
+		r := firstRowOr(rows)
+		payload.Langfuse.Generations = chInt64(r["gens"])
+		payload.Langfuse.CostUsd = chFloat64(r["cost"])
+	}
+	// Langfuse latency (separate query so a Nullable end_time / column mismatch never
+	// zeroes the proven generations+cost number above).
+	if rows, err := aiobject.DatastoreQuery(ctx, aimLangfuseLatencySQL(), sinceTS); err == nil {
+		r := firstRowOr(rows)
+		payload.Langfuse.LatencyMsAvg = chFloat64(r["lat_avg"])
+		payload.Langfuse.LatencyMsP95 = chFloat64(r["lat_p95"])
+	}
+	// Langfuse per-model.
+	if rows, err := aiobject.DatastoreQuery(ctx, aimLangfuseModelsSQL(), sinceTS); err == nil {
+		payload.LangfuseModels = lfModelsFromRows(rows)
+	}
+
+	// ── Per-model usage (fleet) from the live cloud_usage ledger ──
+	if rows, err := aiobject.DatastoreQuery(ctx, aimUsageTotalsSQL(), sinceTS); err == nil {
+		fillAimUsage(&payload.Usage, firstRowOr(rows))
+	}
+	if rows, err := aiobject.DatastoreQuery(ctx, aimTopModelsSQL(), sinceTS); err == nil {
+		payload.TopModels = aimModelsFromRows(rows)
+	}
+
+	// ── Evals (fleet): traces + scores + progress ──
+	if rows, err := aiobject.DatastoreQuery(ctx, aimEvalTracesSQL(), sinceTS); err == nil {
+		fillAimEvalTraces(&payload.Evals, firstRowOr(rows))
+	}
+	if rows, err := aiobject.DatastoreQuery(ctx, aimEvalScoresSQL(), sinceTS); err == nil {
+		fillAimEvalScores(&payload.Evals, firstRowOr(rows))
+	}
+	if rows, err := aiobject.DatastoreQuery(ctx, aimScoreNamesSQL(), sinceTS); err == nil {
+		payload.ScoreNames = scoreNamesFromRows(rows)
+	}
+	if rows, err := aiobject.DatastoreQuery(ctx, aimEvalRunsSQL(), sinceTS); err == nil {
+		payload.EvalRuns = evalRunsFromRows(rows)
+	}
+	if rows, err := aiobject.DatastoreQuery(ctx, aimScoreSeriesSQL(interval), sinceTS); err == nil {
+		payload.ScoreSeries = scoreSeriesFromRows(rows)
+	}
+
+	return core.OK(c, payload)
+}
+
+// ── pure SQL builders (static SQL + one positional time bound; unit-tested) ──
+
+func aimLangfuseTotalsSQL() string {
+	return "SELECT count() AS gens, toFloat64(sum(total_cost)) AS cost FROM " + aimLangfuseObs +
+		" WHERE type = 'GENERATION' AND start_time >= ?"
+}
+
+func aimLangfuseLatencySQL() string {
+	lat := "(toUnixTimestamp64Milli(end_time) - toUnixTimestamp64Milli(start_time))"
+	return "SELECT round(avg(" + lat + "), 2) AS lat_avg, round(quantile(0.95)(" + lat + "), 2) AS lat_p95 " +
+		"FROM " + aimLangfuseObs + " WHERE type = 'GENERATION' AND start_time >= ? AND end_time > start_time"
+}
+
+func aimLangfuseModelsSQL() string {
+	return "SELECT provided_model_name AS model, count() AS gens, toFloat64(sum(total_cost)) AS cost " +
+		"FROM " + aimLangfuseObs + " WHERE type = 'GENERATION' AND start_time >= ? AND provided_model_name != '' " +
+		"GROUP BY model ORDER BY gens DESC LIMIT " + strconv.Itoa(aimTopN)
+}
+
+func aimUsageTotalsSQL() string {
+	return "SELECT count() AS requests, sum(total_tokens) AS tokens, " +
+		"sum(prompt_tokens) AS prompt_tokens, sum(completion_tokens) AS completion_tokens, " +
+		"sum(cost_cents) AS cost_cents, uniqExact(model) AS models " +
+		"FROM " + aimUsageTable + " WHERE timestamp >= ?"
+}
+
+func aimTopModelsSQL() string {
+	return "SELECT model, count() AS requests, sum(total_tokens) AS tokens, " +
+		"sum(cost_cents) AS cost_cents FROM " + aimUsageTable +
+		" WHERE timestamp >= ? AND model != '' GROUP BY model ORDER BY requests DESC LIMIT " + strconv.Itoa(aimTopN)
+}
+
+func aimEvalTracesSQL() string {
+	lat := "(toUnixTimestamp64Milli(end_time) - toUnixTimestamp64Milli(start_time))"
+	return "SELECT count() AS traces, uniqExact(run_name) AS runs, uniqExact(dataset) AS datasets, " +
+		"uniqExact(model) AS models, round(avgIf(" + lat + ", end_time > start_time), 2) AS lat_avg " +
+		"FROM " + aimEvalTraces + " WHERE ts >= ?"
+}
+
+func aimEvalScoresSQL() string {
+	return "SELECT count() AS scores, round(avg(value), 4) AS avg_value, uniqExact(name) AS score_names " +
+		"FROM " + aimEvalScores + " WHERE ts >= ?"
+}
+
+func aimScoreNamesSQL() string {
+	return "SELECT name, count() AS n, round(avg(value), 4) AS avg_value, " +
+		"round(min(value), 4) AS min_value, round(max(value), 4) AS max_value " +
+		"FROM " + aimEvalScores + " WHERE ts >= ? AND name != '' GROUP BY name ORDER BY n DESC LIMIT " + strconv.Itoa(aimTopN)
+}
+
+func aimEvalRunsSQL() string {
+	return "SELECT run_name, any(dataset) AS dataset, count() AS scores, round(avg(value), 4) AS avg_value, " +
+		"max(ts) AS last_ts FROM " + aimEvalScores + " WHERE ts >= ? AND run_name != '' " +
+		"GROUP BY run_name ORDER BY last_ts DESC LIMIT " + strconv.Itoa(aimTopN)
+}
+
+func aimScoreSeriesSQL(interval string) string {
+	return "SELECT toStartOfInterval(ts, INTERVAL " + interval + ") AS ts, " +
+		"round(avg(value), 4) AS avg_value, count() AS n FROM " + aimEvalScores +
+		" WHERE ts >= ? GROUP BY ts ORDER BY ts"
+}
+
+// ── pure row parsers (unit-tested) ──
+
+func fillAimUsage(u *aimUsage, r map[string]any) {
+	u.Requests = chInt64(r["requests"])
+	u.Tokens = chInt64(r["tokens"])
+	u.PromptTokens = chInt64(r["prompt_tokens"])
+	u.CompletionTokens = chInt64(r["completion_tokens"])
+	u.CostCents = chInt64(r["cost_cents"])
+	u.Models = chInt64(r["models"])
+}
+
+func fillAimEvalTraces(e *aimEvals, r map[string]any) {
+	e.Traces = chInt64(r["traces"])
+	e.Runs = chInt64(r["runs"])
+	e.Datasets = chInt64(r["datasets"])
+	e.Models = chInt64(r["models"])
+	e.LatencyMsAvg = chFloat64(r["lat_avg"])
+}
+
+func fillAimEvalScores(e *aimEvals, r map[string]any) {
+	e.Scores = chInt64(r["scores"])
+	e.AvgScore = chFloat64(r["avg_value"])
+	e.ScoreNames = chInt64(r["score_names"])
+}
+
+func aimModelsFromRows(rows []map[string]any) []aimModelStat {
+	out := make([]aimModelStat, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, aimModelStat{
+			Model:     chStr(r["model"]),
+			Requests:  chInt64(r["requests"]),
+			Tokens:    chInt64(r["tokens"]),
+			CostCents: chInt64(r["cost_cents"]),
+		})
+	}
+	return out
+}
+
+func lfModelsFromRows(rows []map[string]any) []aimLfModelStat {
+	out := make([]aimLfModelStat, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, aimLfModelStat{
+			Model:       chStr(r["model"]),
+			Generations: chInt64(r["gens"]),
+			CostUsd:     chFloat64(r["cost"]),
+		})
+	}
+	return out
+}
+
+func scoreNamesFromRows(rows []map[string]any) []aimScoreStat {
+	out := make([]aimScoreStat, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, aimScoreStat{
+			Name:     chStr(r["name"]),
+			Count:    chInt64(r["n"]),
+			AvgValue: chFloat64(r["avg_value"]),
+			MinValue: chFloat64(r["min_value"]),
+			MaxValue: chFloat64(r["max_value"]),
+		})
+	}
+	return out
+}
+
+func evalRunsFromRows(rows []map[string]any) []aimRunStat {
+	out := make([]aimRunStat, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, aimRunStat{
+			RunName:  chStr(r["run_name"]),
+			Dataset:  chStr(r["dataset"]),
+			Scores:   chInt64(r["scores"]),
+			AvgValue: chFloat64(r["avg_value"]),
+			LastTs:   chTime(r["last_ts"]),
+		})
+	}
+	return out
+}
+
+func scoreSeriesFromRows(rows []map[string]any) []aimScorePoint {
+	out := make([]aimScorePoint, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, aimScorePoint{
+			Ts:       chTime(r["ts"]),
+			AvgValue: chFloat64(r["avg_value"]),
+			Count:    chInt64(r["n"]),
+		})
+	}
+	return out
+}

clients/admin/aimetrics_test.go

@@ -0,0 +1,171 @@
+// Copyright 2023-2026 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package admin
+
+import (
+	"strings"
+	"testing"
+)
+
+// TestAimSQL_ReadsCanonicalTables proves every AI-metrics query reads the ONE
+// datastore's canonical table, binds the time bound as a POSITIONAL param (one
+// `?`), and never interpolates user input. The bucket interval is the only rendered
+// value in the series query and it is a server-side constant.
+func TestAimSQL_ReadsCanonicalTables(t *testing.T) {
+	cases := []struct {
+		name, sql, table string
+		wantQMarks       int
+	}{
+		{"langfuseTotals", aimLangfuseTotalsSQL(), "langfuse.observations", 1},
+		{"langfuseLatency", aimLangfuseLatencySQL(), "langfuse.observations", 1},
+		{"langfuseModels", aimLangfuseModelsSQL(), "langfuse.observations", 1},
+		{"usageTotals", aimUsageTotalsSQL(), "hanzo.cloud_usage", 1},
+		{"topModels", aimTopModelsSQL(), "hanzo.cloud_usage", 1},
+		{"evalTraces", aimEvalTracesSQL(), "hanzo.eval_traces", 1},
+		{"evalScores", aimEvalScoresSQL(), "hanzo.eval_scores", 1},
+		{"scoreNames", aimScoreNamesSQL(), "hanzo.eval_scores", 1},
+		{"evalRuns", aimEvalRunsSQL(), "hanzo.eval_scores", 1},
+		{"scoreSeries", aimScoreSeriesSQL("1 DAY"), "hanzo.eval_scores", 1},
+	}
+	for _, c := range cases {
+		if !strings.Contains(c.sql, "FROM "+c.table) {
+			t.Errorf("%s must read %s; got %q", c.name, c.table, c.sql)
+		}
+		if n := strings.Count(c.sql, "?"); n != c.wantQMarks {
+			t.Errorf("%s: %d bind params, want %d (time bound only) — no interpolation; got %q", c.name, n, c.wantQMarks, c.sql)
+		}
+	}
+}
+
+// TestAimLangfuseScopedToGeneration proves the Langfuse lens is scoped to
+// generations only (not spans/events), matching the o11y LLM lens.
+func TestAimLangfuseScopedToGeneration(t *testing.T) {
+	for _, sql := range []string{aimLangfuseTotalsSQL(), aimLangfuseLatencySQL(), aimLangfuseModelsSQL()} {
+		if !strings.Contains(sql, "type = 'GENERATION'") {
+			t.Errorf("langfuse lens must scope to GENERATION observations; got %q", sql)
+		}
+	}
+}
+
+// TestAimTop_LimitAndOrder proves the leaderboards bound + order the result.
+func TestAimTop_LimitAndOrder(t *testing.T) {
+	if !strings.Contains(aimTopModelsSQL(), "ORDER BY requests DESC LIMIT 12") {
+		t.Errorf("topModels must order by requests desc, limit %d", aimTopN)
+	}
+	if !strings.Contains(aimScoreNamesSQL(), "GROUP BY name ORDER BY n DESC LIMIT 12") {
+		t.Errorf("scoreNames must group+order+limit %d", aimTopN)
+	}
+	if !strings.Contains(aimEvalRunsSQL(), "ORDER BY last_ts DESC LIMIT 12") {
+		t.Errorf("evalRuns must order by last_ts desc, limit %d", aimTopN)
+	}
+}
+
+// TestAimScoreSeries_IntervalBound proves the (constant) bucket interval is
+// rendered into the score-trend series query and grouped/ordered by the bucket.
+func TestAimScoreSeries_IntervalBound(t *testing.T) {
+	for _, iv := range []string{"1 HOUR", "6 HOUR", "1 DAY"} {
+		s := aimScoreSeriesSQL(iv)
+		if !strings.Contains(s, "INTERVAL "+iv) || !strings.Contains(s, "GROUP BY ts ORDER BY ts") {
+			t.Errorf("score series must bucket by INTERVAL %s; got %q", iv, s)
+		}
+	}
+}
+
+// TestAimEvalLatencyGuarded proves the latency expressions guard end_time>start_time
+// so a zero/default end_time never contributes a garbage (negative) latency.
+func TestAimEvalLatencyGuarded(t *testing.T) {
+	if !strings.Contains(aimEvalTracesSQL(), "end_time > start_time") {
+		t.Errorf("eval traces latency must guard end_time>start_time; got %q", aimEvalTracesSQL())
+	}
+	if !strings.Contains(aimLangfuseLatencySQL(), "end_time > start_time") {
+		t.Errorf("langfuse latency must guard end_time>start_time; got %q", aimLangfuseLatencySQL())
+	}
+}
+
+// TestFillAimUsage reads a cloud_usage row into the KPI band across the numeric
+// variants the driver returns (uint64/int64/float64), honest zeros on an empty row.
+func TestFillAimUsage(t *testing.T) {
+	var empty aimUsage
+	fillAimUsage(&empty, map[string]any{})
+	if empty.Requests != 0 || empty.Tokens != 0 || empty.Models != 0 {
+		t.Fatalf("empty row must yield honest zeros; got %+v", empty)
+	}
+	var got aimUsage
+	fillAimUsage(&got, map[string]any{
+		"requests": uint64(274), "tokens": uint64(102597), "prompt_tokens": uint64(60000),
+		"completion_tokens": uint64(42597), "cost_cents": uint64(216), "models": uint64(42),
+	})
+	if got.Requests != 274 || got.Tokens != 102597 || got.CostCents != 216 || got.Models != 42 {
+		t.Fatalf("usage totals mis-parsed: %+v", got)
+	}
+}
+
+// TestFillAimEvals maps both eval halves (traces + scores) into the KPI band,
+// including the float latency/score columns (round()/avg() land as float64; a
+// Decimal-as-string is parsed).
+func TestFillAimEvals(t *testing.T) {
+	var e aimEvals
+	fillAimEvalTraces(&e, map[string]any{
+		"traces": uint64(1280), "runs": uint64(16), "datasets": uint64(4),
+		"models": uint64(6), "lat_avg": float64(842.5),
+	})
+	fillAimEvalScores(&e, map[string]any{
+		"scores": uint64(1280), "avg_value": "0.8125", "score_names": uint64(3),
+	})
+	if e.Traces != 1280 || e.Runs != 16 || e.Datasets != 4 || e.Models != 6 || e.LatencyMsAvg != 842.5 {
+		t.Fatalf("eval traces mis-parsed: %+v", e)
+	}
+	if e.Scores != 1280 || e.ScoreNames != 3 || e.AvgScore != 0.8125 { // string→float64 path
+		t.Fatalf("eval scores mis-parsed: %+v", e)
+	}
+}
+
+// TestAimParsers map datastore rows into the view-models and preserve order (the
+// SQL already ORDER BYs; a parser must not reorder or drop rows), with empty input
+// yielding an empty (non-nil) slice rather than a panic.
+func TestAimParsers(t *testing.T) {
+	models := aimModelsFromRows([]map[string]any{
+		{"model": "glm-5.2", "requests": uint64(154), "tokens": uint64(38966), "cost_cents": uint64(114)},
+		{"model": "deepseek-v4-flash", "requests": uint64(118), "tokens": uint64(61550), "cost_cents": uint64(101)},
+	})
+	if len(models) != 2 || models[0].Model != "glm-5.2" || models[1].Model != "deepseek-v4-flash" || models[0].Requests != 154 {
+		t.Fatalf("top models mis-parsed/reordered: %+v", models)
+	}
+	lf := lfModelsFromRows([]map[string]any{
+		{"model": "gpt-4o", "gens": uint64(42), "cost": float64(1.25)},
+	})
+	if len(lf) != 1 || lf[0].Model != "gpt-4o" || lf[0].Generations != 42 || lf[0].CostUsd != 1.25 {
+		t.Fatalf("langfuse models mis-parsed: %+v", lf)
+	}
+	names := scoreNamesFromRows([]map[string]any{
+		{"name": "accuracy", "n": uint64(320), "avg_value": float64(0.82), "min_value": float64(0), "max_value": float64(1)},
+	})
+	if len(names) != 1 || names[0].Name != "accuracy" || names[0].Count != 320 || names[0].AvgValue != 0.82 || names[0].MaxValue != 1 {
+		t.Fatalf("score names mis-parsed: %+v", names)
+	}
+	runs := evalRunsFromRows([]map[string]any{
+		{"run_name": "nightly-2026-07", "dataset": "gsm8k", "scores": uint64(200), "avg_value": float64(0.9), "last_ts": nil},
+	})
+	if len(runs) != 1 || runs[0].RunName != "nightly-2026-07" || runs[0].Dataset != "gsm8k" || runs[0].Scores != 200 || runs[0].AvgValue != 0.9 {
+		t.Fatalf("eval runs mis-parsed: %+v", runs)
+	}
+	// Empty input → empty (non-nil) slices, never a panic.
+	if got := scoreSeriesFromRows(nil); got == nil || len(got) != 0 {
+		t.Errorf("nil rows must yield empty slice, got %v", got)
+	}
+	if got := aimModelsFromRows(nil); got == nil || len(got) != 0 {
+		t.Errorf("nil rows must yield empty slice, got %v", got)
+	}
+}