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

Replaces the pseudo-version pin (v1.49.2-0.20260719024505-24ff20a68f52, the
Bug-A iterator-leak fix only) with the released v1.49.2, which also carries the
Bug-B guard: org.Resolve skips the doomed GetById for a legacy all-digit cached
id (IAM Valkey's stale 1772587477 for 'hanzo') and resolves by name, so the
(*Query).ById legacy-numeric path that hot-looped in v1.801.95 is never taken.
Keeps ai v1.826.4 (in-proc TierReader). go.mod+go.sum only.

apps/wire_test.go

@@ -46,6 +46,7 @@ var frozen = []struct {
 	{"do", false, false},             // was order 123
 	{"platform", true, false},        // was order 124
 	{"projects", false, false},       // was order 125
+	{"dns", false, false},            // new: /v1/dns zone plane (after projects)
 	{"prompts", false, false},        // was order 126
 	{"agents", false, true},          // was order 127
 	{"link", false, true},            // new: unified AI login manager (/v1/links), after agents
@@ -77,6 +78,7 @@ var frozen = []struct {
 	{"graph", false, false},          // was order 135
 	{"security", true, true},         // was order 136
 	{"integrations", false, true},    // was order 137
+	{"cloudflare", false, false},     // new: /v1/cloudflare edge plane (after integrations)
 	{"sbom", true, false},            // was order 137
 	{"team", false, true},            // was order 138
 	{"settings", false, true},        // was order 138

build.go

@@ -176,6 +176,11 @@ func buildMeteringClient(cfg *Config, log luxlog.Logger) *metering.Client {
 		log.Error("billing: invalid commerce URL, gate disabled", "err", err)
 		m, _ = metering.New(metering.Config{})
 	}
+	// Observe every cap-check fail-open (timeout / slow / broken commerce) — a cap that
+	// silently allows must never be silent. The completion still proceeds (fail-open).
+	metering.OnCapError = func(err error) {
+		log.Warn("spend-cap check failed open (allowing completion) — commerce authorize slow/unavailable", "err", err)
+	}
 	if m.Enabled() {
 		log.Info("billing gate enabled", "commerce", boolStr(inProcess, "in-process", "http:"+base), "fail_open", cfg.BillingFailOpen)
 	} else {

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)
+	}
+}

clients/metering/hang_failopen_test.go

@@ -0,0 +1,44 @@
+package metering_test
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/hanzoai/cloud/clients/metering"
+)
+
+// SEV1 regression: a HANGING/hot-looping commerce authorize must NEVER hang the
+// completion path — the cap fails OPEN fast. A funded caller whose /authorize blocks
+// for 10s must still get an ALLOW verdict in ~capAuthorizeTimeout (well under the hang),
+// never a wait. This is the safety the missing timeout lacked.
+func TestAuthorizeVerdict_CapHang_FailsOpenFast(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.Contains(r.URL.Path, "/spend-alerts/authorize") {
+			time.Sleep(10 * time.Second) // simulate the legacy-org hot-loop / stuck handler
+			return
+		}
+		_, _ = io.WriteString(w, `{"available":100000}`) // balance: funded
+	}))
+	defer srv.Close()
+
+	c, _ := metering.New(metering.Config{BaseURL: srv.URL, Token: "t", Org: "hanzo"})
+
+	start := time.Now()
+	v, err := c.AuthorizeVerdict(context.Background(), metering.AuthInput{User: "hanzo", Org: "hanzo", AmountCents: 1})
+	elapsed := time.Since(start)
+
+	if err != nil {
+		t.Fatalf("a hanging cap check must FAIL-OPEN (nil err), got %v", err)
+	}
+	if !v.Allow {
+		t.Fatalf("a hanging cap check must ALLOW (fail-open), got %+v", v)
+	}
+	if elapsed > 4*time.Second {
+		t.Fatalf("cap check took %s against a 10s hang — the completion would HANG; must return in ~1.5s", elapsed)
+	}
+}

clients/metering/metering.go

@@ -76,6 +76,18 @@ const (
 // would silently debit the wrong tenant.
 const headerOrg = "X-Org-Id"
 
+// capAuthorizeTimeout HARD-bounds the per-scope spend-cap check. The cap is a POLICY
+// overlay, NEVER a gate on availability: a slow, broken, or hot-looping commerce
+// authorize must fail-open (allow) FAST, never hang the completion path (the SEV1 that
+// a legacy-org GetById hot-loop caused). Short enough that a healthy in-proc call
+// (sub-50ms) is unaffected, while a stuck one is abandoned and the request proceeds.
+const capAuthorizeTimeout = 1500 * time.Millisecond
+
+// OnCapError, when set, is called (best-effort) whenever the cap check FAILS OPEN — a
+// timeout or any error on the authorize call. It lets the host log/alert on a degraded
+// cap without this leaf package taking a logger dependency. nil = no-op.
+var OnCapError func(error)
+
 // headerTest opts a service-token call into commerce's TEST ledger
 // (org.Live=false): balances and debits hit the sandbox books, not real money.
 // See commerce/middleware/accesstoken.go (c.GetHeader("X-Hanzo-Test")). Sent
@@ -327,7 +339,10 @@ func (c *Client) AuthorizeVerdict(ctx context.Context, in AuthInput) (Verdict, e
 	// Funded — layer the per-scope spend cap. Fail-open on any cap error.
 	sv, serr := c.scopeAuthorize(ctx, in)
 	if serr != nil {
-		return Verdict{Allow: true}, nil
+		if OnCapError != nil {
+			OnCapError(serr) // observe the fail-open (timeout / broken commerce); never block.
+		}
+		return Verdict{Allow: true}, nil // fail-open: a cap-check failure NEVER blocks a completion.
 	}
 	if !sv.Allow && sv.Reason == "spend_cap" {
 		return Verdict{Allow: false, Reason: "spend_cap", CapCents: sv.CapCents, SpentCents: sv.SpentCents}, nil
@@ -393,9 +408,34 @@ func (c *Client) scopeAuthorize(ctx context.Context, in AuthInput) (scopeVerdict
 	}
 	q.Set("currency", currencyOr(in.Currency))
 
-	body, err := c.get(ctx, pathLimitsAuthorize, q, c.orgFor(in.Org))
-	if err != nil {
-		return scopeVerdict{}, err
+	// HARD BOUND (SEV1 safety): the cap check must NEVER hang the completion path. Run
+	// the authorize with a strict deadline AND a select-based hard timeout that returns
+	// to the caller even if the in-proc handler goroutine is STUCK (an unresponsive
+	// handler — e.g. a hot-loop — cannot be interrupted, so ctx cancellation alone would
+	// not unblock c.get). On timeout OR any error the caller (AuthorizeVerdict) fails
+	// open and ALLOWS the request; a stuck goroutine is abandoned (a leak bounded by the
+	// upstream hot-loop fix), never a wait.
+	tctx, cancel := context.WithTimeout(ctx, capAuthorizeTimeout)
+	defer cancel()
+	type getResult struct {
+		body []byte
+		err  error
+	}
+	done := make(chan getResult, 1)
+	go func() {
+		b, e := c.get(tctx, pathLimitsAuthorize, q, c.orgFor(in.Org))
+		done <- getResult{b, e}
+	}()
+
+	var body []byte
+	select {
+	case r := <-done:
+		if r.err != nil {
+			return scopeVerdict{}, r.err
+		}
+		body = r.body
+	case <-tctx.Done():
+		return scopeVerdict{}, fmt.Errorf("metering: cap authorize exceeded %s — failing open: %w", capAuthorizeTimeout, tctx.Err())
 	}
 	var v scopeVerdict
 	if err := json.Unmarshal(body, &v); err != nil {

clients/visor/http_test.go

@@ -26,7 +26,7 @@ type fakeVisor struct {
 	// (/v1/machines → ListComputeMachines) — the live droplet inventory that
 	// listMachines now unions with the registry.
 	liveByOwner map[string][]map[string]any
-	// nodesByOwner is the per-tenant DOKS worker NODES list (/v1/kubernetes-nodes →
+	// nodesByOwner is the per-tenant DOKS worker NODES list (/v1/k8s/nodes →
 	// ListComputeKubernetesNodes) — the THIRD machine source managedMachines unions.
 	nodesByOwner map[string][]map[string]any
 	poolsByOwner map[string][]map[string]any
@@ -53,7 +53,7 @@ func (f *fakeVisor) server(t *testing.T) *httptest.Server {
 		envelope200(w, f.liveByOwner[owner])
 	})
 	// DOKS worker nodes (ListComputeKubernetesNodes) — the third machine source.
-	mux.HandleFunc("/v1/kubernetes-nodes", func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc("/v1/k8s/nodes", func(w http.ResponseWriter, r *http.Request) {
 		owner := r.URL.Query().Get("owner")
 		f.lastOwner = owner
 		envelope200(w, f.nodesByOwner[owner])
@@ -261,7 +261,7 @@ func TestMachinesMergeLiveDOAndRegistry(t *testing.T) {
 }
 
 // TestMachinesMergeDOKSNodes proves the THIRD source: listMachines unions DOKS
-// worker NODES (/v1/kubernetes-nodes) with the live droplet list so a cluster's
+// worker NODES (/v1/k8s/nodes) with the live droplet list so a cluster's
 // nodes appear on the fleet — while a DOKS node whose droplet is ALSO in the live
 // list is deduped by droplet id (never listed twice). This is the visor backport's
 // payoff: cluster NODES show, not just standalone droplets.
@@ -269,7 +269,7 @@ func TestMachinesMergeDOKSNodes(t *testing.T) {
 	f := &fakeVisor{
 		// Live DO reseller list: one standalone droplet, plus a DOKS worker droplet
 		// (drop-node-1) that DO also surfaced under the org tag — so its droplet id
-		// appears in BOTH the live list and the kubernetes-nodes list.
+		// appears in BOTH the live list and the k8s-nodes list.
 		liveByOwner: map[string][]map[string]any{
 			"acme": {
 				{"owner": "acme", "name": "standalone-1", "id": "drop-1",

clients/visor/k8s.go

@@ -0,0 +1,249 @@
+// Copyright 2025 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.
+
+// k8s.go is the UNIFIED /v1/k8s surface — the ONE Kubernetes noun on api.hanzo.ai:
+// list clusters, one cluster's detail (node pools + worker nodes), DEPLOY (create)
+// and delete DOKS clusters, and the fleet-wide worker NODES. Every route is a thin,
+// tenant-scoped proxy to Visor (which OWNS the DigitalOcean lifecycle); this client
+// fabricates nothing — a cluster row is a real DOKS cluster, its nodes are real
+// droplets, and an honestly-absent field is omitted, never invented.
+//
+// Surface (org taken verbatim from the validated IAM owner claim, never a client
+// field, so a caller only ever sees or mutates its OWN tenant's clusters):
+//
+//	GET    /v1/k8s/clusters        list the org's DOKS clusters (+ BYO fold-in) -> {clusters:[clusterView]}
+//	GET    /v1/k8s/clusters/:id    one cluster's detail: pools + worker nodes   -> clusterDetailView (404 if absent)
+//	POST   /v1/k8s/clusters        provision a DOKS cluster    (ADMIN-GATED)     -> clusterView (201)
+//	DELETE /v1/k8s/clusters/:id    destroy a DOKS cluster      (ADMIN-GATED)     -> 204
+//	GET    /v1/k8s/nodes           every DOKS worker node as a machine          -> {nodes:[machineView]}
+//
+// READS are org-scoped (any validated member of the org). MUTATIONS (create/delete)
+// are admin-gated — a SuperAdmin (platform sudo) OR an OrgAdmin of the owning org —
+// because provisioning spends real infrastructure on Hanzo's house account. The gate
+// is the SAME principal predicate the rest of the cloud mutating surface uses.
+package visor
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/hanzoai/cloud"
+	"github.com/hanzoai/cloud/clients/principal"
+	"github.com/zap-proto/zip"
+)
+
+// ---- Visor wire structs (service.KubernetesCluster / KubernetesClusterDetail) ----
+
+// visorKubernetesCluster mirrors visor/service.KubernetesCluster.
+type visorKubernetesCluster struct {
+	ID         string   `json:"id"`
+	Name       string   `json:"name"`
+	RegionSlug string   `json:"regionSlug"`
+	Status     string   `json:"status"`
+	Tags       []string `json:"tags"`
+}
+
+// visorK8sNodePool mirrors visor/service.NodePool (the cluster-detail subset). Its
+// pool id arrives as `id` (not the `poolId` the verb node-pool surface emits), so it
+// is a distinct wire type mapped explicitly to nodePoolView.
+type visorK8sNodePool struct {
+	ID        string `json:"id"`
+	Name      string `json:"name"`
+	Size      string `json:"size"`
+	Count     int    `json:"count"`
+	MinNodes  int    `json:"minNodes"`
+	MaxNodes  int    `json:"maxNodes"`
+	AutoScale bool   `json:"autoScale"`
+}
+
+// visorKubernetesClusterDetail mirrors visor/service.KubernetesClusterDetail: the
+// cluster flattened, plus its node pools and worker nodes (as machines).
+type visorKubernetesClusterDetail struct {
+	visorKubernetesCluster
+	NodePools []visorK8sNodePool `json:"nodePools"`
+	Nodes     []visorMachine     `json:"nodes"`
+}
+
+// clusterDetailView is the GET .../clusters/:id shape: the cluster (with its pools)
+// plus the worker nodes as the SAME machineView the machines surface emits.
+type clusterDetailView struct {
+	clusterView
+	Nodes []machineView `json:"nodes"`
+}
+
+// k8sClusterView maps a Visor cluster to the console clusterView. Kind is "managed"
+// (DOKS-provisioned); pool detail is carried by the DETAIL endpoint, so the list row
+// leaves NodePools empty (honest — the list is lightweight, not a fabricated 0-pool).
+func k8sClusterView(kc visorKubernetesCluster) clusterView {
+	return clusterView{
+		DoksClusterID: kc.ID,
+		DoClusterID:   kc.ID,
+		Name:          kc.Name,
+		Region:        kc.RegionSlug,
+		Status:        firstNonEmpty(kc.Status, "unknown"),
+		Kind:          "managed",
+		NodePools:     []nodePoolView{},
+	}
+}
+
+// requireClusterAdmin is the mutation gate: platform SuperAdmin OR an OrgAdmin of the
+// caller's own org. Returns a 403 error when neither holds (the caller returns it).
+func requireClusterAdmin(c *zip.Ctx) error {
+	if principal.IsSuperAdmin(c) || principal.IsOrgAdmin(c) {
+		return nil
+	}
+	return zip.ErrForbidden("admin required: DOKS cluster provisioning is admin-gated")
+}
+
+// ---- handlers ----
+
+// listK8sClusters lists the org's DOKS clusters (Visor, house account) folded with
+// the org's BYO clusters — ONE fleet cluster view under the unified k8s noun. A Visor
+// outage is logged and skipped so a down optional provider never hides the BYO list.
+func listK8sClusters(s *cloud.Service[state], c *zip.Ctx) error {
+	org, ok := tenant(c)
+	if !ok {
+		return zip.ErrForbidden("X-Org-Id required")
+	}
+	var clusters []visorKubernetesCluster
+	if err := s.State.cl.call(c, http.MethodGet, "/v1/k8s/clusters", q("owner", org), nil, &clusters); err != nil {
+		s.Log.Warn("visor k8s clusters failed; returning BYO-only cluster list", "org", org, "err", err)
+		clusters = nil
+	}
+	out := make([]clusterView, 0, len(clusters))
+	for _, kc := range clusters {
+		out = append(out, k8sClusterView(kc))
+	}
+	out = append(out, byoClusters(s, org, project(c))...)
+	return c.JSON(http.StatusOK, map[string]any{"clusters": out})
+}
+
+// getK8sCluster returns one cluster's detail: node pools + worker nodes. Visor scopes
+// the lookup to the org (a foreign or missing id resolves to not-found), so a tenant
+// can never read another tenant's cluster by guessing an id.
+func getK8sCluster(s *cloud.Service[state], c *zip.Ctx) error {
+	org, ok := tenant(c)
+	if !ok {
+		return zip.ErrForbidden("X-Org-Id required")
+	}
+	id := strings.TrimSpace(c.Param("id"))
+	if id == "" {
+		return zip.ErrBadRequest("cluster id required")
+	}
+	var d visorKubernetesClusterDetail
+	if err := s.State.cl.call(c, http.MethodGet, "/v1/k8s/clusters/"+id, q("owner", org), nil, &d); err != nil {
+		return err
+	}
+	if d.ID == "" && d.Name == "" {
+		return zip.ErrNotFound("cluster not found")
+	}
+	view := clusterDetailView{clusterView: k8sClusterView(d.visorKubernetesCluster)}
+	view.NodePools = make([]nodePoolView, 0, len(d.NodePools))
+	for _, p := range d.NodePools {
+		view.NodePools = append(view.NodePools, nodePoolView{
+			PoolID: firstNonEmpty(p.ID, p.Name), Name: p.Name, Size: p.Size,
+			Count: p.Count, MinNodes: p.MinNodes, MaxNodes: p.MaxNodes, AutoScale: p.AutoScale,
+		})
+		view.NodeCount += p.Count
+		if view.NodeSize == "" {
+			view.NodeSize = p.Size
+		}
+	}
+	view.Nodes = make([]machineView, 0, len(d.Nodes))
+	for _, m := range d.Nodes {
+		view.Nodes = append(view.Nodes, toMachineView(m))
+	}
+	return c.JSON(http.StatusOK, view)
+}
+
+// createClusterReq is the provision body: identity, placement, version and ONE seed
+// node pool. It IS Visor's CreateClusterSpec shape, so it forwards without re-mapping.
+type createClusterReq struct {
+	Name     string `json:"name"`
+	Region   string `json:"region"`
+	Version  string `json:"version,omitempty"`
+	NodePool struct {
+		Name  string `json:"name,omitempty"`
+		Size  string `json:"size"`
+		Count int    `json:"count"`
+	} `json:"nodePool"`
+}
+
+// createK8sCluster provisions a DOKS cluster for the caller's org. ADMIN-GATED: real
+// infrastructure spend on the house account. Validated at this boundary, then Visor
+// owns provisioning + the hanzo-org ownership tag.
+func createK8sCluster(s *cloud.Service[state], c *zip.Ctx) error {
+	org, ok := tenant(c)
+	if !ok {
+		return zip.ErrForbidden("X-Org-Id required")
+	}
+	if err := requireClusterAdmin(c); err != nil {
+		return err
+	}
+	var body createClusterReq
+	if err := c.Bind(&body); err != nil {
+		return err
+	}
+	body.Name = strings.TrimSpace(body.Name)
+	body.Region = strings.TrimSpace(body.Region)
+	body.NodePool.Size = strings.TrimSpace(body.NodePool.Size)
+	if body.Name == "" {
+		return zip.ErrBadRequest("'name' is required")
+	}
+	if body.Region == "" {
+		return zip.ErrBadRequest("'region' is required")
+	}
+	if body.NodePool.Size == "" {
+		return zip.ErrBadRequest("'nodePool.size' is required")
+	}
+	if body.NodePool.Count < 1 {
+		return zip.ErrBadRequest("'nodePool.count' must be at least 1")
+	}
+	var kc visorKubernetesCluster
+	if err := s.State.cl.call(c, http.MethodPost, "/v1/k8s/clusters", q("owner", org), body, &kc); err != nil {
+		return err
+	}
+	return c.JSON(http.StatusCreated, k8sClusterView(kc))
+}
+
+// deleteK8sCluster destroys a DOKS cluster by id. ADMIN-GATED, like create. Visor
+// scopes the delete to the org (refuses a foreign id), so this can only ever remove
+// the caller org's own cluster.
+func deleteK8sCluster(s *cloud.Service[state], c *zip.Ctx) error {
+	org, ok := tenant(c)
+	if !ok {
+		return zip.ErrForbidden("X-Org-Id required")
+	}
+	if err := requireClusterAdmin(c); err != nil {
+		return err
+	}
+	id := strings.TrimSpace(c.Param("id"))
+	if id == "" {
+		return zip.ErrBadRequest("cluster id required")
+	}
+	if err := s.State.cl.call(c, http.MethodDelete, "/v1/k8s/clusters/"+id, q("owner", org), nil, nil); err != nil {
+		return err
+	}
+	return c.NoContent(http.StatusNoContent)
+}
+
+// listK8sNodes returns every DOKS worker node in the org's clusters as a machine —
+// the SAME set the fleet folds in (managedMachines), exposed directly under the k8s
+// noun. House account (hanzo-org cluster tag) + BYOC, deduped by Visor.
+func listK8sNodes(s *cloud.Service[state], c *zip.Ctx) error {
+	org, ok := tenant(c)
+	if !ok {
+		return zip.ErrForbidden("X-Org-Id required")
+	}
+	var nodes []visorMachine
+	if err := s.State.cl.call(c, http.MethodGet, "/v1/k8s/nodes", q("owner", org), nil, &nodes); err != nil {
+		return err
+	}
+	out := make([]machineView, 0, len(nodes))
+	for _, m := range nodes {
+		out = append(out, toMachineView(m))
+	}
+	return c.JSON(http.StatusOK, map[string]any{"nodes": out})
+}

clients/visor/k8s_test.go

@@ -0,0 +1,284 @@
+// Copyright 2025 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.
+
+package visor
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/hanzoai/cloud"
+	luxlog "github.com/luxfi/log"
+	"github.com/zap-proto/zip"
+)
+
+// k8sFake is a stand-in for Visor's /v1/k8s surface. It speaks the casibase
+// {status,msg,data} envelope, scopes every read by ?owner (proving cloud forwards
+// the VALIDATED principal's org), and records the last owner + any mutation it saw
+// so a test can assert an admin-gated call is REFUSED before it ever reaches Visor.
+type k8sFake struct {
+	lastOwner   string
+	createdBody map[string]any
+	deletedID   string
+}
+
+func (f *k8sFake) server(t *testing.T) *httptest.Server {
+	t.Helper()
+	mux := http.NewServeMux()
+	// list (GET) + create (POST) on the bare /clusters literal.
+	mux.HandleFunc("/v1/k8s/clusters", func(w http.ResponseWriter, r *http.Request) {
+		f.lastOwner = r.URL.Query().Get("owner")
+		switch r.Method {
+		case http.MethodGet:
+			var out []map[string]any
+			if f.lastOwner == "acme" {
+				out = []map[string]any{{
+					"id": "cl-1", "name": "prod", "regionSlug": "sfo3", "status": "running",
+					"tags": []string{"managed-by:hanzo-visor", "hanzo-org:acme"},
+				}}
+			}
+			envelope200(w, out)
+		case http.MethodPost:
+			_ = json.NewDecoder(r.Body).Decode(&f.createdBody)
+			envelope200(w, map[string]any{
+				"id": "cl-new", "name": f.createdBody["name"], "regionSlug": f.createdBody["region"],
+				"status": "provisioning", "tags": []string{"hanzo-org:" + f.lastOwner},
+			})
+		default:
+			w.WriteHeader(http.StatusMethodNotAllowed)
+		}
+	})
+	// detail (GET) + delete (DELETE) on /clusters/{id}.
+	mux.HandleFunc("/v1/k8s/clusters/", func(w http.ResponseWriter, r *http.Request) {
+		f.lastOwner = r.URL.Query().Get("owner")
+		id := strings.TrimPrefix(r.URL.Path, "/v1/k8s/clusters/")
+		switch r.Method {
+		case http.MethodGet:
+			envelope200(w, map[string]any{
+				"id": id, "name": "prod", "regionSlug": "sfo3", "status": "running",
+				"tags": []string{"hanzo-org:acme"},
+				"nodePools": []map[string]any{{
+					"id": "p1", "name": "workers", "size": "s-4vcpu-8gb", "count": 3,
+				}},
+				"nodes": []map[string]any{{
+					"owner": "acme", "name": "worker-1", "id": "555",
+					"provider": "DigitalOcean", "size": "s-4vcpu-8gb", "region": "sfo3", "state": "running",
+				}},
+			})
+		case http.MethodDelete:
+			f.deletedID = id
+			envelope200(w, true)
+		default:
+			w.WriteHeader(http.StatusMethodNotAllowed)
+		}
+	})
+	mux.HandleFunc("/v1/k8s/nodes", func(w http.ResponseWriter, r *http.Request) {
+		f.lastOwner = r.URL.Query().Get("owner")
+		var out []map[string]any
+		if f.lastOwner == "acme" {
+			out = []map[string]any{{
+				"owner": "acme", "name": "worker-1", "id": "555",
+				"provider": "DigitalOcean", "size": "s-4vcpu-8gb", "region": "sfo3",
+				"state": "running", "tag": "doks-cluster:prod",
+			}}
+		}
+		envelope200(w, out)
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func mountK8s(t *testing.T, f *k8sFake) *zip.App {
+	t.Helper()
+	srv := f.server(t)
+	t.Setenv("VISOR_URL", srv.URL)
+	t.Setenv("VISOR_CLIENT_ID", "")     // force the bearer-forward path (fake ignores auth)
+	t.Setenv("VISOR_CLIENT_SECRET", "") //
+	app := zip.New(zip.Config{Logger: luxlog.New("test")})
+	if err := Mount(app, cloud.Deps{Logger: luxlog.New("test")}); err != nil {
+		t.Fatalf("Mount: %v", err)
+	}
+	return app
+}
+
+// reqK8s issues a request as a validated principal (X-Org-Id + X-User-Id). admin=true
+// adds the OrgAdmin bit the mutation gate checks — so the same helper drives both the
+// allowed and the refused mutation paths.
+func reqK8s(t *testing.T, app *zip.App, method, path, org string, admin bool, body any) (int, []byte) {
+	t.Helper()
+	var r io.Reader
+	if body != nil {
+		b, _ := json.Marshal(body)
+		r = bytes.NewReader(b)
+	}
+	req := httptest.NewRequest(method, path, r)
+	if body != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	if org != "" {
+		req.Header.Set("X-Org-Id", org)
+		req.Header.Set("X-User-Id", "u-"+org)
+	}
+	if admin {
+		req.Header.Set("X-User-IsOrgAdmin", "true")
+	}
+	resp, err := app.Fiber().Test(req)
+	if err != nil {
+		t.Fatalf("Test %s %s: %v", method, path, err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	b, _ := io.ReadAll(resp.Body)
+	return resp.StatusCode, b
+}
+
+// GET /v1/k8s/clusters is org-scoped: cloud forwards the validated org, and maps
+// Visor clusters to the managed clusterView. No org → 403 (never reaches Visor).
+func TestK8sClustersListTenantScoped(t *testing.T) {
+	f := &k8sFake{}
+	app := mountK8s(t, f)
+
+	if code, _ := reqK8s(t, app, http.MethodGet, "/v1/k8s/clusters", "", false, nil); code != http.StatusForbidden {
+		t.Fatalf("no-org list want 403, got %d", code)
+	}
+
+	code, body := reqK8s(t, app, http.MethodGet, "/v1/k8s/clusters", "acme", false, nil)
+	if code != http.StatusOK {
+		t.Fatalf("list want 200, got %d (%s)", code, body)
+	}
+	if f.lastOwner != "acme" {
+		t.Fatalf("cloud must forward owner=acme, got %q", f.lastOwner)
+	}
+	var out struct {
+		Clusters []clusterView `json:"clusters"`
+	}
+	if err := json.Unmarshal(body, &out); err != nil {
+		t.Fatalf("shape: %v (%s)", err, body)
+	}
+	if len(out.Clusters) != 1 {
+		t.Fatalf("acme want 1 cluster, got %d", len(out.Clusters))
+	}
+	cl := out.Clusters[0]
+	if cl.DoksClusterID != "cl-1" || cl.Name != "prod" || cl.Region != "sfo3" || cl.Status != "running" || cl.Kind != "managed" {
+		t.Fatalf("cluster view mismatch: %+v", cl)
+	}
+}
+
+// GET /v1/k8s/clusters/:id returns detail: node pools (poolId mapped from Visor's
+// id) with the derived node count, and the worker nodes as machineViews.
+func TestK8sClusterDetail(t *testing.T) {
+	f := &k8sFake{}
+	app := mountK8s(t, f)
+
+	code, body := reqK8s(t, app, http.MethodGet, "/v1/k8s/clusters/cl-1", "acme", false, nil)
+	if code != http.StatusOK {
+		t.Fatalf("detail want 200, got %d (%s)", code, body)
+	}
+	var d clusterDetailView
+	if err := json.Unmarshal(body, &d); err != nil {
+		t.Fatalf("shape: %v (%s)", err, body)
+	}
+	if d.DoksClusterID != "cl-1" || len(d.NodePools) != 1 || d.NodePools[0].PoolID != "p1" || d.NodePools[0].Count != 3 {
+		t.Fatalf("detail pools mismatch: %+v", d)
+	}
+	if d.NodeCount != 3 || d.NodeSize != "s-4vcpu-8gb" {
+		t.Fatalf("derived node count/size wrong: count=%d size=%q", d.NodeCount, d.NodeSize)
+	}
+	if len(d.Nodes) != 1 || d.Nodes[0].ID != "worker-1" || d.Nodes[0].Status != "running" {
+		t.Fatalf("detail nodes mismatch: %+v", d.Nodes)
+	}
+}
+
+// GET /v1/k8s/nodes returns every worker node as a machineView, org-scoped.
+func TestK8sNodesTenantScoped(t *testing.T) {
+	f := &k8sFake{}
+	app := mountK8s(t, f)
+
+	if code, _ := reqK8s(t, app, http.MethodGet, "/v1/k8s/nodes", "", false, nil); code != http.StatusForbidden {
+		t.Fatalf("no-org nodes want 403, got %d", code)
+	}
+	code, body := reqK8s(t, app, http.MethodGet, "/v1/k8s/nodes", "acme", false, nil)
+	if code != http.StatusOK {
+		t.Fatalf("nodes want 200, got %d (%s)", code, body)
+	}
+	var out struct {
+		Nodes []machineView `json:"nodes"`
+	}
+	if err := json.Unmarshal(body, &out); err != nil {
+		t.Fatalf("shape: %v (%s)", err, body)
+	}
+	if len(out.Nodes) != 1 || out.Nodes[0].ID != "worker-1" || out.Nodes[0].Type != "s-4vcpu-8gb" {
+		t.Fatalf("node view mismatch: %+v", out.Nodes)
+	}
+}
+
+// POST /v1/k8s/clusters is ADMIN-GATED: a non-admin is refused BEFORE the request
+// reaches Visor; an admin with a valid body provisions and gets 201; a bad body is a
+// 400 at the cloud boundary.
+func TestK8sCreateClusterAdminGated(t *testing.T) {
+	f := &k8sFake{}
+	app := mountK8s(t, f)
+	valid := map[string]any{"name": "prod", "region": "sfo3", "version": "latest",
+		"nodePool": map[string]any{"size": "s-4vcpu-8gb", "count": 2}}
+
+	// No validated principal → 403.
+	if code, _ := reqK8s(t, app, http.MethodPost, "/v1/k8s/clusters", "", false, valid); code != http.StatusForbidden {
+		t.Fatalf("no-org create want 403, got %d", code)
+	}
+	// Validated but NON-admin → 403, and the mutation never reached Visor.
+	if code, _ := reqK8s(t, app, http.MethodPost, "/v1/k8s/clusters", "acme", false, valid); code != http.StatusForbidden {
+		t.Fatalf("non-admin create want 403, got %d", code)
+	}
+	if f.createdBody != nil {
+		t.Fatalf("non-admin create must NOT reach Visor, but body was received: %+v", f.createdBody)
+	}
+	// Admin + valid body → 201, forwarded to Visor with the spec intact.
+	code, body := reqK8s(t, app, http.MethodPost, "/v1/k8s/clusters", "acme", true, valid)
+	if code != http.StatusCreated {
+		t.Fatalf("admin create want 201, got %d (%s)", code, body)
+	}
+	if f.createdBody == nil || f.createdBody["name"] != "prod" || f.createdBody["region"] != "sfo3" {
+		t.Fatalf("create spec not forwarded to Visor: %+v", f.createdBody)
+	}
+	var mv clusterView
+	if err := json.Unmarshal(body, &mv); err != nil {
+		t.Fatalf("shape: %v (%s)", err, body)
+	}
+	if mv.Name != "prod" || mv.Status != "provisioning" || mv.Kind != "managed" {
+		t.Fatalf("created cluster view mismatch: %+v", mv)
+	}
+	// Admin + invalid body (no node pool size) → 400 at the boundary.
+	if code, _ := reqK8s(t, app, http.MethodPost, "/v1/k8s/clusters", "acme", true,
+		map[string]any{"name": "x", "region": "sfo3"}); code != http.StatusBadRequest {
+		t.Fatalf("admin create without pool size want 400, got %d", code)
+	}
+}
+
+// DELETE /v1/k8s/clusters/:id is ADMIN-GATED, like create.
+func TestK8sDeleteClusterAdminGated(t *testing.T) {
+	f := &k8sFake{}
+	app := mountK8s(t, f)
+
+	if code, _ := reqK8s(t, app, http.MethodDelete, "/v1/k8s/clusters/cl-1", "", false, nil); code != http.StatusForbidden {
+		t.Fatalf("no-org delete want 403, got %d", code)
+	}
+	if code, _ := reqK8s(t, app, http.MethodDelete, "/v1/k8s/clusters/cl-1", "acme", false, nil); code != http.StatusForbidden {
+		t.Fatalf("non-admin delete want 403, got %d", code)
+	}
+	if f.deletedID != "" {
+		t.Fatalf("non-admin delete must NOT reach Visor, but id %q was deleted", f.deletedID)
+	}
+	if code, _ := reqK8s(t, app, http.MethodDelete, "/v1/k8s/clusters/cl-1", "acme", true, nil); code != http.StatusNoContent {
+		t.Fatalf("admin delete want 204, got %d", code)
+	}
+	if f.deletedID != "cl-1" {
+		t.Fatalf("admin delete must forward id cl-1 to Visor, got %q", f.deletedID)
+	}
+}

clients/visor/visor.go

@@ -114,6 +114,17 @@ func Mount(app *zip.App, deps cloud.Deps) error {
 	app.Post("/v1/clusters/:clusterId/pools/:poolId/scale", cloud.Handle(s, scalePool))
 	app.Delete("/v1/clusters/:clusterId/pools/:poolId", cloud.Handle(s, deletePool))
 
+	// Unified /v1/k8s — the ONE Kubernetes noun (k8s.go): DOKS cluster lifecycle
+	// (list / detail+nodes / create / delete) plus the fleet-wide worker NODES,
+	// proxied to Visor. Reads are org-scoped; create/delete are admin-gated (real
+	// house-account infra spend). Static /clusters registers before its :id sibling
+	// so a cluster id never captures the literal.
+	app.Get("/v1/k8s/clusters", cloud.Handle(s, listK8sClusters))
+	app.Post("/v1/k8s/clusters", cloud.Handle(s, createK8sCluster))
+	app.Get("/v1/k8s/clusters/:id", cloud.Handle(s, getK8sCluster))
+	app.Delete("/v1/k8s/clusters/:id", cloud.Handle(s, deleteK8sCluster))
+	app.Get("/v1/k8s/nodes", cloud.Handle(s, listK8sNodes))
+
 	// Compute catalog: the global region + size lists that back the Machines/GPUs
 	// launch drawer. Namespaced under /v1/compute (visor's domain) — "sizes"/"regions"
 	// are catalog dimensions shared by machines AND gpus, not owned nouns, so they
@@ -168,7 +179,7 @@ func project(c *zip.Ctx) string { return principal.Project(c) }
 //   - LIVE DigitalOcean reseller list (GET /v1/machines → ListComputeMachines →
 //     service.ListOrgMachines): every droplet currently tagged to the org in
 //     Hanzo's house DO account, straight from the live DO API.
-//   - DOKS worker NODES (GET /v1/kubernetes-nodes → ListComputeKubernetesNodes):
+//   - DOKS worker NODES (GET /v1/k8s/nodes → ListComputeKubernetesNodes):
 //     each managed-Kubernetes node as a Machine (Id=droplet id), unioned across the
 //     house account (hanzo-org cluster tag) and BYOC providers (Provider.ClusterID)
 //     so the fleet shows cluster NODES, not just standalone droplets.
@@ -194,14 +205,14 @@ func managedMachines(s *cloud.Service[state], c *zip.Ctx, org string) []visorMac
 		s.Log.Warn("visor list-compute-machines failed; live DO machines omitted", "org", org, "err", err)
 		live = nil
 	}
-	// THIRD source: DOKS worker NODES (GET /v1/kubernetes-nodes → visor unions the
+	// THIRD source: DOKS worker NODES (GET /v1/k8s/nodes → visor unions the
 	// house-account hanzo-org-tagged clusters + BYOC Provider.ClusterID clusters).
 	// A cluster's node is a real droplet, so the world fleet should show the NODES,
 	// not just standalone droplets. A DOKS node whose droplet is ALSO in the live
 	// list dedupes by droplet id below (Machine.Id == DropletID), so it never lists
 	// twice. Independently resilient like the other two.
-	if err := s.State.cl.call(c, http.MethodGet, "/v1/kubernetes-nodes", q("owner", org), nil, &nodes); err != nil {
-		s.Log.Warn("visor kubernetes-nodes failed; DOKS nodes omitted", "org", org, "err", err)
+	if err := s.State.cl.call(c, http.MethodGet, "/v1/k8s/nodes", q("owner", org), nil, &nodes); err != nil {
+		s.Log.Warn("visor k8s nodes failed; DOKS nodes omitted", "org", org, "err", err)
 		nodes = nil
 	}
 

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/google/go-containerregistry v0.21.7
 	github.com/google/go-github/v52 v52.0.0
 	github.com/hanzoai/account v0.2.0
-	github.com/hanzoai/commerce v1.49.2-0.20260718193926-c59b46959b14
+	github.com/hanzoai/commerce v1.49.2
 	github.com/hanzoai/decimal v0.1.1
 	github.com/hanzoai/go-openai v1.41.0
 	github.com/hanzoai/goa v1.0.0
@@ -841,7 +841,7 @@ require (
 	github.com/hanzo-ds/go v1.0.1
 	github.com/hanzo-ds/native v0.72.0 // indirect
 	github.com/hanzoai/agent v0.1.3
-	github.com/hanzoai/ai v1.826.2
+	github.com/hanzoai/ai v1.826.4
 	github.com/hanzoai/authz v1.10.7
 	github.com/hanzoai/base v1.5.7
 	github.com/hanzoai/licensing v0.1.5

go.sum

@@ -1212,10 +1212,8 @@ github.com/hanzoai/account v0.2.0 h1:WxIut3YMz8JNdHerlRIqP1a+k3P79BIKM1mjfRFDG0Y
 github.com/hanzoai/account v0.2.0/go.mod h1:8OzIGRphAhlabOI74O4GoL3RM0y8mbUV0pQUKgXLjkw=
 github.com/hanzoai/agent v0.1.3 h1:zzV4t8kN/m/wTLrqzEy0fxxONSZbx3XSVH7TIR9gZNU=
 github.com/hanzoai/agent v0.1.3/go.mod h1:Z3hCBdSeN/nGV4o+3F4psQ2bbFk17+tMP5l+G2ssNNA=
-github.com/hanzoai/ai v1.826.0 h1:kt33dpYopDODksj65N2eWo4OmKNOajU52UVfpsttZHY=
-github.com/hanzoai/ai v1.826.0/go.mod h1:LkSrjXJjFS9weIQmhXl53x/Dmt90KgkulEkx8O1Gd5U=
-github.com/hanzoai/ai v1.826.2 h1:aR9ILhNJEjH/M10JlNX5KFqsk/2aSjepGDfSeR5KVJ4=
-github.com/hanzoai/ai v1.826.2/go.mod h1:LkSrjXJjFS9weIQmhXl53x/Dmt90KgkulEkx8O1Gd5U=
+github.com/hanzoai/ai v1.826.4 h1:k4kooLThqCHSfzqMy1y1UdZ5YtgwS3oPElNg8VfYkKc=
+github.com/hanzoai/ai v1.826.4/go.mod h1:LkSrjXJjFS9weIQmhXl53x/Dmt90KgkulEkx8O1Gd5U=
 github.com/hanzoai/authz v1.10.7 h1:JrHljH29mbmVi8u6/6EVG7R0NiFhIYYm2WUBBuBmFq0=
 github.com/hanzoai/authz v1.10.7/go.mod h1:9wf6n6BvrvxRULUtL3yc+vmprwwvAGZLGJoMHhiQSK4=
 github.com/hanzoai/authzstore v0.1.1 h1:4GsvB+bKs+gFtfKDMoYq/C7KxJAHrXLwtwdSutTakbo=
@@ -1228,8 +1226,8 @@ github.com/hanzoai/builder v0.3.13 h1:tAOJ+0Q0xrrovk7lkvaZxuKZ4lqENIB6tE0Rr9+6Bo
 github.com/hanzoai/builder v0.3.13/go.mod h1:TWZaiP0Y9tCMwtLH2EvQqBAeT1f3aJI5Y0XPM8S0wcE=
 github.com/hanzoai/captable v1.0.0 h1:utXPsOaPL+QV0rJaoNDFHvWFv7etf2kfm1bsFh6JRD0=
 github.com/hanzoai/captable v1.0.0/go.mod h1:czdMnzEvb8FWNKCzmh+z8PmwYeQp4MGyTPgo+Q5O+mk=
-github.com/hanzoai/commerce v1.49.2-0.20260718193926-c59b46959b14 h1:pMld5NtpzN7CDj0Kxx0a/tUtQyQwLHIO28juX2YXZH4=
-github.com/hanzoai/commerce v1.49.2-0.20260718193926-c59b46959b14/go.mod h1:+Bp6hVxJQmv71D/ZPzs5qiEVHyDNUchNsqakd+778ps=
+github.com/hanzoai/commerce v1.49.2 h1:zIuXaCz6CvjiF0/z/9ub+y0dkKTaonewNpmRHroVimg=
+github.com/hanzoai/commerce v1.49.2/go.mod h1:+Bp6hVxJQmv71D/ZPzs5qiEVHyDNUchNsqakd+778ps=
 github.com/hanzoai/csqlite v0.1.0 h1:suwC3dh0INlfP/U0Es6cDf6JNQ+2+GVLLATPWCUux6k=
 github.com/hanzoai/csqlite v0.1.0/go.mod h1:H31a/O6VXuklR9UBkgY++bmAK5uzVfXPqU0F6P9Wsos=
 github.com/hanzoai/dashscope-go-sdk v0.0.2 h1:L/FlStjXeehrNSBqU8y/ecSG/MnaAJqfyfjmLYmY344=