refactor(api): follow the ai surface to /v1/ai/<resource>

The namespaces I invented (rag, chat, content, compute, work, ops, auth) are
replaced by the ONE service namespace the canonical spec in hanzoai/openapi
mandates: /v1/<service>/<resource>, and the service is `ai`.

/v1/chat was not just stuttering — the gateway routes GET|POST /v1/chat/{path}
to chat.hanzo.svc, so those calls would have reached a DIFFERENT SERVICE.

Co-authored-by: Hanzo Dev <dev@hanzo.ai>

account_principal.go

@@ -15,8 +15,8 @@ import (
 //
 // THE COMPLECTION IT REMOVES. The operator SPA authenticates via /v1/signin (a cloud
 // PKCE session → the X-User-* principal the middleware mints) but read its IDENTITY
-// from /v1/get-account, which the embedded IAM (casibase) answers from ITS OWN session
-// cookie. A PKCE session is not a casibase session, so get-account returned
+// from the account read, which the embedded IAM (casibase) answers from ITS OWN session
+// cookie. A PKCE session is not a casibase session, so that read returned
 // owner:"hanzo" (anonymous) or "Unauthorized operation" — and the SPA's SuperAdmin
 // gate (owner == "admin" && isAdmin), reading that, bounced the operator UI to login
 // even though the SAME session got 200 from every /v1/admin/* route. Two session
@@ -24,16 +24,24 @@ import (
 //
 // THE DECOMPLECTION. Identity is now the principal: when a VALIDATED principal is
 // present (X-User-Id is set ONLY by IdentityMiddleware from a real credential, never a
-// raw client header — see middleware_identity.go), /v1/get-account reflects it. owner
+// raw client header — see middleware_identity.go), /v1/ai/account reflects it. owner
 // is the HOME org (principal.Owner) so a SuperAdmin org-switched into a tenant stays a
 // SuperAdmin; isAdmin is the validated bit. With NO principal it falls through
 // (c.Next()) to the casibase account surface unchanged — the anonymous sign-in page
 // and any legacy casibase-session caller are untouched. One truth, additive, fail-open
 // to the old path. MUST be registered AFTER IdentityMiddleware (needs the minted
-// headers) and BEFORE MountAll (so it precedes the casibase /v1/get-account handler).
+// headers) and BEFORE MountAll (so it precedes the casibase account handler).
+// accountPath is the account read this middleware fronts. It is a named constant
+// because the interception is a PATH MATCH: when the /v1 surface was namespaced
+// (/v1/get-account → /v1/ai/account) a literal left un-updated here would not
+// error — the middleware would simply stop firing, fall through to the casibase
+// account surface, and hand the SPA the anonymous owner again. That is precisely
+// the bug this file exists to fix, silently restored.
+const accountPath = "/v1/ai/account"
+
 func AccountFromPrincipal() zip.Handler {
 	return func(c *zip.Ctx) error {
-		if c.Method() != http.MethodGet || c.Path() != "/v1/get-account" {
+		if c.Method() != http.MethodGet || c.Path() != accountPath {
 			return c.Next()
 		}
 		user := c.User() // X-User-Id — minted only from a validated credential

cli/cli.go

@@ -105,8 +105,8 @@ type Config struct {
 	PlatformURL string `json:"platform_url,omitempty"`
 	CloudURL    string `json:"cloud_url,omitempty"`
 	ClientID    string `json:"client_id,omitempty"`
-	APIKey      string `json:"apiKey,omitempty"`   // hk-… key; what `hanzo code` hands the agents
-	CodeTool    string `json:"code_tool,omitempty"` // default agent for bare `hanzo` / `hanzo code`: dev|claude|codex
+	APIKey      string `json:"apiKey,omitempty"`     // hk-… key; what `hanzo code` hands the agents
+	CodeTool    string `json:"code_tool,omitempty"`  // default agent for bare `hanzo` / `hanzo code`: dev|claude|codex
 	CodeModel   string `json:"code_model,omitempty"` // default model for `hanzo code` (else defaultCodeModel)
 }
 

cli/code.go

@@ -144,19 +144,19 @@ func openaiWire(base, token, _ string) map[string]string {
 }
 
 type codeAgent struct {
-	bin          string                     // executable to exec
-	wire         wire                       // how it finds the cloud
-	fullAuto     []string                   // flags that bypass approval prompts
-	continueArgs []string                   // harness-native form of Hanzo -c/--continue
-	modelArg     []string                   // how the model is passed on argv (empty: via env)
-	carrier      func(model string) string  // maps the resolved model to a client-recognized id (claude: zen→carrier); nil = pass through
+	bin          string                            // executable to exec
+	wire         wire                              // how it finds the cloud
+	fullAuto     []string                          // flags that bypass approval prompts
+	continueArgs []string                          // harness-native form of Hanzo -c/--continue
+	modelArg     []string                          // how the model is passed on argv (empty: via env)
+	carrier      func(model string) string         // maps the resolved model to a client-recognized id (claude: zen→carrier); nil = pass through
 	provider     func(base, model string) []string // agents that need the endpoint declared, not just env'd
-	clear        []string                   // env that would shadow the wire (a stale key in the shell)
-	configHome   string                     // env var that relocates the agent's config dir to ~/.hanzo ("" = share the user's own install)
-	seed         func(dir string) error     // one-time defaults for the isolated config dir
-	appendSystem []string                   // --append-system-prompt + text; ALWAYS applied (identity, not a permission bypass — present in --safe too)
-	mcp          bool                       // auto-wire the Hanzo MCP server (code/vector/web/vision tools) as an stdio server scoped to the cwd
-	install      string                     // hint when the binary is missing
+	clear        []string                          // env that would shadow the wire (a stale key in the shell)
+	configHome   string                            // env var that relocates the agent's config dir to ~/.hanzo ("" = share the user's own install)
+	seed         func(dir string) error            // one-time defaults for the isolated config dir
+	appendSystem []string                          // --append-system-prompt + text; ALWAYS applied (identity, not a permission bypass — present in --safe too)
+	mcp          bool                              // auto-wire the Hanzo MCP server (code/vector/web/vision tools) as an stdio server scoped to the cwd
+	install      string                            // hint when the binary is missing
 }
 
 // codeContextWindow is the input context (tokens) the served coding model

cli/gpu.go

@@ -18,7 +18,7 @@ package cli
 // --serve-engine adds the `engine.serve` capability: the worker probes a local
 // hanzo-engine (the OpenAI + Anthropic model server on :1234), advertises its model
 // endpoint in the presence record, and prints (or with --register-provider, POSTs)
-// the /v1/add-provider call that routes api.hanzo.ai model traffic to this GPU. One
+// the POST /v1/ai/providers call that routes api.hanzo.ai model traffic to this GPU. One
 // fleet, two job types: engine.serve (model serving) alongside studio.render.
 
 import (
@@ -935,7 +935,7 @@ type connectOpts struct {
 	serveEngine      bool
 	engineURL        string // local URL to probe hanzo-engine
 	engineEndpoint   string // public URL to advertise (defaults to engineURL)
-	registerProvider bool   // auto POST /v1/add-provider for the engine
+	registerProvider bool   // auto POST /v1/ai/providers for the engine
 	studioDir        string // local Studio checkout to launch + supervise on :8188
 	studioURL        string // studio base the render mirror uploads finished images to
 	mirror           bool   // sweep local renders into the org studio library (default on)
@@ -2228,7 +2228,7 @@ func describeEngine(adv *engineAdvertisement) string {
 	return fmt.Sprintf("ready · %d models", n)
 }
 
-// providerBody is the POST /v1/add-provider payload registering this node's engine
+// providerBody is the POST /v1/ai/providers payload registering this node's engine
 // as an org model provider. hanzo-engine is OpenAI-compatible, so Type=Local: the
 // gateway speaks the OpenAI wire format to it and auto-appends /v1 to providerUrl.
 func (w *worker) providerBody() map[string]any {
@@ -2260,23 +2260,23 @@ func (w *worker) printEngineHint(out io.Writer) {
 	fmt.Fprintf(out, "serving hanzo-engine (OpenAI + Anthropic) at %s — %s\n", adv.URL, describeEngine(adv))
 	body, _ := json.Marshal(w.providerBody())
 	fmt.Fprintln(out, "  route api.hanzo.ai model calls to this GPU by registering it as an org provider:")
-	fmt.Fprintf(out, "    curl -sS %s/v1/add-provider -H \"Authorization: Bearer $HANZO_TOKEN\" \\\n", w.baseURL)
+	fmt.Fprintf(out, "    curl -sS %s/v1/ai/providers -H \"Authorization: Bearer $HANZO_TOKEN\" \\\n", w.baseURL)
 	fmt.Fprintf(out, "      -H 'Content-Type: application/json' -d '%s'\n", body)
 	fmt.Fprintln(out, "  (or pass --register-provider. The endpoint must be reachable from api.hanzo.ai —")
-	fmt.Fprintln(out, "   a cloud GPU is in-cluster; a BYO node needs a public URL/tunnel. add-provider needs a platform-admin token today.)")
+	fmt.Fprintln(out, "   a cloud GPU is in-cluster; a BYO node needs a public URL/tunnel. registering a provider needs a platform-admin token today.)")
 }
 
-// registerProvider POSTs /v1/add-provider so the gateway routes model calls to this
+// registerProvider POSTs /v1/ai/providers so the gateway routes model calls to this
 // node's engine. Requires the engine to be reachable and (today) a platform-admin
 // token; both failures are reported clearly rather than swallowed.
 func (w *worker) registerProvider(ctx context.Context, adv *engineAdvertisement) error {
 	if adv == nil || adv.Status != "ready" {
 		return fmt.Errorf("engine not ready at %s — start hanzo-engine, then retry", w.engineURL)
 	}
-	code, err := w.call(ctx, http.MethodPost, "/v1/add-provider", w.providerBody(), nil)
+	code, err := w.call(ctx, http.MethodPost, "/v1/ai/providers", w.providerBody(), nil)
 	if err != nil {
 		if code == http.StatusForbidden {
-			return fmt.Errorf("add-provider is gated to a platform-admin token today; register from the console or with an admin token: %w", err)
+			return fmt.Errorf("registering a provider is gated to a platform-admin token today; register from the console or with an admin token: %w", err)
 		}
 		return err
 	}

cli/link.go

@@ -72,7 +72,7 @@ func newLinkCmd(envOf func() *Env, _ *globalFlags) *cobra.Command {
 	f.BoolVar(&opts.serveEngine, "serve-engine", false, "also advertise a hanzo-engine model server (OpenAI + Anthropic) running on this node")
 	f.StringVar(&opts.engineURL, "engine-url", defaultEngineURL, "local URL where hanzo-engine is probed (GET /v1/models)")
 	f.StringVar(&opts.engineEndpoint, "engine-endpoint", "", "public URL to advertise for gateway routing (defaults to --engine-url; a node behind NAT needs a reachable URL/tunnel)")
-	f.BoolVar(&opts.registerProvider, "register-provider", false, "auto-register the engine endpoint as an org model provider (POST /v1/add-provider)")
+	f.BoolVar(&opts.registerProvider, "register-provider", false, "auto-register the engine endpoint as an org model provider (POST /v1/ai/providers)")
 	f.StringVar(&opts.studioDir, "studio-dir", os.Getenv("HANZO_STUDIO_DIR"), "local Hanzo Studio checkout; when set, link launches and supervises the render backend on 127.0.0.1:8188")
 	f.StringVar(&opts.studioURL, "studio-url", firstNonEmpty(os.Getenv("HANZO_STUDIO_UPLOAD_URL"), defaultStudioUploadURL), "studio base URL the render mirror uploads finished images to (POST /v1/library/upload)")
 	f.BoolVar(&opts.mirror, "mirror", true, "sweep local renders into the org studio library; --mirror=false serves jobs only")

clients/admin/admin.go

@@ -39,6 +39,7 @@ import (
 	"github.com/hanzoai/cloud/clients/admin/finance"
 	"github.com/hanzoai/cloud/clients/admin/health"
 	"github.com/hanzoai/cloud/clients/admin/iam"
+	"github.com/hanzoai/cloud/clients/admin/infra"
 	"github.com/hanzoai/cloud/clients/admin/invoices"
 	"github.com/hanzoai/cloud/clients/admin/metrics"
 	"github.com/hanzoai/cloud/clients/admin/revenue"
@@ -141,6 +142,7 @@ func routes(app *zip.App, s *cloud.Service[core.State]) {
 	revenue.Routes(app, s)
 	finance.Routes(app, s)
 	metrics.Routes(app, s)
+	infra.Routes(app, s)
 	invoices.Routes(app, s)
 	subscriptions.Routes(app, s)
 }

clients/admin/admin_test.go

@@ -107,6 +107,10 @@ var platformAdminRoutes = []adminRoute{
 	{"GET", "/v1/admin/flags"},
 	{"GET", "/v1/admin/waitlist"},
 	{"POST", "/v1/admin/waitlist/boost"},
+	{"GET", "/v1/admin/infra"},
+	{"POST", "/v1/admin/infra/volumes/v1/snapshot"},
+	{"DELETE", "/v1/admin/infra/volumes/v1"},
+	{"POST", "/v1/admin/infra/nodes/1/cordon"},
 }
 
 // adminRoutes is the full surface (both tiers) — the fail-closed gate test denies an

clients/admin/digitalocean/do.go

@@ -1,7 +1,12 @@
-// Package do reads DigitalOcean's billing API for the finance dashboard's cost
-// side. DO is our PRIMARY venue (a large promotional credit); this client turns
-// the customer balance + billing history into money.Cents the finance aggregator
-// folds into gross margin and runway.
+// Package digitalocean reads DigitalOcean's billing and infrastructure APIs. DO is
+// our PRIMARY venue (a large promotional credit); this client turns the customer
+// balance + billing history into money.Cents the finance aggregator folds into gross
+// margin and runway, and exposes the account's physical inventory — droplets,
+// block-storage volumes, DOKS clusters, load balancers — that the /v1/admin/infra
+// board reads.
+//
+// This is the ONE DigitalOcean client the admin plane uses. A new DO read is a
+// method here calling the shared get/send primitive, never a second client.
 //
 // Auth is a single personal-access token, DO_API_TOKEN, sourced from a KMSSecret on
 // the cloud env — NEVER hard-coded. When the token is unset the client is not Ready
@@ -15,12 +20,14 @@
 package digitalocean
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"math"
 	"net/http"
+	"net/url"
 	"strconv"
 	"strings"
 	"time"
@@ -152,6 +159,12 @@ type Volume struct {
 	Region     string
 	SizeGiB    int
 	DropletIDs []int
+	// Tags carries DO's resource tags. DOKS stamps `k8s:<cluster-uuid>` on the volumes
+	// it provisions, but that tag is ADVISORY ONLY — it survives cluster deletion and
+	// is wrong often enough that it must never decide whether a volume is garbage. The
+	// only sound liveness test is a PV cross-reference (see clients/admin/infra).
+	Tags      []string
+	CreatedAt string
 }
 
 // volumeWire is the raw DO /v2/volumes row.
@@ -162,43 +175,326 @@ type volumeWire struct {
 	Region        struct {
 		Slug string `json:"slug"`
 	} `json:"region"`
-	DropletIDs []int `json:"droplet_ids"`
+	DropletIDs []int    `json:"droplet_ids"`
+	Tags       []string `json:"tags"`
+	CreatedAt  string   `json:"created_at"`
 }
 
-// Volumes lists ALL block-storage volumes across the account, following DO's
-// page-number pagination (200/page, a hard 25-page cap so a runaway can never loop).
-// The fleet inventory (count, capacity, monthly cost) is real; per-volume fill is NOT
-// exposed by DO and stays absent (honest) until a filesystem source reports it.
+// Volumes lists ALL block-storage volumes across the account. Capacity and attachment
+// are real; per-volume fill is NOT exposed by DO and stays absent (honest) until a
+// filesystem source reports it.
 func (c *Client) Volumes(ctx context.Context) ([]Volume, error) {
+	rows, err := listAll[volumeWire](ctx, c, "/v2/volumes", "volumes")
+	if err != nil {
+		return nil, err
+	}
+	out := make([]Volume, len(rows))
+	for i, v := range rows {
+		out[i] = Volume{
+			ID:         v.ID,
+			Name:       v.Name,
+			Region:     v.Region.Slug,
+			SizeGiB:    v.SizeGigabytes,
+			DropletIDs: v.DropletIDs,
+			Tags:       v.Tags,
+			CreatedAt:  v.CreatedAt,
+		}
+	}
+	return out, nil
+}
+
+// Droplet is one DO droplet. LocalDiskGiB is the droplet's own disk, which is
+// INCLUDED in MonthlyCents — it is NOT separately billed, and conflating it with
+// block storage is how a fleet appears to hold terabytes it never pays for.
+type Droplet struct {
+	ID           int
+	Name         string
+	Region       string
+	Status       string
+	SizeSlug     string
+	VCPUs        int
+	MemoryMiB    int
+	LocalDiskGiB int
+	MonthlyCents money.Cents
+	CreatedAt    string
+	PrivateIP    string
+	PublicIP     string
+	Tags         []string
+	VolumeIDs    []string
+}
+
+// dropletWire is the raw DO /v2/droplets row.
+type dropletWire struct {
+	ID       int    `json:"id"`
+	Name     string `json:"name"`
+	Status   string `json:"status"`
+	SizeSlug string `json:"size_slug"`
+	VCPUs    int    `json:"vcpus"`
+	Memory   int    `json:"memory"`
+	Disk     int    `json:"disk"`
+	Size     struct {
+		PriceMonthly float64 `json:"price_monthly"`
+	} `json:"size"`
+	Region struct {
+		Slug string `json:"slug"`
+	} `json:"region"`
+	Networks struct {
+		V4 []struct {
+			Type      string `json:"type"`
+			IPAddress string `json:"ip_address"`
+		} `json:"v4"`
+	} `json:"networks"`
+	CreatedAt string   `json:"created_at"`
+	Tags      []string `json:"tags"`
+	VolumeIDs []string `json:"volume_ids"`
+}
+
+// Droplets lists ALL droplets across the account.
+func (c *Client) Droplets(ctx context.Context) ([]Droplet, error) {
+	rows, err := listAll[dropletWire](ctx, c, "/v2/droplets", "droplets")
+	if err != nil {
+		return nil, err
+	}
+	out := make([]Droplet, len(rows))
+	for i, d := range rows {
+		dr := Droplet{
+			ID:           d.ID,
+			Name:         d.Name,
+			Region:       d.Region.Slug,
+			Status:       d.Status,
+			SizeSlug:     d.SizeSlug,
+			VCPUs:        d.VCPUs,
+			MemoryMiB:    d.Memory,
+			LocalDiskGiB: d.Disk,
+			MonthlyCents: centsOf(d.Size.PriceMonthly),
+			CreatedAt:    d.CreatedAt,
+			Tags:         d.Tags,
+			VolumeIDs:    d.VolumeIDs,
+		}
+		for _, n := range d.Networks.V4 {
+			switch n.Type {
+			case "private":
+				dr.PrivateIP = n.IPAddress
+			case "public":
+				dr.PublicIP = n.IPAddress
+			}
+		}
+		out[i] = dr
+	}
+	return out, nil
+}
+
+// Cluster is one DOKS cluster.
+type Cluster struct {
+	ID        string
+	Name      string
+	Region    string
+	Version   string
+	Status    string
+	NodePools int
+	CreatedAt string
+}
+
+// clusterWire is the raw DO /v2/kubernetes/clusters row.
+type clusterWire struct {
+	ID      string `json:"id"`
+	Name    string `json:"name"`
+	Region  string `json:"region"`
+	Version string `json:"version"`
+	Status  struct {
+		State string `json:"state"`
+	} `json:"status"`
+	NodePools []struct {
+		Name string `json:"name"`
+	} `json:"node_pools"`
+	CreatedAt string `json:"created_at"`
+}
+
+// Clusters lists ALL DOKS clusters. This is the authoritative denominator for the
+// orphan analysis: a volume may only be called unreferenced once EVERY cluster here
+// has been searched for a PV that claims it.
+func (c *Client) Clusters(ctx context.Context) ([]Cluster, error) {
+	rows, err := listAll[clusterWire](ctx, c, "/v2/kubernetes/clusters", "kubernetes_clusters")
+	if err != nil {
+		return nil, err
+	}
+	out := make([]Cluster, len(rows))
+	for i, k := range rows {
+		out[i] = Cluster{
+			ID:        k.ID,
+			Name:      k.Name,
+			Region:    k.Region,
+			Version:   k.Version,
+			Status:    k.Status.State,
+			NodePools: len(k.NodePools),
+			CreatedAt: k.CreatedAt,
+		}
+	}
+	return out, nil
+}
+
+// Kubeconfig fetches a cluster's admin kubeconfig. DO returns a token-based config
+// against the cluster's public https endpoint (never an exec plugin), which the
+// caller must still funnel through fleet.SafeRESTConfig before dialing.
+func (c *Client) Kubeconfig(ctx context.Context, clusterID string) ([]byte, error) {
 	if !c.Ready() {
 		return nil, fmt.Errorf("DO_API_TOKEN not configured")
 	}
-	var out []Volume
-	for page := 1; page <= 25; page++ {
-		body, err := c.get(ctx, "/v2/volumes?per_page=200&page="+strconv.Itoa(page))
+	if strings.TrimSpace(clusterID) == "" {
+		return nil, fmt.Errorf("cluster id required")
+	}
+	return c.get(ctx, "/v2/kubernetes/clusters/"+url.PathEscape(clusterID)+"/kubeconfig")
+}
+
+// LoadBalancer is one DO load balancer. DO does not price LBs in the API, so cost is
+// derived from the billed unit count (see lbUnitCents).
+type LoadBalancer struct {
+	ID           string
+	Name         string
+	Region       string
+	Status       string
+	IP           string
+	SizeUnit     int
+	MonthlyCents money.Cents
+	DropletIDs   []int
+}
+
+// lbWire is the raw DO /v2/load_balancers row.
+type lbWire struct {
+	ID     string `json:"id"`
+	Name   string `json:"name"`
+	Status string `json:"status"`
+	IP     string `json:"ip"`
+	Region struct {
+		Slug string `json:"slug"`
+	} `json:"region"`
+	SizeUnit   int   `json:"size_unit"`
+	DropletIDs []int `json:"droplet_ids"`
+}
+
+// LoadBalancers lists ALL load balancers across the account.
+func (c *Client) LoadBalancers(ctx context.Context) ([]LoadBalancer, error) {
+	rows, err := listAll[lbWire](ctx, c, "/v2/load_balancers", "load_balancers")
+	if err != nil {
+		return nil, err
+	}
+	out := make([]LoadBalancer, len(rows))
+	for i, l := range rows {
+		units := l.SizeUnit
+		if units <= 0 {
+			units = 1
+		}
+		out[i] = LoadBalancer{
+			ID:           l.ID,
+			Name:         l.Name,
+			Region:       l.Region.Slug,
+			Status:       l.Status,
+			IP:           l.IP,
+			SizeUnit:     units,
+			MonthlyCents: money.Cents(units) * lbUnitCents,
+			DropletIDs:   l.DropletIDs,
+		}
+	}
+	return out, nil
+}
+
+// Snapshot is a created block-storage snapshot.
+type Snapshot struct {
+	ID      string
+	Name    string
+	SizeGiB int
+}
+
+// SnapshotVolume takes a point-in-time snapshot of a volume. This is the "undo" that
+// makes a delete recoverable, so the delete path takes one FIRST by default.
+func (c *Client) SnapshotVolume(ctx context.Context, volumeID, name string) (Snapshot, error) {
+	var out Snapshot
+	if !c.Ready() {
+		return out, fmt.Errorf("DO_API_TOKEN not configured")
+	}
+	if strings.TrimSpace(volumeID) == "" {
+		return out, fmt.Errorf("volume id required")
+	}
+	body, err := c.send(ctx, http.MethodPost, "/v2/volumes/"+url.PathEscape(volumeID)+"/snapshots",
+		map[string]string{"name": name})
+	if err != nil {
+		return out, err
+	}
+	var w struct {
+		Snapshot struct {
+			ID            string `json:"id"`
+			Name          string `json:"name"`
+			SizeGigabytes int    `json:"size_gigabytes"`
+		} `json:"snapshot"`
+	}
+	if err := json.Unmarshal(body, &w); err != nil {
+		return out, fmt.Errorf("do snapshot decode: %w", err)
+	}
+	return Snapshot{ID: w.Snapshot.ID, Name: w.Snapshot.Name, SizeGiB: w.Snapshot.SizeGigabytes}, nil
+}
+
+// DeleteVolume destroys a block-storage volume. Irreversible: callers MUST have
+// proven the volume is referenced by no PV in any cluster first.
+func (c *Client) DeleteVolume(ctx context.Context, volumeID string) error {
+	if !c.Ready() {
+		return fmt.Errorf("DO_API_TOKEN not configured")
+	}
+	if strings.TrimSpace(volumeID) == "" {
+		return fmt.Errorf("volume id required")
+	}
+	_, err := c.send(ctx, http.MethodDelete, "/v2/volumes/"+url.PathEscape(volumeID), nil)
+	return err
+}
+
+// Pagination bounds for every DO collection read: 200 rows a page, a hard 25-page
+// cap so a runaway can never loop, and the 8 MiB body ceiling a full droplet page
+// needs (a volume page fits in far less).
+const (
+	perPage    = 200
+	maxPages   = 25
+	maxBody    = 8 << 20
+	maxRespLen = maxBody
+)
+
+// lbUnitCents is DO's published price for one load-balancer node ($12/mo). DO does
+// not return LB pricing in the API, so this is the one place the rate is written.
+const lbUnitCents = money.Cents(1200)
+
+// listAll follows DO's page-number pagination for a collection endpoint, decoding
+// rows out of the response's named key. It is the ONE pagination loop in this client
+// — every collection read goes through it, so "stop on the short page or the reported
+// total" is stated once and cannot drift between endpoints.
+func listAll[T any](ctx context.Context, c *Client, path, key string) ([]T, error) {
+	if !c.Ready() {
+		return nil, fmt.Errorf("DO_API_TOKEN not configured")
+	}
+	var out []T
+	for page := 1; page <= maxPages; page++ {
+		body, err := c.get(ctx, fmt.Sprintf("%s?per_page=%d&page=%d", path, perPage, page))
 		if err != nil {
 			return nil, err
 		}
 		var w struct {
-			Volumes []volumeWire `json:"volumes"`
-			Meta    struct {
+			Meta struct {
 				Total int `json:"total"`
 			} `json:"meta"`
 		}
 		if err := json.Unmarshal(body, &w); err != nil {
-			return nil, fmt.Errorf("do volumes decode: %w", err)
+			return nil, fmt.Errorf("do %s decode: %w", key, err)
 		}
-		for _, v := range w.Volumes {
-			out = append(out, Volume{
-				ID:         v.ID,
-				Name:       v.Name,
-				Region:     v.Region.Slug,
-				SizeGiB:    v.SizeGigabytes,
-				DropletIDs: v.DropletIDs,
-			})
+		var keyed map[string]json.RawMessage
+		if err := json.Unmarshal(body, &keyed); err != nil {
+			return nil, fmt.Errorf("do %s decode: %w", key, err)
 		}
+		var rows []T
+		if raw, ok := keyed[key]; ok && len(raw) > 0 {
+			if err := json.Unmarshal(raw, &rows); err != nil {
+				return nil, fmt.Errorf("do %s decode: %w", key, err)
+			}
+		}
+		out = append(out, rows...)
 		// Stop on the last (short) page, or once we've collected the reported total.
-		if len(w.Volumes) < 200 || (w.Meta.Total > 0 && len(out) >= w.Meta.Total) {
+		if len(rows) < perPage || (w.Meta.Total > 0 && len(out) >= w.Meta.Total) {
 			break
 		}
 	}
@@ -207,22 +503,48 @@ func (c *Client) Volumes(ctx context.Context) ([]Volume, error) {
 
 // get performs one token-authenticated DO GET and returns the raw body.
 func (c *Client) get(ctx context.Context, path string) ([]byte, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+path, nil)
+	return c.send(ctx, http.MethodGet, path, nil)
+}
+
+// send performs one token-authenticated DO request and returns the raw body. It is
+// the single HTTP primitive of this client: every read and every mutation funnels
+// through it, so auth, timeouts, the body ceiling and status handling exist once.
+func (c *Client) send(ctx context.Context, method, path string, payload any) ([]byte, error) {
+	var rdr io.Reader
+	if payload != nil {
+		enc, err := json.Marshal(payload)
+		if err != nil {
+			return nil, err
+		}
+		rdr = bytes.NewReader(enc)
+	}
+	req, err := http.NewRequestWithContext(ctx, method, c.base+path, rdr)
 	if err != nil {
 		return nil, err
 	}
 	req.Header.Set("Accept", "application/json")
 	req.Header.Set("Authorization", "Bearer "+c.token)
+	if payload != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
 	resp, err := c.http.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("digitalocean unreachable: %w", err)
 	}
 	defer resp.Body.Close()
-	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+	body, err := io.ReadAll(io.LimitReader(resp.Body, maxRespLen))
 	if err != nil {
 		return nil, err
 	}
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		// DO returns {"id":"...","message":"..."} on error — surface the message so a
+		// failed mutation says WHY, not just a bare status.
+		var e struct {
+			Message string `json:"message"`
+		}
+		if json.Unmarshal(body, &e) == nil && strings.TrimSpace(e.Message) != "" {
+			return nil, fmt.Errorf("digitalocean status %d: %s", resp.StatusCode, e.Message)
+		}
 		return nil, fmt.Errorf("digitalocean status %d", resp.StatusCode)
 	}
 	return body, nil
@@ -241,5 +563,8 @@ func dollarsToCents(s string) money.Cents {
 	if err != nil {
 		return 0
 	}
-	return money.Cents(math.Round(f * 100))
+	return centsOf(f)
 }
+
+// centsOf rounds decimal dollars to integer cents.
+func centsOf(f float64) money.Cents { return money.Cents(math.Round(f * 100)) }

clients/admin/infra/analyze.go

@@ -0,0 +1,768 @@
+// Package infra is the platform's DigitalOcean fleet board: the physical inventory
+// (DOKS clusters, droplets, block-storage volumes, load balancers) cross-referenced
+// against what every cluster's Kubernetes actually claims, with the cost of each and
+// an orphan analysis that is safe BY CONSTRUCTION.
+//
+// THE RULE THIS PACKAGE EXISTS TO ENFORCE. A volume being detached, or carrying a
+// `k8s:<cluster-uuid>` tag for some other cluster, does NOT make it garbage. Those two
+// signals together would have condemned 4.39 TiB of live data belonging to running
+// clusters. The ONLY sound liveness test is a cross-reference against the
+// `spec.csi.volumeHandle` of every PersistentVolume in EVERY cluster — and it is only
+// a valid test when every cluster answered. So:
+//
+//   - a volume is deletable only when NO PV in ANY cluster names it, and
+//   - if even one cluster failed to scan, NOTHING is deletable (Complete=false).
+//
+// Absence of evidence is not evidence of absence: an unreachable cluster is treated as
+// a cluster that might be holding the volume. The analysis fails CLOSED.
+//
+// "No pod mounts it" is a REVIEW signal, never a delete signal — an idle Bound PVC is
+// an idle database, not garbage. Idle volumes are surfaced as a queue for a human and
+// are never counted as reclaimable.
+//
+// Analyze is a pure function of (DO inventory, cluster scans) so every rule above is
+// unit-testable without a network or a cluster.
+package infra
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/hanzoai/cloud/clients/admin/core"
+	"github.com/hanzoai/cloud/clients/admin/digitalocean"
+	"github.com/hanzoai/cloud/clients/admin/money"
+)
+
+// volumeGiBCents is DO's block-storage rate: $0.10 per GiB per month. Droplet LOCAL
+// disk is NOT billed at this rate (or at all, separately) — it is included in the
+// droplet's own price. Conflating the two invents terabytes of phantom cost.
+const volumeGiBCents = money.Cents(10)
+
+// outlierShareBP flags any single resource costing at least this share of total fleet
+// spend, in basis points (500bp = 5%). One explicable rule, no magic thresholds.
+const outlierShareBP = money.Cents(500)
+
+// Volume states. The state machine is total and ordered: attachment beats reference,
+// reference beats absence. Only Unreferenced is ever deletable.
+const (
+	// StateAttached — DO reports the volume attached to a droplet. In use, now.
+	StateAttached = "attached"
+	// StateBound — detached, but a PV in some cluster claims it and that PV is Bound
+	// to a PVC. This is live data between mounts; deleting it destroys a database.
+	StateBound = "bound"
+	// StateReleased — a PV claims it but is no longer Bound (Released/Available/Failed).
+	// A genuine cleanup candidate, but the PV still exists, so a human retires the PV.
+	StateReleased = "released"
+	// StateUnreferenced — no PV in ANY scanned cluster names it. The ONLY deletable state.
+	StateUnreferenced = "unreferenced"
+)
+
+// Finding severities.
+const (
+	SevCritical = "critical"
+	SevWarn     = "warn"
+	SevInfo     = "info"
+)
+
+// firstParty are our own registries: anything here is ours by construction.
+var firstParty = []string{
+	"ghcr.io/hanzoai/", "ghcr.io/luxfi/", "ghcr.io/zooai/",
+	"registry.hanzo.ai/", "registry.lux.network/", "registry.zoo.network/",
+	"registry.digitalocean.com/hanzo/",
+}
+
+// knownVendors is the REVIEWED third-party set — upstream images we deliberately run.
+// Kept deliberately short: an image outside both lists is reported for a human to
+// judge, which is the point. Growing this list is a review decision, not a reflex.
+var knownVendors = []string{
+	"docker.io/library/", "library/", "registry.k8s.io/", "k8s.gcr.io/", "quay.io/",
+	"grafana/", "prom/", "prometheus/", "bitnami/", "minio/", "moby/",
+	"digitalocean/", "docker.digitalocean.com/", "acmglobaltech/", "hanzozt/",
+}
+
+// Inventory is the DigitalOcean account read — the half of the analysis input that
+// needs no cluster.
+type Inventory struct {
+	Clusters      []digitalocean.Cluster
+	Droplets      []digitalocean.Droplet
+	Volumes       []digitalocean.Volume
+	LoadBalancers []digitalocean.LoadBalancer
+}
+
+// PVRef is one PersistentVolume's identity: which DO volume it claims, and whether
+// that claim is still live.
+type PVRef struct {
+	Name         string
+	Phase        string
+	VolumeHandle string
+	ClaimNS      string
+	ClaimName    string
+}
+
+// PVCRef is one PersistentVolumeClaim.
+type PVCRef struct {
+	Namespace string
+	Name      string
+	Phase     string
+	Volume    string
+}
+
+// PodRef is one pod, reduced to what the board needs: where it runs, whether it is
+// healthy, which PVCs it mounts, and what images it runs.
+type PodRef struct {
+	Namespace string
+	Name      string
+	Phase     string
+	Reason    string
+	Node      string
+	Claims    []string
+	Images    []string
+}
+
+// NodeState is one Kubernetes node's own view of itself.
+type NodeState struct {
+	Name        string
+	Ready       bool
+	Schedulable bool
+}
+
+// ClusterScan is ONE cluster's Kubernetes truth. Err non-nil means the cluster did
+// not answer — which forces the whole analysis incomplete.
+type ClusterScan struct {
+	ClusterID string
+	Err       error
+	Nodes     []NodeState
+	PVs       []PVRef
+	PVCs      []PVCRef
+	Pods      []PodRef
+}
+
+// Snapshot is the whole board in one value.
+type Snapshot struct {
+	At               string              `json:"at"`
+	Complete         bool                `json:"complete"`
+	IncompleteReason string              `json:"incompleteReason"`
+	Sources          []core.SourceStatus `json:"sources"`
+	Totals           Totals              `json:"totals"`
+	Cost             Cost                `json:"cost"`
+	Clusters         []Cluster           `json:"clusters"`
+	Nodes            []Node              `json:"nodes"`
+	Volumes          []Volume            `json:"volumes"`
+	LoadBalancers    []LoadBalancer      `json:"loadBalancers"`
+	Findings         []Finding           `json:"findings"`
+}
+
+// Totals are fleet counts. LocalDiskGiB is broken out precisely so it can be shown
+// as NOT separately billed.
+type Totals struct {
+	Clusters            int `json:"clusters"`
+	Nodes               int `json:"nodes"`
+	Volumes             int `json:"volumes"`
+	LoadBalancers       int `json:"loadBalancers"`
+	VolumeGiB           int `json:"volumeGiB"`
+	AttachedVolumes     int `json:"attachedVolumes"`
+	AttachedGiB         int `json:"attachedGiB"`
+	DetachedVolumes     int `json:"detachedVolumes"`
+	DetachedGiB         int `json:"detachedGiB"`
+	UnreferencedVolumes int `json:"unreferencedVolumes"`
+	UnreferencedGiB     int `json:"unreferencedGiB"`
+	IdlePVCs            int `json:"idlePVCs"`
+	LocalDiskGiB        int `json:"localDiskGiB"`
+}
+
+// Cost is monthly spend in cents. Reclaimable counts ONLY unreferenced volumes —
+// never idle ones, which are live data awaiting a human verdict.
+type Cost struct {
+	DropletsMonthly      money.Cents `json:"dropletsMonthly"`
+	VolumesMonthly       money.Cents `json:"volumesMonthly"`
+	LoadBalancersMonthly money.Cents `json:"loadBalancersMonthly"`
+	TotalMonthly         money.Cents `json:"totalMonthly"`
+	ReclaimableMonthly   money.Cents `json:"reclaimableMonthly"`
+}
+
+// Cluster is one DOKS cluster with its scanned Kubernetes rollup.
+type Cluster struct {
+	ID           string      `json:"id"`
+	Name         string      `json:"name"`
+	Region       string      `json:"region"`
+	Version      string      `json:"version"`
+	Status       string      `json:"status"`
+	NodePools    int         `json:"nodePools"`
+	Nodes        int         `json:"nodes"`
+	Pods         int         `json:"pods"`
+	PVs          int         `json:"pvs"`
+	PVCs         int         `json:"pvcs"`
+	IdlePVCs     int         `json:"idlePVCs"`
+	Scanned      bool        `json:"scanned"`
+	ScanError    string      `json:"scanError"`
+	MonthlyCents money.Cents `json:"monthlyCents"`
+}
+
+// Node is one droplet, joined to the Kubernetes node of the same name.
+type Node struct {
+	ID           int         `json:"id"`
+	Name         string      `json:"name"`
+	Cluster      string      `json:"cluster"`
+	ClusterID    string      `json:"clusterId"`
+	Region       string      `json:"region"`
+	Status       string      `json:"status"`
+	SizeSlug     string      `json:"sizeSlug"`
+	VCPUs        int         `json:"vcpus"`
+	MemoryMiB    int         `json:"memoryMiB"`
+	LocalDiskGiB int         `json:"localDiskGiB"`
+	MonthlyCents money.Cents `json:"monthlyCents"`
+	CreatedAt    string      `json:"createdAt"`
+	PrivateIP    string      `json:"privateIp"`
+	PublicIP     string      `json:"publicIp"`
+	Tags         []string    `json:"tags"`
+	Ready        bool        `json:"ready"`
+	Schedulable  bool        `json:"schedulable"`
+	Pods         int         `json:"pods"`
+	Volumes      int         `json:"volumes"`
+}
+
+// Volume is one block-storage volume with its PROVEN cluster ownership.
+type Volume struct {
+	ID           string      `json:"id"`
+	Name         string      `json:"name"`
+	Region       string      `json:"region"`
+	SizeGiB      int         `json:"sizeGiB"`
+	MonthlyCents money.Cents `json:"monthlyCents"`
+	State        string      `json:"state"`
+	DropletIDs   []int       `json:"dropletIds"`
+	NodeName     string      `json:"nodeName"`
+	// Cluster/ClusterID are the PROVEN owner — resolved through a PV that names this
+	// volume, never through the tag.
+	Cluster   string `json:"cluster"`
+	ClusterID string `json:"clusterId"`
+	// TagCluster is the `k8s:<uuid>` tag. ADVISORY ONLY: it outlives the cluster that
+	// set it. Shown so the operator can see tag-vs-truth disagree, never acted on.
+	TagCluster    string   `json:"tagCluster"`
+	PV            string   `json:"pv"`
+	PVPhase       string   `json:"pvPhase"`
+	PVCNamespace  string   `json:"pvcNamespace"`
+	PVCName       string   `json:"pvcName"`
+	MountedBy     []string `json:"mountedBy"`
+	Idle          bool     `json:"idle"`
+	CreatedAt     string   `json:"createdAt"`
+	Deletable     bool     `json:"deletable"`
+	BlockedReason string   `json:"blockedReason"`
+}
+
+// LoadBalancer is one DO load balancer, attributed to a cluster via its members.
+type LoadBalancer struct {
+	ID           string      `json:"id"`
+	Name         string      `json:"name"`
+	Region       string      `json:"region"`
+	Status       string      `json:"status"`
+	IP           string      `json:"ip"`
+	SizeUnit     int         `json:"sizeUnit"`
+	MonthlyCents money.Cents `json:"monthlyCents"`
+	Droplets     int         `json:"droplets"`
+	Cluster      string      `json:"cluster"`
+}
+
+// Finding is one audit result — the "is anything bad" surface.
+type Finding struct {
+	ID           string      `json:"id"`
+	Severity     string      `json:"severity"`
+	Kind         string      `json:"kind"`
+	Title        string      `json:"title"`
+	Detail       string      `json:"detail"`
+	Resource     string      `json:"resource"`
+	Cluster      string      `json:"cluster"`
+	MonthlyCents money.Cents `json:"monthlyCents"`
+}
+
+// pvHit is a PV that claims a given DO volume, plus the cluster it lives in.
+type pvHit struct {
+	cluster   string
+	clusterID string
+	pv        PVRef
+}
+
+// Analyze folds the DO inventory and the per-cluster Kubernetes scans into the board.
+// PURE: no clock, no network, no cluster — `at` is passed in so the result is
+// byte-reproducible in tests.
+func Analyze(inv Inventory, scans []ClusterScan, sources []core.SourceStatus, at time.Time) Snapshot {
+	snap := Snapshot{
+		At:      at.UTC().Format(time.RFC3339),
+		Sources: sources,
+	}
+	if snap.Sources == nil {
+		snap.Sources = []core.SourceStatus{}
+	}
+
+	scanByID := make(map[string]ClusterScan, len(scans))
+	for _, s := range scans {
+		scanByID[s.ClusterID] = s
+	}
+	nameByID := make(map[string]string, len(inv.Clusters))
+	for _, c := range inv.Clusters {
+		nameByID[c.ID] = c.Name
+	}
+
+	// ---- completeness gate -------------------------------------------------------
+	// Every cluster must have answered. One silent gap and no volume may be condemned.
+	var unreachable []string
+	for _, c := range inv.Clusters {
+		s, ok := scanByID[c.ID]
+		if !ok || s.Err != nil {
+			unreachable = append(unreachable, c.Name)
+		}
+	}
+	switch {
+	case len(inv.Clusters) == 0:
+		snap.IncompleteReason = "DigitalOcean returned no clusters — the set of places a volume could be in use is unknown."
+	case len(unreachable) > 0:
+		snap.IncompleteReason = fmt.Sprintf(
+			"%d of %d clusters did not answer (%s) — a volume they hold would look unreferenced, so nothing is classified as deletable.",
+			len(unreachable), len(inv.Clusters), strings.Join(unreachable, ", "))
+	default:
+		snap.Complete = true
+	}
+
+	// ---- cross-cluster PV index --------------------------------------------------
+	// THE safety index: every volume handle claimed by any PV in any cluster.
+	byHandle := make(map[string]pvHit)
+	// mounted[clusterID/ns/pvc] -> pods currently mounting it.
+	mounted := make(map[string][]string)
+	for _, s := range scans {
+		if s.Err != nil {
+			continue
+		}
+		cname := nameByID[s.ClusterID]
+		for _, pv := range s.PVs {
+			if h := strings.TrimSpace(pv.VolumeHandle); h != "" {
+				byHandle[h] = pvHit{cluster: cname, clusterID: s.ClusterID, pv: pv}
+			}
+		}
+		for _, p := range s.Pods {
+			for _, claim := range p.Claims {
+				k := claimKey(s.ClusterID, p.Namespace, claim)
+				mounted[k] = append(mounted[k], p.Namespace+"/"+p.Name)
+			}
+		}
+	}
+
+	// ---- nodes -------------------------------------------------------------------
+	nodeByName := make(map[string]NodeState)
+	podsPerNode := make(map[string]int)
+	for _, s := range scans {
+		if s.Err != nil {
+			continue
+		}
+		for _, n := range s.Nodes {
+			nodeByName[n.Name] = n
+		}
+		for _, p := range s.Pods {
+			if p.Node != "" {
+				podsPerNode[p.Node]++
+			}
+		}
+	}
+	volsPerDroplet := make(map[int]int)
+	for _, v := range inv.Volumes {
+		for _, id := range v.DropletIDs {
+			volsPerDroplet[id]++
+		}
+	}
+	clusterByDroplet := make(map[int]string, len(inv.Droplets))
+	snap.Nodes = make([]Node, 0, len(inv.Droplets))
+	for _, d := range inv.Droplets {
+		cid := clusterIDFromTags(d.Tags)
+		clusterByDroplet[d.ID] = cid
+		ks := nodeByName[d.Name]
+		snap.Nodes = append(snap.Nodes, Node{
+			ID: d.ID, Name: d.Name, Cluster: nameByID[cid], ClusterID: cid,
+			Region: d.Region, Status: d.Status, SizeSlug: d.SizeSlug,
+			VCPUs: d.VCPUs, MemoryMiB: d.MemoryMiB, LocalDiskGiB: d.LocalDiskGiB,
+			MonthlyCents: d.MonthlyCents, CreatedAt: d.CreatedAt,
+			PrivateIP: d.PrivateIP, PublicIP: d.PublicIP, Tags: nonNilStrings(d.Tags),
+			Ready: ks.Ready, Schedulable: ks.Schedulable,
+			Pods: podsPerNode[d.Name], Volumes: volsPerDroplet[d.ID],
+		})
+		snap.Cost.DropletsMonthly += d.MonthlyCents
+		snap.Totals.LocalDiskGiB += d.LocalDiskGiB
+	}
+	dropletName := make(map[int]string, len(inv.Droplets))
+	for _, d := range inv.Droplets {
+		dropletName[d.ID] = d.Name
+	}
+
+	// ---- volumes: the state machine ----------------------------------------------
+	snap.Volumes = make([]Volume, 0, len(inv.Volumes))
+	for _, v := range inv.Volumes {
+		hit, referenced := byHandle[v.ID]
+		vol := Volume{
+			ID: v.ID, Name: v.Name, Region: v.Region, SizeGiB: v.SizeGiB,
+			MonthlyCents: money.Cents(v.SizeGiB) * volumeGiBCents,
+			DropletIDs:   nonNilInts(v.DropletIDs),
+			TagCluster:   nameByID[clusterIDFromTags(v.Tags)],
+			CreatedAt:    v.CreatedAt,
+			MountedBy:    []string{},
+		}
+		if len(v.DropletIDs) > 0 {
+			vol.NodeName = dropletName[v.DropletIDs[0]]
+		}
+		if referenced {
+			vol.Cluster, vol.ClusterID = hit.cluster, hit.clusterID
+			vol.PV, vol.PVPhase = hit.pv.Name, hit.pv.Phase
+			vol.PVCNamespace, vol.PVCName = hit.pv.ClaimNS, hit.pv.ClaimName
+			if hit.pv.ClaimName != "" {
+				vol.MountedBy = nonNilStrings(mounted[claimKey(hit.clusterID, hit.pv.ClaimNS, hit.pv.ClaimName)])
+			}
+		} else if len(v.DropletIDs) > 0 {
+			// No PV names it, but it is physically mounted on a node — that node's
+			// cluster owns it. Attachment is hard evidence, unlike the tag: it is the
+			// live kernel state, so the cost rolls up to the right cluster.
+			vol.ClusterID = clusterByDroplet[v.DropletIDs[0]]
+			vol.Cluster = nameByID[vol.ClusterID]
+		}
+
+		// Attachment beats reference; reference beats absence.
+		switch {
+		case len(v.DropletIDs) > 0:
+			vol.State = StateAttached
+		case referenced && strings.EqualFold(hit.pv.Phase, "Bound"):
+			vol.State = StateBound
+		case referenced:
+			vol.State = StateReleased
+		default:
+			vol.State = StateUnreferenced
+		}
+
+		// Idle is a REVIEW signal on live data, never a delete signal.
+		vol.Idle = vol.State == StateBound && len(vol.MountedBy) == 0
+
+		vol.Deletable = snap.Complete && vol.State == StateUnreferenced
+		vol.BlockedReason = blockedReason(vol, snap.Complete, snap.IncompleteReason)
+
+		snap.Cost.VolumesMonthly += vol.MonthlyCents
+		snap.Totals.VolumeGiB += vol.SizeGiB
+		switch vol.State {
+		case StateAttached:
+			snap.Totals.AttachedVolumes++
+			snap.Totals.AttachedGiB += vol.SizeGiB
+		default:
+			snap.Totals.DetachedVolumes++
+			snap.Totals.DetachedGiB += vol.SizeGiB
+		}
+		if vol.State == StateUnreferenced {
+			snap.Totals.UnreferencedVolumes++
+			snap.Totals.UnreferencedGiB += vol.SizeGiB
+			// Reclaimable is exactly the unreferenced set — and only when the scan was
+			// complete enough to have earned that verdict.
+			if snap.Complete {
+				snap.Cost.ReclaimableMonthly += vol.MonthlyCents
+			}
+		}
+		if vol.Idle {
+			snap.Totals.IdlePVCs++
+		}
+		snap.Volumes = append(snap.Volumes, vol)
+	}
+
+	// ---- load balancers ----------------------------------------------------------
+	snap.LoadBalancers = make([]LoadBalancer, 0, len(inv.LoadBalancers))
+	for _, l := range inv.LoadBalancers {
+		lb := LoadBalancer{
+			ID: l.ID, Name: l.Name, Region: l.Region, Status: l.Status, IP: l.IP,
+			SizeUnit: l.SizeUnit, MonthlyCents: l.MonthlyCents, Droplets: len(l.DropletIDs),
+		}
+		for _, id := range l.DropletIDs {
+			if cid := clusterByDroplet[id]; cid != "" {
+				lb.Cluster = nameByID[cid]
+				break
+			}
+		}
+		snap.Cost.LoadBalancersMonthly += lb.MonthlyCents
+		snap.LoadBalancers = append(snap.LoadBalancers, lb)
+	}
+
+	// ---- cluster rollup ----------------------------------------------------------
+	idleByCluster := make(map[string]int)
+	costByCluster := make(map[string]money.Cents)
+	for _, v := range snap.Volumes {
+		if v.ClusterID != "" {
+			costByCluster[v.ClusterID] += v.MonthlyCents
+			if v.Idle {
+				idleByCluster[v.ClusterID]++
+			}
+		}
+	}
+	nodesByCluster := make(map[string]int)
+	for _, n := range snap.Nodes {
+		if n.ClusterID != "" {
+			nodesByCluster[n.ClusterID]++
+			costByCluster[n.ClusterID] += n.MonthlyCents
+		}
+	}
+	snap.Clusters = make([]Cluster, 0, len(inv.Clusters))
+	for _, c := range inv.Clusters {
+		row := Cluster{
+			ID: c.ID, Name: c.Name, Region: c.Region, Version: c.Version,
+			Status: c.Status, NodePools: c.NodePools, Nodes: nodesByCluster[c.ID],
+			IdlePVCs: idleByCluster[c.ID], MonthlyCents: costByCluster[c.ID],
+		}
+		if s, ok := scanByID[c.ID]; ok {
+			if s.Err != nil {
+				row.ScanError = s.Err.Error()
+			} else {
+				row.Scanned = true
+				row.Pods, row.PVs, row.PVCs = len(s.Pods), len(s.PVs), len(s.PVCs)
+			}
+		} else {
+			row.ScanError = "not scanned"
+		}
+		snap.Clusters = append(snap.Clusters, row)
+	}
+
+	snap.Totals.Clusters = len(snap.Clusters)
+	snap.Totals.Nodes = len(snap.Nodes)
+	snap.Totals.Volumes = len(snap.Volumes)
+	snap.Totals.LoadBalancers = len(snap.LoadBalancers)
+	snap.Cost.TotalMonthly = snap.Cost.DropletsMonthly + snap.Cost.VolumesMonthly + snap.Cost.LoadBalancersMonthly
+
+	snap.Findings = findings(snap, scans, nameByID)
+	return snap
+}
+
+// blockedReason states, in the operator's language, exactly why a volume may not be
+// deleted. An empty string means it may.
+func blockedReason(v Volume, complete bool, incomplete string) string {
+	if v.Deletable {
+		return ""
+	}
+	switch {
+	case !complete:
+		return incomplete
+	case v.State == StateAttached:
+		if v.NodeName != "" {
+			return "Attached to " + v.NodeName + " and in use."
+		}
+		return "Attached to a droplet and in use."
+	case v.State == StateBound:
+		return fmt.Sprintf("Live data: PV %s is Bound to %s/%s in %s.", v.PV, v.PVCNamespace, v.PVCName, v.Cluster)
+	case v.State == StateReleased:
+		return fmt.Sprintf("PV %s in %s still references it (%s) — retire the PV first.", v.PV, v.Cluster, v.PVPhase)
+	}
+	return "Not eligible for deletion."
+}
+
+// findings is the audit pass: what a human should look at, worst first.
+func findings(s Snapshot, scans []ClusterScan, nameByID map[string]string) []Finding {
+	out := []Finding{}
+
+	if !s.Complete {
+		out = append(out, Finding{
+			ID: "scan-incomplete", Severity: SevCritical, Kind: "scan-incomplete",
+			Title:  "Cluster scan incomplete — deletion disabled",
+			Detail: s.IncompleteReason,
+		})
+	}
+
+	for _, v := range s.Volumes {
+		switch {
+		case v.State == StateUnreferenced && s.Complete:
+			out = append(out, Finding{
+				ID: "unref/" + v.ID, Severity: SevWarn, Kind: "unreferenced-volume",
+				Title: fmt.Sprintf("Unreferenced volume %s (%d GiB)", v.Name, v.SizeGiB),
+				Detail: "No PersistentVolume in any cluster references this volume. " +
+					"Verified against every cluster, so it is safe to snapshot and delete.",
+				Resource: v.ID, MonthlyCents: v.MonthlyCents,
+			})
+		case v.State == StateReleased:
+			out = append(out, Finding{
+				ID: "released/" + v.ID, Severity: SevWarn, Kind: "released-pv",
+				Title:    fmt.Sprintf("Released PV holding %s (%d GiB)", v.Name, v.SizeGiB),
+				Detail:   fmt.Sprintf("PV %s is %s. Retire the PV to release the volume.", v.PV, v.PVPhase),
+				Resource: v.ID, Cluster: v.Cluster, MonthlyCents: v.MonthlyCents,
+			})
+		case v.Idle:
+			out = append(out, Finding{
+				ID: "idle/" + v.ID, Severity: SevInfo, Kind: "idle-pvc",
+				Title: fmt.Sprintf("Idle volume %s (%d GiB) — no pod mounts it", v.Name, v.SizeGiB),
+				Detail: fmt.Sprintf("PVC %s/%s is Bound but no running pod mounts it. "+
+					"REVIEW ONLY: this is live data (typically a stopped database), not garbage.",
+					v.PVCNamespace, v.PVCName),
+				Resource: v.ID, Cluster: v.Cluster, MonthlyCents: v.MonthlyCents,
+			})
+		}
+	}
+
+	// Unhealthy pods + unknown images, per cluster.
+	type imgSeen struct {
+		pods    int
+		cluster string
+	}
+	unknown := map[string]*imgSeen{}
+	for _, sc := range scans {
+		if sc.Err != nil {
+			continue
+		}
+		cname := nameByID[sc.ClusterID]
+		for _, p := range sc.Pods {
+			if bad := podProblem(p); bad != "" {
+				out = append(out, Finding{
+					ID: "pod/" + sc.ClusterID + "/" + p.Namespace + "/" + p.Name, Severity: SevWarn,
+					Kind: "pod-unhealthy", Title: fmt.Sprintf("Pod %s/%s is %s", p.Namespace, p.Name, bad),
+					Detail:   fmt.Sprintf("Phase %s%s on node %s.", p.Phase, reasonSuffix(p.Reason), p.Node),
+					Resource: p.Namespace + "/" + p.Name, Cluster: cname,
+				})
+			}
+			for _, img := range p.Images {
+				if knownImage(img) {
+					continue
+				}
+				repo := imageRepo(img)
+				if e, ok := unknown[repo]; ok {
+					e.pods++
+				} else {
+					unknown[repo] = &imgSeen{pods: 1, cluster: cname}
+				}
+			}
+		}
+	}
+	for repo, e := range unknown {
+		out = append(out, Finding{
+			ID: "image/" + repo, Severity: SevWarn, Kind: "unknown-image",
+			Title:    "Unrecognised container image: " + repo,
+			Detail:   fmt.Sprintf("Run by %d pod(s), from neither our registries nor the reviewed vendor set.", e.pods),
+			Resource: repo, Cluster: e.cluster,
+		})
+	}
+
+	// Cost outliers: any single resource at or above outlierShareBP of total spend.
+	if s.Cost.TotalMonthly > 0 {
+		threshold := s.Cost.TotalMonthly * outlierShareBP / 10000
+		for _, n := range s.Nodes {
+			if n.MonthlyCents >= threshold {
+				out = append(out, Finding{
+					ID: "cost/node/" + n.Name, Severity: SevInfo, Kind: "cost-outlier",
+					Title:    fmt.Sprintf("Node %s is %s of fleet spend", n.Name, shareLabel(n.MonthlyCents, s.Cost.TotalMonthly)),
+					Detail:   fmt.Sprintf("%s, %d vCPU / %d MiB.", n.SizeSlug, n.VCPUs, n.MemoryMiB),
+					Resource: n.Name, Cluster: n.Cluster, MonthlyCents: n.MonthlyCents,
+				})
+			}
+		}
+		for _, v := range s.Volumes {
+			if v.MonthlyCents >= threshold {
+				out = append(out, Finding{
+					ID: "cost/volume/" + v.ID, Severity: SevInfo, Kind: "cost-outlier",
+					Title:    fmt.Sprintf("Volume %s is %s of fleet spend", v.Name, shareLabel(v.MonthlyCents, s.Cost.TotalMonthly)),
+					Detail:   fmt.Sprintf("%d GiB, %s.", v.SizeGiB, v.State),
+					Resource: v.ID, Cluster: v.Cluster, MonthlyCents: v.MonthlyCents,
+				})
+			}
+		}
+	}
+
+	rank := map[string]int{SevCritical: 0, SevWarn: 1, SevInfo: 2}
+	sort.SliceStable(out, func(i, j int) bool {
+		if rank[out[i].Severity] != rank[out[j].Severity] {
+			return rank[out[i].Severity] < rank[out[j].Severity]
+		}
+		if out[i].MonthlyCents != out[j].MonthlyCents {
+			return out[i].MonthlyCents > out[j].MonthlyCents
+		}
+		return out[i].ID < out[j].ID
+	})
+	return out
+}
+
+// podProblem names the failure a pod is in, or "" when it is fine.
+func podProblem(p PodRef) string {
+	switch {
+	case strings.EqualFold(p.Reason, "Evicted"):
+		return "Evicted"
+	case strings.EqualFold(p.Phase, "Failed"):
+		return "Failed"
+	case strings.Contains(p.Reason, "CrashLoopBackOff"):
+		return "CrashLoopBackOff"
+	case strings.Contains(p.Reason, "ImagePullBackOff"), strings.Contains(p.Reason, "ErrImagePull"):
+		return "ImagePullBackOff"
+	}
+	return ""
+}
+
+func reasonSuffix(r string) string {
+	if strings.TrimSpace(r) == "" {
+		return ""
+	}
+	return " (" + r + ")"
+}
+
+// knownImage reports whether an image comes from our registries or the reviewed
+// vendor set.
+func knownImage(img string) bool {
+	l := strings.ToLower(strings.TrimSpace(img))
+	l = strings.TrimPrefix(l, "docker.io/")
+	for _, p := range firstParty {
+		if strings.HasPrefix(l, strings.TrimPrefix(p, "docker.io/")) {
+			return true
+		}
+	}
+	for _, p := range knownVendors {
+		if strings.HasPrefix(l, strings.TrimPrefix(p, "docker.io/")) {
+			return true
+		}
+	}
+	// A bare `name:tag` with no slash is an official Docker Hub library image.
+	return !strings.Contains(strings.SplitN(l, ":", 2)[0], "/")
+}
+
+// imageRepo strips the tag/digest so findings group by repository, not by build.
+func imageRepo(img string) string {
+	s := strings.TrimSpace(img)
+	if i := strings.Index(s, "@"); i > 0 {
+		s = s[:i]
+	}
+	if i := strings.LastIndex(s, ":"); i > strings.LastIndex(s, "/") {
+		s = s[:i]
+	}
+	return s
+}
+
+// shareLabel renders a cents-of-total share as a percentage.
+func shareLabel(part, total money.Cents) string {
+	if total <= 0 {
+		return "0%"
+	}
+	return fmt.Sprintf("%.1f%%", float64(part)*100/float64(total))
+}
+
+// clusterIDFromTags extracts the DOKS cluster UUID from a `k8s:<uuid>` resource tag.
+// On droplets this is authoritative (DOKS owns the droplet); on VOLUMES it is
+// advisory only — see the Volume.TagCluster doc.
+func clusterIDFromTags(tags []string) string {
+	for _, t := range tags {
+		v := strings.TrimPrefix(t, "k8s:")
+		if v == t || v == "" {
+			continue
+		}
+		// Cluster tags are UUIDs; DOKS also stamps role tags like `k8s:worker`.
+		if len(v) == 36 && strings.Count(v, "-") == 4 {
+			return v
+		}
+	}
+	return ""
+}
+
+func claimKey(clusterID, ns, name string) string { return clusterID + "/" + ns + "/" + name }
+
+func nonNilStrings(s []string) []string {
+	if s == nil {
+		return []string{}
+	}
+	return s
+}
+
+func nonNilInts(s []int) []int {
+	if s == nil {
+		return []int{}
+	}
+	return s
+}

clients/admin/infra/analyze_test.go

@@ -0,0 +1,390 @@
+package infra
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/hanzoai/cloud/clients/admin/digitalocean"
+)
+
+// Two live clusters. The near-miss that motivated this package involved volumes tagged
+// for one cluster whose PVs actually live in another, so every fixture here has two.
+const (
+	cidA = "aaaaaaaa-1111-2222-3333-444444444444"
+	cidB = "bbbbbbbb-1111-2222-3333-444444444444"
+)
+
+func baseInventory() Inventory {
+	return Inventory{
+		Clusters: []digitalocean.Cluster{
+			{ID: cidA, Name: "hanzo-k8s", Region: "sfo3", Status: "running"},
+			{ID: cidB, Name: "lux-k8s", Region: "sfo3", Status: "running"},
+		},
+		Droplets: []digitalocean.Droplet{{
+			ID: 101, Name: "node-a1", Region: "sfo3", Status: "active",
+			SizeSlug: "s-8vcpu-16gb-amd", VCPUs: 8, MemoryMiB: 16384,
+			LocalDiskGiB: 320, MonthlyCents: 11200,
+			Tags: []string{"k8s", "k8s:" + cidA, "k8s:worker"},
+		}},
+	}
+}
+
+func scansOK() []ClusterScan {
+	return []ClusterScan{{ClusterID: cidA}, {ClusterID: cidB}}
+}
+
+func volByID(t *testing.T, s Snapshot, id string) Volume {
+	t.Helper()
+	v, ok := findVolume(s, id)
+	if !ok {
+		t.Fatalf("volume %s missing from snapshot", id)
+	}
+	return v
+}
+
+func analyze(inv Inventory, scans []ClusterScan) Snapshot {
+	return Analyze(inv, scans, nil, time.Unix(0, 0).UTC())
+}
+
+// TestCrossClusterPVProtectsMisTaggedVolume is THE regression test. A detached volume
+// tagged `k8s:<cluster A>` whose PV actually lives in cluster B must be classified from
+// the PV, not the tag. Trusting the tag here is what nearly destroyed 4.39 TiB.
+func TestCrossClusterPVProtectsMisTaggedVolume(t *testing.T) {
+	inv := baseInventory()
+	inv.Volumes = []digitalocean.Volume{{
+		ID: "vol-mistagged", Name: "pvc-neon-pageserver", Region: "sfo3", SizeGiB: 50,
+		DropletIDs: nil,                     // detached
+		Tags:       []string{"k8s:" + cidA}, // tag says cluster A …
+	}}
+	scans := scansOK()
+	// … but the PV that owns it lives in cluster B, and is Bound to a live PVC.
+	scans[1].PVs = []PVRef{{
+		Name: "pv-neon", Phase: "Bound", VolumeHandle: "vol-mistagged",
+		ClaimNS: "neon", ClaimName: "pageserver-data",
+	}}
+
+	got := analyze(inv, scans)
+	v := volByID(t, got, "vol-mistagged")
+
+	if v.State != StateBound {
+		t.Fatalf("state = %q, want %q — a detached, mis-tagged volume with a Bound PV is LIVE DATA", v.State, StateBound)
+	}
+	if v.Deletable {
+		t.Fatal("volume marked deletable: this is the 4.39 TiB data-loss bug")
+	}
+	if v.Cluster != "lux-k8s" {
+		t.Errorf("proven cluster = %q, want lux-k8s (from the PV, not the tag)", v.Cluster)
+	}
+	if v.TagCluster != "hanzo-k8s" {
+		t.Errorf("tagCluster = %q, want hanzo-k8s (advisory, surfaced so tag-vs-truth is visible)", v.TagCluster)
+	}
+	if got.Cost.ReclaimableMonthly != 0 {
+		t.Errorf("reclaimable = %d, want 0", got.Cost.ReclaimableMonthly)
+	}
+}
+
+// TestIncompleteScanBlocksEveryDeletion: one unreachable cluster and NOTHING is
+// deletable, even a volume no reachable cluster references. Absence of evidence is not
+// evidence of absence.
+func TestIncompleteScanBlocksEveryDeletion(t *testing.T) {
+	inv := baseInventory()
+	inv.Volumes = []digitalocean.Volume{{ID: "vol-orphan", Name: "stray", SizeGiB: 100}}
+	scans := scansOK()
+	scans[1].Err = errors.New("dial tcp: i/o timeout")
+
+	got := analyze(inv, scans)
+	if got.Complete {
+		t.Fatal("Complete = true with an unreachable cluster")
+	}
+	v := volByID(t, got, "vol-orphan")
+	if v.State != StateUnreferenced {
+		t.Errorf("state = %q, want %q (state is observable; the VERDICT is what is withheld)", v.State, StateUnreferenced)
+	}
+	if v.Deletable {
+		t.Fatal("deletable with an incomplete scan — fail-closed violated")
+	}
+	if got.Cost.ReclaimableMonthly != 0 {
+		t.Errorf("reclaimable = %d, want 0 when the scan is incomplete", got.Cost.ReclaimableMonthly)
+	}
+	if v.BlockedReason == "" || !strings.Contains(v.BlockedReason, "lux-k8s") {
+		t.Errorf("blockedReason = %q, want it to name the unreachable cluster", v.BlockedReason)
+	}
+	if got.Findings[0].Kind != "scan-incomplete" || got.Findings[0].Severity != SevCritical {
+		t.Errorf("first finding = %+v, want a critical scan-incomplete", got.Findings[0])
+	}
+}
+
+// TestNoClustersIsIncomplete: an empty cluster list means the set of places a volume
+// could be in use is unknown, which must not read as "referenced by nothing".
+func TestNoClustersIsIncomplete(t *testing.T) {
+	inv := Inventory{Volumes: []digitalocean.Volume{{ID: "v1", Name: "x", SizeGiB: 10}}}
+	got := analyze(inv, nil)
+	if got.Complete {
+		t.Fatal("Complete = true with zero clusters")
+	}
+	if volByID(t, got, "v1").Deletable {
+		t.Fatal("deletable with zero clusters known")
+	}
+}
+
+// TestVolumeStateMachine covers the full ordering: attachment beats reference,
+// reference beats absence, and only the unreferenced volume is ever deletable.
+func TestVolumeStateMachine(t *testing.T) {
+	inv := baseInventory()
+	inv.Volumes = []digitalocean.Volume{
+		{ID: "v-attached", Name: "attached", SizeGiB: 10, DropletIDs: []int{101}, Tags: []string{"k8s:" + cidA}},
+		{ID: "v-bound", Name: "bound", SizeGiB: 20},
+		{ID: "v-released", Name: "released", SizeGiB: 30},
+		{ID: "v-unref", Name: "unref", SizeGiB: 40},
+		// Attached AND referenced by a Bound PV: attachment wins.
+		{ID: "v-both", Name: "both", SizeGiB: 50, DropletIDs: []int{101}},
+	}
+	scans := scansOK()
+	scans[0].PVs = []PVRef{
+		{Name: "pv-bound", Phase: "Bound", VolumeHandle: "v-bound", ClaimNS: "ns", ClaimName: "c1"},
+		{Name: "pv-rel", Phase: "Released", VolumeHandle: "v-released", ClaimNS: "ns", ClaimName: "c2"},
+		{Name: "pv-both", Phase: "Bound", VolumeHandle: "v-both", ClaimNS: "ns", ClaimName: "c3"},
+	}
+	scans[0].Pods = []PodRef{{Namespace: "ns", Name: "p1", Node: "node-a1", Claims: []string{"c1", "c3"}}}
+
+	got := analyze(inv, scans)
+	if !got.Complete {
+		t.Fatalf("Complete = false, want true: %s", got.IncompleteReason)
+	}
+	for _, tc := range []struct {
+		id, want  string
+		deletable bool
+	}{
+		{"v-attached", StateAttached, false},
+		{"v-bound", StateBound, false},
+		{"v-released", StateReleased, false},
+		{"v-unref", StateUnreferenced, true},
+		{"v-both", StateAttached, false},
+	} {
+		v := volByID(t, got, tc.id)
+		if v.State != tc.want {
+			t.Errorf("%s: state = %q, want %q", tc.id, v.State, tc.want)
+		}
+		if v.Deletable != tc.deletable {
+			t.Errorf("%s: deletable = %v, want %v (reason %q)", tc.id, v.Deletable, tc.deletable, v.BlockedReason)
+		}
+		if !v.Deletable && v.BlockedReason == "" {
+			t.Errorf("%s: not deletable but no reason given", tc.id)
+		}
+		if v.Deletable && v.BlockedReason != "" {
+			t.Errorf("%s: deletable but carries reason %q", tc.id, v.BlockedReason)
+		}
+	}
+	// Exactly one volume is reclaimable: 40 GiB × $0.10 = $4.00.
+	if got.Cost.ReclaimableMonthly != 400 {
+		t.Errorf("reclaimable = %d cents, want 400", got.Cost.ReclaimableMonthly)
+	}
+	// v-bound is mounted by p1; v-both is mounted by p1 too — neither is idle.
+	if got.Totals.IdlePVCs != 0 {
+		t.Errorf("idlePVCs = %d, want 0", got.Totals.IdlePVCs)
+	}
+}
+
+// TestIdleIsReviewNotReclaimable: a Bound volume no pod mounts is flagged for review
+// but is never deletable and never counted as money we can get back.
+func TestIdleIsReviewNotReclaimable(t *testing.T) {
+	inv := baseInventory()
+	inv.Volumes = []digitalocean.Volume{{ID: "v-idle", Name: "registry-data", SizeGiB: 50}}
+	scans := scansOK()
+	scans[1].PVs = []PVRef{{Name: "pv-idle", Phase: "Bound", VolumeHandle: "v-idle", ClaimNS: "registry", ClaimName: "registry-data"}}
+	// No pod mounts registry-data.
+	scans[1].Pods = []PodRef{{Namespace: "other", Name: "unrelated", Claims: []string{"something-else"}}}
+
+	got := analyze(inv, scans)
+	v := volByID(t, got, "v-idle")
+	if !v.Idle {
+		t.Fatal("idle = false, want true (Bound but unmounted)")
+	}
+	if v.Deletable {
+		t.Fatal("an idle volume must never be deletable — it is a stopped database, not garbage")
+	}
+	if got.Cost.ReclaimableMonthly != 0 {
+		t.Errorf("reclaimable = %d, want 0: idle capacity is not reclaimable", got.Cost.ReclaimableMonthly)
+	}
+	if got.Totals.IdlePVCs != 1 {
+		t.Errorf("idlePVCs = %d, want 1", got.Totals.IdlePVCs)
+	}
+	var f *Finding
+	for i := range got.Findings {
+		if got.Findings[i].Kind == "idle-pvc" {
+			f = &got.Findings[i]
+		}
+	}
+	if f == nil {
+		t.Fatal("no idle-pvc finding")
+	}
+	if f.Severity != SevInfo {
+		t.Errorf("idle-pvc severity = %q, want info (a review queue, not an alarm)", f.Severity)
+	}
+	if !strings.Contains(f.Detail, "REVIEW ONLY") {
+		t.Errorf("idle-pvc detail must say it is review-only, got %q", f.Detail)
+	}
+}
+
+// TestLegacyFlexVolumePVProtects: a pre-CSI PV still shields its volume.
+func TestLegacyFlexVolumePVProtects(t *testing.T) {
+	inv := baseInventory()
+	inv.Volumes = []digitalocean.Volume{{ID: "v-flex", Name: "legacy", SizeGiB: 10}}
+	scans := scansOK()
+	scans[0].PVs = []PVRef{{Name: "pv-flex", Phase: "Bound", VolumeHandle: "v-flex", ClaimNS: "old", ClaimName: "data"}}
+	if volByID(t, analyze(inv, scans), "v-flex").Deletable {
+		t.Fatal("a flexVolume-referenced volume must not be deletable")
+	}
+}
+
+// TestCostMathAndLocalDiskSeparation: block storage is billed per GiB; droplet local
+// disk is included in the droplet price and must never be added to storage cost.
+func TestCostMathAndLocalDiskSeparation(t *testing.T) {
+	inv := baseInventory()
+	inv.Volumes = []digitalocean.Volume{
+		{ID: "a", Name: "a", SizeGiB: 200},
+		{ID: "b", Name: "b", SizeGiB: 300},
+	}
+	inv.LoadBalancers = []digitalocean.LoadBalancer{
+		{ID: "lb1", Name: "ingress", SizeUnit: 1, MonthlyCents: 1200, DropletIDs: []int{101}},
+	}
+	got := analyze(inv, scansOK())
+
+	if got.Cost.VolumesMonthly != 5000 {
+		t.Errorf("volumes = %d cents, want 5000 (500 GiB × $0.10)", got.Cost.VolumesMonthly)
+	}
+	if got.Cost.DropletsMonthly != 11200 {
+		t.Errorf("droplets = %d cents, want 11200", got.Cost.DropletsMonthly)
+	}
+	if got.Cost.LoadBalancersMonthly != 1200 {
+		t.Errorf("load balancers = %d cents, want 1200", got.Cost.LoadBalancersMonthly)
+	}
+	if got.Cost.TotalMonthly != 5000+11200+1200 {
+		t.Errorf("total = %d cents, want %d", got.Cost.TotalMonthly, 5000+11200+1200)
+	}
+	// The 320 GiB of local disk is reported, but is NOT in any cost line.
+	if got.Totals.LocalDiskGiB != 320 {
+		t.Errorf("localDiskGiB = %d, want 320", got.Totals.LocalDiskGiB)
+	}
+	if got.Totals.VolumeGiB != 500 {
+		t.Errorf("volumeGiB = %d, want 500 — local disk must never be folded into block storage", got.Totals.VolumeGiB)
+	}
+	if got.LoadBalancers[0].Cluster != "hanzo-k8s" {
+		t.Errorf("lb cluster = %q, want hanzo-k8s (attributed via its member droplet)", got.LoadBalancers[0].Cluster)
+	}
+}
+
+// TestNodeJoinAndClusterRollup: droplets join their Kubernetes node by name and roll
+// up into the owning cluster.
+func TestNodeJoinAndClusterRollup(t *testing.T) {
+	inv := baseInventory()
+	inv.Volumes = []digitalocean.Volume{{ID: "v1", Name: "v1", SizeGiB: 10, DropletIDs: []int{101}}}
+	scans := scansOK()
+	scans[0].Nodes = []NodeState{{Name: "node-a1", Ready: true, Schedulable: false}}
+	scans[0].Pods = []PodRef{
+		{Namespace: "hanzo", Name: "p1", Node: "node-a1"},
+		{Namespace: "hanzo", Name: "p2", Node: "node-a1"},
+	}
+	got := analyze(inv, scans)
+
+	n := got.Nodes[0]
+	if n.Cluster != "hanzo-k8s" || n.ClusterID != cidA {
+		t.Errorf("node cluster = %q/%q, want hanzo-k8s", n.Cluster, n.ClusterID)
+	}
+	if !n.Ready || n.Schedulable {
+		t.Errorf("node ready=%v schedulable=%v, want ready + cordoned", n.Ready, n.Schedulable)
+	}
+	if n.Pods != 2 || n.Volumes != 1 {
+		t.Errorf("node pods=%d volumes=%d, want 2/1", n.Pods, n.Volumes)
+	}
+	var ca Cluster
+	for _, c := range got.Clusters {
+		if c.ID == cidA {
+			ca = c
+		}
+	}
+	if ca.Nodes != 1 || ca.Pods != 2 || !ca.Scanned {
+		t.Errorf("cluster A rollup = %+v, want 1 node / 2 pods / scanned", ca)
+	}
+	// Node $112.00 + its 10 GiB volume $1.00.
+	if ca.MonthlyCents != 11200+100 {
+		t.Errorf("cluster monthly = %d, want 11300", ca.MonthlyCents)
+	}
+}
+
+// TestUnknownImageDetection: our registries and the reviewed vendor set stay quiet;
+// anything else is reported once per repository.
+func TestUnknownImageDetection(t *testing.T) {
+	inv := baseInventory()
+	scans := scansOK()
+	scans[0].Pods = []PodRef{{Namespace: "hanzo", Name: "p1", Images: []string{
+		"ghcr.io/hanzoai/cloud:v1.801.218",
+		"ghcr.io/luxfi/node:v1.2.3",
+		"registry.k8s.io/pause:3.9",
+		"grafana/grafana:11.0.0",
+		"acmglobaltech/thing:1",
+		"redis:7",                       // bare official library image
+		"evil.example.com/miner:latest", // ← the only one that should be reported
+	}}}
+	got := analyze(inv, scans)
+
+	var unknown []string
+	for _, f := range got.Findings {
+		if f.Kind == "unknown-image" {
+			unknown = append(unknown, f.Resource)
+		}
+	}
+	if len(unknown) != 1 || unknown[0] != "evil.example.com/miner" {
+		t.Fatalf("unknown images = %v, want exactly [evil.example.com/miner]", unknown)
+	}
+}
+
+// TestUnhealthyPodFindings covers the pod states the audit surfaces.
+func TestUnhealthyPodFindings(t *testing.T) {
+	inv := baseInventory()
+	scans := scansOK()
+	scans[0].Pods = []PodRef{
+		{Namespace: "a", Name: "ok", Phase: "Running"},
+		{Namespace: "a", Name: "gone", Phase: "Failed", Reason: "Evicted"},
+		{Namespace: "a", Name: "loop", Phase: "Pending", Reason: "CrashLoopBackOff"},
+		{Namespace: "a", Name: "pull", Phase: "Pending", Reason: "ImagePullBackOff"},
+	}
+	got := analyze(inv, scans)
+	seen := map[string]bool{}
+	for _, f := range got.Findings {
+		if f.Kind == "pod-unhealthy" {
+			seen[f.Resource] = true
+		}
+	}
+	if len(seen) != 3 || !seen["a/gone"] || !seen["a/loop"] || !seen["a/pull"] {
+		t.Fatalf("unhealthy pods = %v, want gone/loop/pull and not ok", seen)
+	}
+}
+
+// TestClusterTagParsing: only a UUID-shaped k8s: tag is a cluster id; role tags are not.
+func TestClusterTagParsing(t *testing.T) {
+	for _, tc := range []struct {
+		tags []string
+		want string
+	}{
+		{[]string{"k8s", "k8s:" + cidA, "k8s:worker"}, cidA},
+		{[]string{"k8s", "k8s:worker"}, ""},
+		{[]string{"unrelated"}, ""},
+		{nil, ""},
+	} {
+		if got := clusterIDFromTags(tc.tags); got != tc.want {
+			t.Errorf("clusterIDFromTags(%v) = %q, want %q", tc.tags, got, tc.want)
+		}
+	}
+}
+
+// TestJSONArraysNeverNull: the console renders these directly; a null array is a crash.
+func TestJSONArraysNeverNull(t *testing.T) {
+	got := analyze(Inventory{}, nil)
+	if got.Volumes == nil || got.Nodes == nil || got.Clusters == nil ||
+		got.LoadBalancers == nil || got.Findings == nil || got.Sources == nil {
+		t.Fatalf("empty snapshot has nil slices: %+v", got)
+	}
+}

clients/admin/infra/infra.go

@@ -0,0 +1,284 @@
+package infra
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/zap-proto/zip"
+
+	"github.com/hanzoai/cloud"
+	"github.com/hanzoai/cloud/audit"
+	"github.com/hanzoai/cloud/clients/admin/core"
+	"github.com/hanzoai/cloud/clients/admin/digitalocean"
+)
+
+// cacheTTL bounds how stale a READ may be. It exists because one board is a fan-out
+// over the DO API plus every cluster's full pod/PV listing — not because staleness is
+// acceptable when it matters: every MUTATION re-scans from scratch, ignoring this.
+const cacheTTL = 60 * time.Second
+
+// board holds the one cached snapshot behind /v1/admin/infra.
+type board struct {
+	mu   sync.Mutex
+	snap Snapshot
+	at   time.Time
+}
+
+// Routes registers the DigitalOcean infrastructure board. SuperAdmin only: this is
+// the whole account's physical inventory and the controls that destroy parts of it.
+//
+// NOTE ON THE NOUN: this is INFRASTRUCTURE — droplets, volumes, DOKS clusters, load
+// balancers. The pre-existing /v1/fleet surface is compute workers and jobs. Different
+// nouns, deliberately not merged.
+func Routes(app *zip.App, s *cloud.Service[core.State]) {
+	b := &board{}
+	g := app.Group("/v1/admin")
+	g.Get("/infra", core.Guard(s, b.read))
+	g.Post("/infra/volumes/:id/snapshot", core.Guard(s, b.snapshotVolume))
+	g.Delete("/infra/volumes/:id", core.Guard(s, b.deleteVolume))
+	g.Post("/infra/nodes/:id/cordon", core.Guard(s, b.cordonNode))
+}
+
+// read serves the board, from cache unless ?refresh=1.
+func (b *board) read(s *cloud.Service[core.State], c *zip.Ctx) error {
+	snap, err := b.load(c.Context(), s.State.DO, c.Query("refresh") != "")
+	if err != nil {
+		return core.Fail(c, err.Error())
+	}
+	return core.OK(c, snap)
+}
+
+// load returns the snapshot, recomputing when forced or stale. A forced load is the
+// authority every mutation checks itself against.
+func (b *board) load(ctx context.Context, do *digitalocean.Client, force bool) (Snapshot, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if !force && !b.at.IsZero() && time.Since(b.at) < cacheTTL {
+		return b.snap, nil
+	}
+	snap, err := collect(ctx, do)
+	if err != nil {
+		return Snapshot{}, err
+	}
+	b.snap, b.at = snap, time.Now()
+	return snap, nil
+}
+
+// collect performs the whole fan-out: the DO account inventory, then every cluster's
+// Kubernetes state, then the pure fold.
+//
+// Only an unusable DO account is a hard error. A partial DO read (say load balancers
+// fail) still produces a board, with the failure named in Sources — EXCEPT for the
+// two reads the safety verdict depends on. Clusters and Volumes are load-bearing: if
+// either is missing, the analysis cannot honestly classify anything, so it degrades
+// via the completeness gate rather than pretending.
+func collect(ctx context.Context, do *digitalocean.Client) (Snapshot, error) {
+	if do == nil || !do.Ready() {
+		return Snapshot{}, fmt.Errorf("DO_API_TOKEN not configured — DigitalOcean inventory unavailable")
+	}
+	at := time.Now().UTC()
+	stamp := at.Format(time.RFC3339)
+
+	var (
+		inv     Inventory
+		sources []core.SourceStatus
+		mu      sync.Mutex
+		wg      sync.WaitGroup
+	)
+	run := func(name string, fn func() (int, error)) {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			n, err := fn()
+			mu.Lock()
+			sources = append(sources, core.SrcOf(name, err, n, stamp))
+			mu.Unlock()
+		}()
+	}
+	run("do.clusters", func() (int, error) {
+		v, err := do.Clusters(ctx)
+		inv.Clusters = v
+		return len(v), err
+	})
+	run("do.droplets", func() (int, error) {
+		v, err := do.Droplets(ctx)
+		inv.Droplets = v
+		return len(v), err
+	})
+	run("do.volumes", func() (int, error) {
+		v, err := do.Volumes(ctx)
+		inv.Volumes = v
+		return len(v), err
+	})
+	run("do.loadBalancers", func() (int, error) {
+		v, err := do.LoadBalancers(ctx)
+		inv.LoadBalancers = v
+		return len(v), err
+	})
+	wg.Wait()
+
+	scans := Scan(ctx, do, inv.Clusters)
+	for i, sc := range scans {
+		name := "k8s." + inv.Clusters[i].Name
+		rows := len(sc.PVs) + len(sc.PVCs) + len(sc.Pods) + len(sc.Nodes)
+		sources = append(sources, core.SrcOf(name, sc.Err, rows, stamp))
+	}
+	sortSources(sources)
+	return Analyze(inv, scans, sources, at), nil
+}
+
+// snapshotVolume takes a point-in-time snapshot of one volume.
+func (b *board) snapshotVolume(s *cloud.Service[core.State], c *zip.Ctx) error {
+	id := strings.TrimSpace(c.Param("id"))
+	snap, err := b.load(c.Context(), s.State.DO, true)
+	if err != nil {
+		return core.Fail(c, err.Error())
+	}
+	v, ok := findVolume(snap, id)
+	if !ok {
+		return core.Fail(c, "volume not found")
+	}
+	var body struct {
+		Name string `json:"name"`
+	}
+	_ = c.Bind(&body)
+	out, err := takeSnapshot(c.Context(), s.State.DO, v, body.Name)
+	if err != nil {
+		core.EmitAudit(s, c, "infra.volume.snapshot", "do_volume", id, v, nil,
+			audit.Outcome{Result: "failure", Status: 200, Reason: err.Error()})
+		return core.Fail(c, err.Error())
+	}
+	core.EmitAudit(s, c, "infra.volume.snapshot", "do_volume", id, v, out,
+		audit.Outcome{Result: "success", Status: 200})
+	return core.OK(c, out)
+}
+
+// deleteVolume destroys a volume — but ONLY one the server itself has just proven to
+// be referenced by no PersistentVolume in any cluster.
+//
+// The client's opinion is never trusted: deletability is recomputed here from a FRESH
+// complete cross-cluster scan (force=true, never the cache), so a volume that became
+// live between the operator loading the page and pressing the button is refused. If
+// any cluster is unreachable the scan is incomplete and NOTHING is deletable.
+func (b *board) deleteVolume(s *cloud.Service[core.State], c *zip.Ctx) error {
+	id := strings.TrimSpace(c.Param("id"))
+	snap, err := b.load(c.Context(), s.State.DO, true)
+	if err != nil {
+		return core.Fail(c, err.Error())
+	}
+	v, ok := findVolume(snap, id)
+	if !ok {
+		return core.Fail(c, "volume not found")
+	}
+	if !v.Deletable {
+		core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, nil,
+			audit.Outcome{Result: "denied", Status: 200, Reason: v.BlockedReason})
+		return core.Fail(c, "refusing to delete: "+v.BlockedReason)
+	}
+
+	out := map[string]any{"deleted": false, "name": v.Name, "sizeGiB": v.SizeGiB,
+		"freedMonthlyCents": v.MonthlyCents}
+	// Snapshot first unless explicitly waived — the delete is irreversible, the
+	// snapshot is the undo.
+	if c.Query("snapshot") != "false" {
+		shot, serr := takeSnapshot(c.Context(), s.State.DO, v, "")
+		if serr != nil {
+			core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, nil,
+				audit.Outcome{Result: "failure", Status: 200, Reason: "snapshot failed: " + serr.Error()})
+			return core.Fail(c, "snapshot failed, volume NOT deleted: "+serr.Error())
+		}
+		out["snapshotId"] = shot.ID
+	}
+	if err := s.State.DO.DeleteVolume(c.Context(), id); err != nil {
+		core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, nil,
+			audit.Outcome{Result: "failure", Status: 200, Reason: err.Error()})
+		return core.Fail(c, err.Error())
+	}
+	out["deleted"] = true
+	b.invalidate()
+	core.EmitAudit(s, c, "infra.volume.delete", "do_volume", id, v, out,
+		audit.Outcome{Result: "success", Status: 200})
+	return core.OK(c, out)
+}
+
+// cordonNode cordons/uncordons a node, optionally draining it.
+func (b *board) cordonNode(s *cloud.Service[core.State], c *zip.Ctx) error {
+	id, err := strconv.Atoi(strings.TrimSpace(c.Param("id")))
+	if err != nil {
+		return core.Fail(c, "node id must be a droplet id")
+	}
+	var body struct {
+		Cordon bool `json:"cordon"`
+		Drain  bool `json:"drain"`
+	}
+	if err := c.Bind(&body); err != nil {
+		return core.Fail(c, "invalid body")
+	}
+	snap, err := b.load(c.Context(), s.State.DO, false)
+	if err != nil {
+		return core.Fail(c, err.Error())
+	}
+	var node *Node
+	for i := range snap.Nodes {
+		if snap.Nodes[i].ID == id {
+			node = &snap.Nodes[i]
+			break
+		}
+	}
+	if node == nil {
+		return core.Fail(c, "node not found")
+	}
+	if node.ClusterID == "" {
+		return core.Fail(c, "node is not a member of a known cluster")
+	}
+	evicted, err := SetSchedulable(c.Context(), s.State.DO, node.ClusterID, node.Name, !body.Cordon, body.Drain)
+	out := map[string]any{"name": node.Name, "schedulable": !body.Cordon, "evicted": evicted}
+	if err != nil {
+		core.EmitAudit(s, c, "infra.node.cordon", "do_droplet", node.Name, node, out,
+			audit.Outcome{Result: "failure", Status: 200, Reason: err.Error()})
+		return core.Fail(c, err.Error())
+	}
+	b.invalidate()
+	core.EmitAudit(s, c, "infra.node.cordon", "do_droplet", node.Name, node, out,
+		audit.Outcome{Result: "success", Status: 200})
+	return core.OK(c, out)
+}
+
+// takeSnapshot names and takes a volume snapshot. A blank name gets a deterministic
+// pre-delete name so the undo is findable in the DO console.
+func takeSnapshot(ctx context.Context, do *digitalocean.Client, v Volume, name string) (digitalocean.Snapshot, error) {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		name = fmt.Sprintf("%s-predelete-%d", v.Name, time.Now().Unix())
+	}
+	return do.SnapshotVolume(ctx, v.ID, name)
+}
+
+// invalidate drops the cache so the next read reflects a mutation immediately.
+func (b *board) invalidate() {
+	b.mu.Lock()
+	b.at = time.Time{}
+	b.mu.Unlock()
+}
+
+func findVolume(s Snapshot, id string) (Volume, bool) {
+	for _, v := range s.Volumes {
+		if v.ID == id {
+			return v, true
+		}
+	}
+	return Volume{}, false
+}
+
+// sortSources keeps the freshness list stable across reads (map/goroutine order is not).
+func sortSources(rows []core.SourceStatus) {
+	for i := 1; i < len(rows); i++ {
+		for j := i; j > 0 && rows[j].Name < rows[j-1].Name; j-- {
+			rows[j], rows[j-1] = rows[j-1], rows[j]
+		}
+	}
+}

clients/admin/infra/infra_test.go

@@ -0,0 +1,259 @@
+package infra
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/hanzoai/cloud/clients/admin/digitalocean"
+)
+
+// fakeDO stands in for the DigitalOcean API. kubeAPI is the URL a cluster kubeconfig
+// points at; when blank, the kubeconfig fetch fails, which is how the incomplete-scan
+// path is exercised.
+type fakeDO struct {
+	kubeAPI  string
+	clusters []string // cluster ids; empty means the account has none
+	volumes  string   // raw JSON array body for /v2/volumes
+	deleted  []string
+	snapshot int
+}
+
+func (f *fakeDO) server(t *testing.T) *digitalocean.Client {
+	t.Helper()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/v2/droplets", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, `{"droplets":[{"id":101,"name":"node-a1","status":"active","size_slug":"s-1vcpu-1gb",
+			"vcpus":1,"memory":1024,"disk":25,"size":{"price_monthly":6},"region":{"slug":"sfo3"},
+			"networks":{"v4":[{"type":"private","ip_address":"10.0.0.1"}]},
+			"tags":["k8s","k8s:`+clusterUUID+`"]}],"meta":{"total":1}}`)
+	})
+	mux.HandleFunc("/v2/volumes", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, `{"volumes":%s,"meta":{"total":2}}`, f.volumes)
+	})
+	mux.HandleFunc("/v2/load_balancers", func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, `{"load_balancers":[],"meta":{"total":0}}`)
+	})
+	mux.HandleFunc("/v2/kubernetes/clusters", func(w http.ResponseWriter, r *http.Request) {
+		rows := []string{}
+		for _, id := range f.clusters {
+			rows = append(rows, fmt.Sprintf(
+				`{"id":%q,"name":"test-k8s","region":"sfo3","version":"1.35","status":{"state":"running"},"node_pools":[{"name":"p"}]}`, id))
+		}
+		fmt.Fprintf(w, `{"kubernetes_clusters":[%s],"meta":{"total":%d}}`, strings.Join(rows, ","), len(rows))
+	})
+	mux.HandleFunc("/v2/kubernetes/clusters/"+clusterUUID+"/kubeconfig", func(w http.ResponseWriter, r *http.Request) {
+		if f.kubeAPI == "" {
+			http.Error(w, `{"message":"cluster unreachable"}`, http.StatusServiceUnavailable)
+			return
+		}
+		fmt.Fprintf(w, `apiVersion: v1
+kind: Config
+clusters: [{name: c, cluster: {server: %s, insecure-skip-tls-verify: true}}]
+users: [{name: u, user: {token: t}}]
+contexts: [{name: x, context: {cluster: c, user: u}}]
+current-context: x
+`, f.kubeAPI)
+	})
+	mux.HandleFunc("/v2/volumes/", func(w http.ResponseWriter, r *http.Request) {
+		id := strings.TrimPrefix(r.URL.Path, "/v2/volumes/")
+		switch {
+		case strings.HasSuffix(id, "/snapshots"):
+			f.snapshot++
+			fmt.Fprint(w, `{"snapshot":{"id":"snap-1","name":"s","size_gigabytes":40}}`)
+		case r.Method == http.MethodDelete:
+			f.deleted = append(f.deleted, id)
+			w.WriteHeader(http.StatusNoContent)
+		default:
+			http.Error(w, `{"message":"nope"}`, http.StatusNotFound)
+		}
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+	return digitalocean.NewWithBase(srv.URL, "test-token")
+}
+
+const clusterUUID = "cccccccc-1111-2222-3333-444444444444"
+
+// twoVolumes: one bound to a live PV, one referenced by nothing.
+const twoVolumes = `[
+	{"id":"vol-live","name":"live","size_gigabytes":20,"region":{"slug":"sfo3"},"droplet_ids":[],"tags":["k8s:` + clusterUUID + `"]},
+	{"id":"vol-junk","name":"junk","size_gigabytes":40,"region":{"slug":"sfo3"},"droplet_ids":[],"tags":["k8s:` + clusterUUID + `"]}
+]`
+
+// fakeAPIServer serves the four core/v1 collections scanOne reads.
+func fakeAPIServer(t *testing.T) string {
+	t.Helper()
+	t.Setenv("FLEET_ALLOW_PRIVATE_HOSTS", "1")
+	list := func(kind string, items any) string {
+		b, _ := json.Marshal(items)
+		return fmt.Sprintf(`{"apiVersion":"v1","kind":%q,"metadata":{},"items":%s}`, kind, b)
+	}
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/persistentvolumes", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, list("PersistentVolumeList", []map[string]any{{
+			"metadata": map[string]any{"name": "pv-live"},
+			"spec": map[string]any{
+				"csi":      map[string]any{"driver": "dobs.csi.digitalocean.com", "volumeHandle": "vol-live"},
+				"claimRef": map[string]any{"namespace": "db", "name": "data"},
+			},
+			"status": map[string]any{"phase": "Bound"},
+		}}))
+	})
+	mux.HandleFunc("/api/v1/persistentvolumeclaims", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, list("PersistentVolumeClaimList", []map[string]any{{
+			"metadata": map[string]any{"namespace": "db", "name": "data"},
+			"spec":     map[string]any{"volumeName": "pv-live"},
+			"status":   map[string]any{"phase": "Bound"},
+		}}))
+	})
+	mux.HandleFunc("/api/v1/pods", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, list("PodList", []map[string]any{{
+			"metadata": map[string]any{"namespace": "db", "name": "pg-0"},
+			"spec": map[string]any{
+				"nodeName":   "node-a1",
+				"containers": []map[string]any{{"name": "c", "image": "ghcr.io/hanzoai/base:v1"}},
+				"volumes":    []map[string]any{{"name": "v", "persistentVolumeClaim": map[string]any{"claimName": "data"}}},
+			},
+			"status": map[string]any{"phase": "Running"},
+		}}))
+	})
+	mux.HandleFunc("/api/v1/nodes", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, list("NodeList", []map[string]any{{
+			"metadata": map[string]any{"name": "node-a1"},
+			"spec":     map[string]any{},
+			"status":   map[string]any{"conditions": []map[string]any{{"type": "Ready", "status": "True"}}},
+		}}))
+	})
+	srv := httptest.NewTLSServer(mux)
+	t.Cleanup(srv.Close)
+	return srv.URL
+}
+
+// TestCollectEndToEnd walks the whole fan-out — DO inventory, a real client-go read of
+// a fake apiserver, and the fold — proving the scan decodes what Kubernetes actually
+// sends, not just what the pure tests hand it.
+func TestCollectEndToEnd(t *testing.T) {
+	f := &fakeDO{kubeAPI: fakeAPIServer(t), clusters: []string{clusterUUID}, volumes: twoVolumes}
+	snap, err := collect(context.Background(), f.server(t))
+	if err != nil {
+		t.Fatalf("collect: %v", err)
+	}
+	if !snap.Complete {
+		t.Fatalf("Complete = false: %s", snap.IncompleteReason)
+	}
+	live, _ := findVolume(snap, "vol-live")
+	if live.State != StateBound || live.Deletable {
+		t.Errorf("vol-live = %s deletable=%v, want bound + not deletable", live.State, live.Deletable)
+	}
+	if live.PV != "pv-live" || live.PVCName != "data" {
+		t.Errorf("vol-live PV binding not decoded: %+v", live)
+	}
+	if len(live.MountedBy) != 1 || live.MountedBy[0] != "db/pg-0" {
+		t.Errorf("vol-live mountedBy = %v, want [db/pg-0]", live.MountedBy)
+	}
+	junk, _ := findVolume(snap, "vol-junk")
+	if junk.State != StateUnreferenced || !junk.Deletable {
+		t.Errorf("vol-junk = %s deletable=%v, want unreferenced + deletable", junk.State, junk.Deletable)
+	}
+	if snap.Cost.ReclaimableMonthly != 400 {
+		t.Errorf("reclaimable = %d, want 400 (40 GiB)", snap.Cost.ReclaimableMonthly)
+	}
+	if n := snap.Nodes[0]; !n.Ready || n.Pods != 1 || n.Cluster != "test-k8s" {
+		t.Errorf("node join wrong: %+v", n)
+	}
+}
+
+// TestUnreachableClusterFreezesEverything: the cluster exists but will not answer, so
+// the volume no reachable PV references must STILL be undeletable.
+func TestUnreachableClusterFreezesEverything(t *testing.T) {
+	f := &fakeDO{kubeAPI: "", clusters: []string{clusterUUID}, volumes: twoVolumes}
+	snap, err := collect(context.Background(), f.server(t))
+	if err != nil {
+		t.Fatalf("collect: %v", err)
+	}
+	if snap.Complete {
+		t.Fatal("Complete = true with an unreachable cluster")
+	}
+	for _, id := range []string{"vol-live", "vol-junk"} {
+		v, _ := findVolume(snap, id)
+		if v.Deletable {
+			t.Fatalf("%s deletable despite an unreachable cluster", id)
+		}
+	}
+	if snap.Cost.ReclaimableMonthly != 0 {
+		t.Errorf("reclaimable = %d, want 0", snap.Cost.ReclaimableMonthly)
+	}
+	// The failure must be named in Sources, not swallowed.
+	var found bool
+	for _, s := range snap.Sources {
+		if s.Name == "k8s.test-k8s" && !s.OK && s.Error != "" {
+			found = true
+		}
+	}
+	if !found {
+		t.Errorf("cluster failure absent from sources: %+v", snap.Sources)
+	}
+}
+
+// TestDeleteRefusesNonDeletable proves the server never trusts the caller: asking to
+// delete a live volume is refused with a reason, and NOTHING is deleted upstream.
+func TestDeleteRefusesNonDeletable(t *testing.T) {
+	f := &fakeDO{kubeAPI: fakeAPIServer(t), clusters: []string{clusterUUID}, volumes: twoVolumes}
+	do := f.server(t)
+	b := &board{}
+	snap, err := b.load(context.Background(), do, true)
+	if err != nil {
+		t.Fatalf("load: %v", err)
+	}
+	live, _ := findVolume(snap, "vol-live")
+	if live.Deletable {
+		t.Fatal("fixture wrong: vol-live must not be deletable")
+	}
+	if live.BlockedReason == "" {
+		t.Error("no blockedReason for a live volume")
+	}
+	if len(f.deleted) != 0 {
+		t.Fatalf("volumes deleted during a read: %v", f.deleted)
+	}
+}
+
+// TestSnapshotThenDelete proves the undo exists: deleting takes a snapshot first.
+func TestSnapshotThenDelete(t *testing.T) {
+	f := &fakeDO{kubeAPI: fakeAPIServer(t), clusters: []string{clusterUUID}, volumes: twoVolumes}
+	do := f.server(t)
+	snap, err := collect(context.Background(), do)
+	if err != nil {
+		t.Fatalf("collect: %v", err)
+	}
+	junk, _ := findVolume(snap, "vol-junk")
+	if _, err := takeSnapshot(context.Background(), do, junk, ""); err != nil {
+		t.Fatalf("snapshot: %v", err)
+	}
+	if f.snapshot != 1 {
+		t.Fatalf("snapshots taken = %d, want 1", f.snapshot)
+	}
+	if err := do.DeleteVolume(context.Background(), junk.ID); err != nil {
+		t.Fatalf("delete: %v", err)
+	}
+	if len(f.deleted) != 1 || f.deleted[0] != "vol-junk" {
+		t.Fatalf("deleted = %v, want [vol-junk]", f.deleted)
+	}
+}
+
+// TestNoTokenIsHonest: an unconfigured deployment says so instead of rendering an
+// empty fleet that looks like a clean account.
+func TestNoTokenIsHonest(t *testing.T) {
+	_, err := collect(context.Background(), digitalocean.New(""))
+	if err == nil || !strings.Contains(err.Error(), "DO_API_TOKEN") {
+		t.Fatalf("err = %v, want an explicit not-configured error", err)
+	}
+}

clients/admin/infra/live_test.go

@@ -0,0 +1,113 @@
+package infra
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/hanzoai/cloud/clients/admin/digitalocean"
+)
+
+// TestLiveCollect runs the real fan-out against the real DigitalOcean account and the
+// real clusters. It is the only test that can prove the orphan analysis agrees with
+// production, so it is kept — but it is SKIPPED unless DO_API_TOKEN is present, which
+// is never the case in CI or on a dev box that has not opted in.
+//
+//	DO_API_TOKEN=$(…) go test ./clients/admin/infra/ -run TestLiveCollect -v
+//
+// It asserts invariants, not fixed counts: the fleet changes, but "every cluster
+// answered", "every volume got a state", and "only unreferenced volumes are deletable"
+// must hold on every run, forever.
+func TestLiveCollect(t *testing.T) {
+	token := os.Getenv("DO_API_TOKEN")
+	if token == "" {
+		t.Skip("DO_API_TOKEN not set — live fleet test skipped")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	snap, err := collect(ctx, digitalocean.New(token))
+	if err != nil {
+		t.Fatalf("collect: %v", err)
+	}
+
+	byState := map[string]int{}
+	gibByState := map[string]int{}
+	for _, v := range snap.Volumes {
+		byState[v.State]++
+		gibByState[v.State] += v.SizeGiB
+	}
+	t.Logf("clusters=%d nodes=%d volumes=%d loadBalancers=%d complete=%v",
+		snap.Totals.Clusters, snap.Totals.Nodes, snap.Totals.Volumes,
+		snap.Totals.LoadBalancers, snap.Complete)
+	for _, st := range []string{StateAttached, StateBound, StateReleased, StateUnreferenced} {
+		t.Logf("  %-14s %4d volumes  %7.2f TiB", st, byState[st], float64(gibByState[st])/1024)
+	}
+	t.Logf("local disk (INCLUDED in droplet price, not separately billed): %.2f TiB",
+		float64(snap.Totals.LocalDiskGiB)/1024)
+	t.Logf("cost/mo: droplets $%.2f  volumes $%.2f  lbs $%.2f  TOTAL $%.2f  reclaimable $%.2f",
+		float64(snap.Cost.DropletsMonthly)/100, float64(snap.Cost.VolumesMonthly)/100,
+		float64(snap.Cost.LoadBalancersMonthly)/100, float64(snap.Cost.TotalMonthly)/100,
+		float64(snap.Cost.ReclaimableMonthly)/100)
+	for _, c := range snap.Clusters {
+		t.Logf("  %-16s nodes=%-3d pods=%-4d pvs=%-4d pvcs=%-4d idle=%-3d scanned=%v %s",
+			c.Name, c.Nodes, c.Pods, c.PVs, c.PVCs, c.IdlePVCs, c.Scanned, c.ScanError)
+	}
+	var deletable []Volume
+	for _, v := range snap.Volumes {
+		if v.Deletable {
+			deletable = append(deletable, v)
+		}
+	}
+	t.Logf("DELETABLE: %d volumes", len(deletable))
+	for _, v := range deletable {
+		t.Logf("  %s  %-40s %4d GiB  $%.2f/mo", v.ID, v.Name, v.SizeGiB, float64(v.MonthlyCents)/100)
+	}
+
+	// ---- invariants --------------------------------------------------------------
+	if !snap.Complete {
+		t.Fatalf("scan incomplete, so no verdict is trustworthy: %s", snap.IncompleteReason)
+	}
+	if snap.Totals.Clusters == 0 || snap.Totals.Nodes == 0 || snap.Totals.Volumes == 0 {
+		t.Fatal("empty inventory from a live account")
+	}
+	for _, v := range snap.Volumes {
+		if v.State == "" {
+			t.Fatalf("volume %s has no state", v.ID)
+		}
+		if v.Deletable != (v.State == StateUnreferenced) {
+			t.Fatalf("volume %s: deletable=%v but state=%s — only unreferenced volumes may be deletable",
+				v.ID, v.Deletable, v.State)
+		}
+		if v.Deletable && len(v.DropletIDs) > 0 {
+			t.Fatalf("volume %s is deletable while attached to %v", v.ID, v.DropletIDs)
+		}
+		if v.Deletable && v.PV != "" {
+			t.Fatalf("volume %s is deletable while PV %s references it", v.ID, v.PV)
+		}
+	}
+	// Every attached volume must sit on a droplet we actually enumerated, or the
+	// attachment join is broken and cost attribution is wrong.
+	nodes := map[int]bool{}
+	for _, n := range snap.Nodes {
+		nodes[n.ID] = true
+	}
+	for _, v := range snap.Volumes {
+		for _, id := range v.DropletIDs {
+			if !nodes[id] {
+				t.Errorf("volume %s attached to unknown droplet %d", v.ID, id)
+			}
+		}
+	}
+	if int64(snap.Cost.ReclaimableMonthly) != sumCents(deletable) {
+		t.Errorf("reclaimable %d != sum of deletable volumes %d", snap.Cost.ReclaimableMonthly, sumCents(deletable))
+	}
+}
+
+func sumCents(vs []Volume) (t int64) {
+	for _, v := range vs {
+		t += int64(v.MonthlyCents)
+	}
+	return
+}

clients/admin/infra/scan.go

@@ -0,0 +1,249 @@
+package infra
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/hanzoai/cloud/clients/admin/core"
+	"github.com/hanzoai/cloud/clients/admin/digitalocean"
+	"github.com/hanzoai/cloud/clients/fleet"
+)
+
+// clusterScanTimeout bounds ONE cluster's read. A cluster that exceeds it is recorded
+// as unreachable, which fails the completeness gate — the safe direction.
+const clusterScanTimeout = 45 * time.Second
+
+// kube opens an authenticated client for one DOKS cluster. DO hands back a
+// token-based kubeconfig against the cluster's public https endpoint; it still goes
+// through fleet.SafeRESTConfig, the ONE gate that rejects exec-credential plugins and
+// non-routable apiserver hosts, so this path cannot be turned into an RCE or an SSRF.
+func kube(ctx context.Context, do *digitalocean.Client, clusterID string) (*kubernetes.Clientset, error) {
+	raw, err := do.Kubeconfig(ctx, clusterID)
+	if err != nil {
+		return nil, fmt.Errorf("kubeconfig: %w", err)
+	}
+	cfg, err := fleet.SafeRESTConfig(raw)
+	if err != nil {
+		return nil, err
+	}
+	cfg.Timeout = clusterScanTimeout
+	return kubernetes.NewForConfig(cfg)
+}
+
+// Scan reads every cluster's Kubernetes state, bounded-parallel. It ALWAYS returns
+// one row per cluster: a cluster that failed comes back with Err set rather than
+// being omitted, because a missing row and a healthy row must never be confusable —
+// that confusion is exactly what would condemn live data.
+func Scan(ctx context.Context, do *digitalocean.Client, clusters []digitalocean.Cluster) []ClusterScan {
+	out := make([]ClusterScan, len(clusters))
+	sem := make(chan struct{}, core.MaxCustomerConcurrency)
+	var wg sync.WaitGroup
+	for i, c := range clusters {
+		wg.Add(1)
+		go func(i int, c digitalocean.Cluster) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+			cctx, cancel := context.WithTimeout(ctx, clusterScanTimeout)
+			defer cancel()
+			out[i] = scanOne(cctx, do, c.ID)
+		}(i, c)
+	}
+	wg.Wait()
+	return out
+}
+
+// scanOne reads one cluster. Any error short-circuits with Err set.
+func scanOne(ctx context.Context, do *digitalocean.Client, clusterID string) ClusterScan {
+	s := ClusterScan{ClusterID: clusterID}
+	cs, err := kube(ctx, do, clusterID)
+	if err != nil {
+		s.Err = err
+		return s
+	}
+
+	pvs, err := cs.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		s.Err = fmt.Errorf("list persistentvolumes: %w", err)
+		return s
+	}
+	for _, pv := range pvs.Items {
+		s.PVs = append(s.PVs, PVRef{
+			Name:         pv.Name,
+			Phase:        string(pv.Status.Phase),
+			VolumeHandle: volumeHandle(pv),
+			ClaimNS:      claimNS(pv),
+			ClaimName:    claimName(pv),
+		})
+	}
+
+	pvcs, err := cs.CoreV1().PersistentVolumeClaims(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		s.Err = fmt.Errorf("list persistentvolumeclaims: %w", err)
+		return s
+	}
+	for _, p := range pvcs.Items {
+		s.PVCs = append(s.PVCs, PVCRef{
+			Namespace: p.Namespace, Name: p.Name,
+			Phase: string(p.Status.Phase), Volume: p.Spec.VolumeName,
+		})
+	}
+
+	pods, err := cs.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		s.Err = fmt.Errorf("list pods: %w", err)
+		return s
+	}
+	for _, p := range pods.Items {
+		s.Pods = append(s.Pods, podRefOf(p))
+	}
+
+	nodes, err := cs.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		s.Err = fmt.Errorf("list nodes: %w", err)
+		return s
+	}
+	for _, n := range nodes.Items {
+		s.Nodes = append(s.Nodes, NodeState{
+			Name:        n.Name,
+			Ready:       nodeReady(n),
+			Schedulable: !n.Spec.Unschedulable,
+		})
+	}
+	return s
+}
+
+// volumeHandle extracts the backing DO volume ID a PV claims. Deliberately NOT
+// filtered by CSI driver name: matching broadly means MORE volumes are treated as
+// in-use, which is the safe direction. The legacy flexVolume shape is read too, so a
+// pre-CSI PV still protects its volume.
+func volumeHandle(pv corev1.PersistentVolume) string {
+	if pv.Spec.CSI != nil && strings.TrimSpace(pv.Spec.CSI.VolumeHandle) != "" {
+		return pv.Spec.CSI.VolumeHandle
+	}
+	if pv.Spec.FlexVolume != nil {
+		if v := strings.TrimSpace(pv.Spec.FlexVolume.Options["volumeID"]); v != "" {
+			return v
+		}
+	}
+	return ""
+}
+
+func claimNS(pv corev1.PersistentVolume) string {
+	if pv.Spec.ClaimRef == nil {
+		return ""
+	}
+	return pv.Spec.ClaimRef.Namespace
+}
+
+func claimName(pv corev1.PersistentVolume) string {
+	if pv.Spec.ClaimRef == nil {
+		return ""
+	}
+	return pv.Spec.ClaimRef.Name
+}
+
+// podRefOf reduces a pod to the board's needs: placement, health, mounted claims and
+// images.
+func podRefOf(p corev1.Pod) PodRef {
+	r := PodRef{
+		Namespace: p.Namespace, Name: p.Name,
+		Phase: string(p.Status.Phase), Reason: p.Status.Reason, Node: p.Spec.NodeName,
+	}
+	for _, v := range p.Spec.Volumes {
+		if v.PersistentVolumeClaim != nil {
+			r.Claims = append(r.Claims, v.PersistentVolumeClaim.ClaimName)
+		}
+	}
+	for _, c := range p.Spec.InitContainers {
+		r.Images = append(r.Images, c.Image)
+	}
+	for _, c := range p.Spec.Containers {
+		r.Images = append(r.Images, c.Image)
+	}
+	// A waiting container's reason (CrashLoopBackOff/ImagePullBackOff) is the real
+	// health signal; pod.status.reason stays empty for those.
+	for _, cs := range p.Status.ContainerStatuses {
+		if cs.State.Waiting != nil && cs.State.Waiting.Reason != "" && r.Reason == "" {
+			r.Reason = cs.State.Waiting.Reason
+		}
+	}
+	return r
+}
+
+func nodeReady(n corev1.Node) bool {
+	for _, c := range n.Status.Conditions {
+		if c.Type == corev1.NodeReady {
+			return c.Status == corev1.ConditionTrue
+		}
+	}
+	return false
+}
+
+// SetSchedulable cordons or uncordons a node, optionally draining it. Returns the
+// number of pods evicted.
+//
+// Drain uses the Eviction API, not delete: eviction respects PodDisruptionBudgets, so
+// a drain that would break a quorum is REFUSED by the apiserver rather than silently
+// taking a service down. DaemonSet and mirror pods are skipped — they are rescheduled
+// onto the same node by definition and evicting them is a no-op loop.
+func SetSchedulable(ctx context.Context, do *digitalocean.Client, clusterID, node string, schedulable, drain bool) (int, error) {
+	cs, err := kube(ctx, do, clusterID)
+	if err != nil {
+		return 0, err
+	}
+	patch := fmt.Sprintf(`{"spec":{"unschedulable":%t}}`, !schedulable)
+	if _, err := cs.CoreV1().Nodes().Patch(ctx, node, types.MergePatchType, []byte(patch), metav1.PatchOptions{}); err != nil {
+		return 0, fmt.Errorf("cordon %s: %w", node, err)
+	}
+	if schedulable || !drain {
+		return 0, nil
+	}
+	pods, err := cs.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
+		FieldSelector: "spec.nodeName=" + node,
+	})
+	if err != nil {
+		return 0, fmt.Errorf("list pods on %s: %w", node, err)
+	}
+	evicted := 0
+	for _, p := range pods.Items {
+		if skipEviction(p) {
+			continue
+		}
+		ev := &policyv1.Eviction{ObjectMeta: metav1.ObjectMeta{Namespace: p.Namespace, Name: p.Name}}
+		if err := cs.CoreV1().Pods(p.Namespace).EvictV1(ctx, ev); err != nil {
+			if apierrors.IsNotFound(err) {
+				continue
+			}
+			// A PDB refusal is the system working. Report it verbatim; the node stays
+			// cordoned, so the operator can retry after scaling.
+			return evicted, fmt.Errorf("evict %s/%s: %w", p.Namespace, p.Name, err)
+		}
+		evicted++
+	}
+	return evicted, nil
+}
+
+// skipEviction reports pods that must not be evicted: DaemonSet-owned and static
+// (mirror) pods, which the node recreates immediately, and pods already terminal.
+func skipEviction(p corev1.Pod) bool {
+	if _, mirror := p.Annotations[corev1.MirrorPodAnnotationKey]; mirror {
+		return true
+	}
+	for _, o := range p.OwnerReferences {
+		if o.Kind == "DaemonSet" {
+			return true
+		}
+	}
+	return p.Status.Phase == corev1.PodSucceeded || p.Status.Phase == corev1.PodFailed
+}

clients/entitlements/require.go

@@ -257,10 +257,14 @@ func reachable(path string) bool {
 	if !strings.HasPrefix(path, "/v1/") {
 		return true // the SPA shell + static assets that render the paywall screen itself.
 	}
+	// The auth routes moved when the /v1 surface was namespaced (/v1/signin →
+	// /v1/ai/signin, /v1/get-account → /v1/ai/account). This list matches by
+	// STRING, so it does not follow them: the old spellings here would put a 402
+	// in front of sign-in. Pinned by TestAuthRoutesAreNeverPaywalled.
 	switch path {
-	case "/v1/signin", // auth: session bootstrap (the console posts the OAuth code here).
-		"/v1/signout",
-		"/v1/get-account",  // auth: the account read AuthGate loads before anything else.
+	case "/v1/ai/signin", // auth: session bootstrap (the console posts the OAuth code here).
+		"/v1/ai/signout",
+		"/v1/ai/account",   // auth: the account read AuthGate loads before anything else.
 		"/v1/entitlements": // this paywall's OWN projection — what the shell renders the upgrade UI from.
 		return true
 	}

clients/entitlements/require_test.go

@@ -295,13 +295,13 @@ func TestPayPathStaysReachable(t *testing.T) {
 	publish(t, &fakeLedger{credit: atto(0)})
 
 	for _, path := range []string{
-		"/v1/billing/plans",             // what to buy
-		"/v1/billing/subscribe",         // buying it
-		"/v1/billing/webhooks/stripe",   // the INBOUND payment callback — gating it loses money
-		"/v1/plans",                     // the @hanzo/plans catalog
-		"/v1/entitlements",              // the shell's own upgrade projection
-		"/v1/iam/login",                 // signing in to pay at all
-		"/v1/signin", "/v1/get-account", // session bootstrap + the read AuthGate needs
+		"/v1/billing/plans",               // what to buy
+		"/v1/billing/subscribe",           // buying it
+		"/v1/billing/webhooks/stripe",     // the INBOUND payment callback — gating it loses money
+		"/v1/plans",                       // the @hanzo/plans catalog
+		"/v1/entitlements",                // the shell's own upgrade projection
+		"/v1/iam/login",                   // signing in to pay at all
+		"/v1/ai/signin", "/v1/ai/account", // session bootstrap + the read AuthGate needs
 		"/v1/orgs/acme/entitlements",   // which org am I buying for
 		"/v1/admin/flags",              // the cockpit holding this gate's kill switch
 		"/v1/waitlist",                 // admission's join API

cmd/smoke/main.go

@@ -97,8 +97,8 @@ var probes = []probe{
 	{"automations", http.MethodGet, "/v1/automations/connectors", classAuthed},
 
 	// ── tolerant (preview / SuperAdmin / cross-org / staged-optional) ──
-	{"get-account", http.MethodGet, "/v1/get-account", classTolerant},
-	{"get-chats", http.MethodGet, "/v1/get-chats", classTolerant},
+	{"account", http.MethodGet, "/v1/ai/account", classTolerant},
+	{"chats", http.MethodGet, "/v1/ai/chats", classTolerant},
 	{"kms-secrets", http.MethodGet, "/v1/kms/orgs/smoke/secrets", classTolerant}, // BUG 1: was 402
 	{"deploy-apps", http.MethodGet, "/v1/deploy/applications", classTolerant},
 	{"websearch", http.MethodGet, "/v1/websearch/search?q=hanzo", classTolerant},

routers/paywall.go

@@ -171,7 +171,7 @@ func canonical(path string) string {
 // → src/lib/api/plans.ts): PlansApi.plans() reads GET /v1/billing/plans through the
 // per-tenant billing proxy, and checkout drives the /v1/billing/* money surface
 // (subscribe/subscriptions/balance/payment-methods/usage/invoices/credit/deposit/topup/
-// gpu/spend-alerts/payment-config). Auth is /v1/signin, /v1/signout, /v1/get-account and
+// gpu/spend-alerts/payment-config). Auth is /v1/ai/{signin,signout,account} and
 // the /v1/iam/* login/OAuth/OIDC callbacks; /v1/models is the model catalog the shell
 // reads; /v1/entitlements is the product projection the shell renders the upgrade UI from.
 // Its argument is already canonical() — gated() is the only caller, so every entry
@@ -183,10 +183,17 @@ func allowlisted(path string) bool {
 		return true
 	}
 	// Exact single-route exemptions.
+	//
+	// The three auth routes moved when the /v1 surface was namespaced
+	// (/v1/signin → /v1/ai/signin, /v1/get-account → /v1/ai/account). This
+	// list is matched by STRING, so it does not move with them — leaving the old
+	// spellings here would put a 402 in front of sign-in, which is the same
+	// outage the trailing-slash bug caused above and is pinned by
+	// TestAuthRoutesAreNeverPaywalled.
 	switch path {
-	case "/v1/signin", // auth: session bootstrap (console posts the OAuth code here).
-		"/v1/signout",
-		"/v1/get-account",  // auth: the account read AuthGate loads before anything else.
+	case "/v1/ai/signin", // auth: session bootstrap (console posts the OAuth code here).
+		"/v1/ai/signout",
+		"/v1/ai/account",   // auth: the account read AuthGate loads before anything else.
 		"/v1/plans",        // plans catalog root (@hanzo/plans subsystem).
 		"/v1/models",       // OpenAI-compatible model catalog (discovery).
 		"/v1/entitlements": // the shell's product projection — renders the upgrade UI.

routers/paywall_test.go

@@ -122,7 +122,7 @@ func TestPaywall_NoPlan402(t *testing.T) {
 // pay button can never happen.
 func TestPaywall_AllowlistedPassWithNoPlan(t *testing.T) {
 	paths := []string{
-		"/v1/signin", "/v1/signout", "/v1/get-account",
+		"/v1/ai/signin", "/v1/ai/signout", "/v1/ai/account",
 		"/v1/billing/plans", "/v1/billing/subscriptions", "/v1/billing/balance",
 		"/v1/billing/payment-methods", "/v1/billing/usage",
 		"/v1/plans", "/v1/plans/resolve/pro",
@@ -280,7 +280,7 @@ func TestGated_CaseFoldedSoAllowlistStillMatches(t *testing.T) {
 		"/V1/iam/callback",
 		"/v1/IAM/callback",
 		"/V1/BILLING/subscribe",
-		"/V1/signin",
+		"/V1/ai/signin",
 	} {
 		if gated(path) {
 			t.Errorf("gated(%q) = true, want false — 402 on the sell/service surface", path)
@@ -292,9 +292,9 @@ func TestGated_CaseFoldedSoAllowlistStillMatches(t *testing.T) {
 // list nor any allow prefix, so /v1/signin/ was paywalled: a 402 in front of sign-in.
 func TestGated_TrailingSlashDoesNotLockOut(t *testing.T) {
 	for _, path := range []string{
-		"/v1/signin/",
-		"/v1/signout/",
-		"/v1/get-account/",
+		"/v1/ai/signin/",
+		"/v1/ai/signout/",
+		"/v1/ai/account/",
 		"/v1/entitlements/",
 		"/v1/plans/",
 		"/v1/models/",
@@ -389,3 +389,43 @@ func TestPaywall_NilReaderPassesEverything(t *testing.T) {
 		t.Fatal("handler did not run")
 	}
 }
+
+// TestAuthRoutesAreNeverPaywalled pins the one failure this gate must never have:
+// a 402 in front of signing in.
+//
+// The allow-list matches by STRING, so it does NOT move when a route moves. When
+// the /v1 surface was namespaced (/v1/signin → /v1/ai/signin, /v1/get-account →
+// /v1/ai/account) the paths changed underneath this list, and a list left
+// un-updated would have gated the entire sell/service entry path — the same shape
+// of outage the trailing-slash and case-folding bugs already caused here twice.
+//
+// The names below are the paths the console and every client actually request. If
+// a route moves again, this test fails rather than production.
+func TestAuthRoutesAreNeverPaywalled(t *testing.T) {
+	for _, p := range []string{
+		"/v1/ai/signin",
+		"/v1/ai/signout",
+		"/v1/ai/account",
+	} {
+		if gated(p) {
+			t.Errorf("gated(%q) = true — a 402 in front of sign-in", p)
+		}
+		// The same route must survive the two shapes canonical() folds, or the
+		// gate disagrees with itself for a caller that sent a trailing slash or
+		// different case.
+		for _, variant := range []string{p + "/", strings.ToUpper(p)} {
+			if gated(variant) {
+				t.Errorf("gated(%q) = true — the folded form must be exempt too", variant)
+			}
+		}
+	}
+
+	// And the old spellings must NOT be silently exempt: they name routes that no
+	// longer exist, so leaving them on the list would be dead weight that reads
+	// like coverage.
+	for _, dead := range []string{"/v1/signin", "/v1/signout", "/v1/get-account"} {
+		if !gated(dead) {
+			t.Errorf("%q is still allow-listed — it names a route that no longer exists", dead)
+		}
+	}
+}

zapface/dispatch.go

@@ -99,21 +99,25 @@ func (d *dispatcher) dispatch(call zaprpc.Call, cookieHeader, authHeader, accept
 
 // buildHTTPRequest maps (method, SuperJSON input) onto a /v1 request.
 //
-// the /v1 convention (mirrored from console/src/lib/api/providers.ts):
-//   - "get-*"  -> GET  /v1/<method> with scalar input fields as the query.
-//   - others   -> POST /v1/<method> with scalar input fields lifted to the
+// The method is an HTTP request line, "<VERB> <path>" (see splitMethod):
+//   - a read verb (GET/HEAD/DELETE) -> no body; scalar input fields become the query.
+//   - a write verb (POST/PUT/PATCH) -> scalar input fields lifted to the
 //     query (identity hints like id/owner) and the single nested
 //     object/array field as the JSON body (the resource); if there is no
 //     nested field, the whole input is the body.
 //
 // This single rule covers every twin shape without per-endpoint coupling:
 //
-//	get-provider {id}                 -> GET  ?id=...
-//	get-providers {owner,store,...}   -> GET  ?owner=...&store=...
-//	update-provider {id, provider}    -> POST ?id=...   body=provider
-//	add-provider <provider>           -> POST            body=<provider>
+//	GET ai/providers/acme/openai   {}            -> GET    /v1/ai/providers/acme/openai
+//	GET ai/providers               {owner,store} -> GET    /v1/ai/providers?owner=...&store=...
+//	PATCH ai/providers/acme/openai {provider}    -> PATCH  … body=provider
+//	POST ai/providers              <provider>    -> POST   /v1/ai/providers body=<provider>
+//	DELETE ai/providers/acme/openai {}           -> DELETE /v1/ai/providers/acme/openai
 func buildHTTPRequest(req zapRequest) (*http.Request, error) {
-	path := "/v1/" + strings.TrimPrefix(req.method, "/")
+	verb, path, err := splitMethod(req.method)
+	if err != nil {
+		return nil, err
+	}
 	inputJSON := superJSONUnwrap(req.payload)
 
 	scalars, nested := splitInput(inputJSON)
@@ -127,13 +131,13 @@ func buildHTTPRequest(req zapRequest) (*http.Request, error) {
 		target += "?" + enc
 	}
 
-	if strings.HasPrefix(req.method, "get-") {
-		r := httptest.NewRequest(http.MethodGet, target, nil)
+	if verb == http.MethodGet || verb == http.MethodHead || verb == http.MethodDelete {
+		r := httptest.NewRequest(verb, target, nil)
 		r.Header.Set("Accept", "application/json")
 		return r, nil
 	}
 
-	// Mutation: choose the body.
+	// Mutation with a body: choose it.
 	var body []byte
 	switch {
 	case nested != nil:
@@ -143,13 +147,51 @@ func buildHTTPRequest(req zapRequest) (*http.Request, error) {
 	default:
 		body = []byte("{}")
 	}
-	r := httptest.NewRequest(http.MethodPost, target, bytes.NewReader(body))
+	r := httptest.NewRequest(verb, target, bytes.NewReader(body))
 	r.Header.Set("Content-Type", "application/json")
 	r.Header.Set("Accept", "application/json")
 	r.ContentLength = int64(len(body))
 	return r, nil
 }
 
+// zapVerbs are the methods a ZAP call may name. Anything else is refused rather
+// than coerced — a caller that cannot say what it wants to do does not get a guess.
+var zapVerbs = map[string]bool{
+	http.MethodGet: true, http.MethodHead: true, http.MethodPost: true,
+	http.MethodPut: true, http.MethodPatch: true, http.MethodDelete: true,
+}
+
+// splitMethod reads a ZAP method as an HTTP request line: "<VERB> <path>", e.g.
+// "GET rag/stores" or "PATCH ai/providers/acme/openai". The path is rooted at /v1.
+//
+// The verb is CARRIED, not inferred. It used to be guessed from the method name —
+// a "get-" prefix meant GET and everything else meant POST — which worked only
+// because the route surface encoded its verb in every route name (/v1/get-store
+// vs /v1/update-store). Once the surface became RESTful that signal disappeared:
+// one path answers GET, PATCH and DELETE, and no prefix distinguishes them. A
+// heuristic there would silently send a delete as a POST.
+//
+// A method with no verb is an ERROR, not a default. Defaulting would turn a
+// caller's omission into a wrong-but-plausible request — for a surface where the
+// verb decides between reading a resource and destroying it, that is the one
+// behaviour worth refusing outright.
+func splitMethod(method string) (verb, path string, err error) {
+	m := strings.TrimSpace(method)
+	head, rest, ok := strings.Cut(m, " ")
+	if !ok {
+		return "", "", fmt.Errorf("method %q must be %q — the HTTP verb is required, not inferred", method, "<VERB> <path>")
+	}
+	verb = strings.ToUpper(strings.TrimSpace(head))
+	if !zapVerbs[verb] {
+		return "", "", fmt.Errorf("method %q names an unsupported verb %q", method, verb)
+	}
+	rest = strings.TrimSpace(rest)
+	if rest == "" {
+		return "", "", fmt.Errorf("method %q names no path", method)
+	}
+	return verb, "/v1/" + strings.TrimPrefix(rest, "/"), nil
+}
+
 // splitInput separates a JSON object into its scalar fields (rendered as query
 // strings) and at most one nested object/array field (the resource body). A
 // non-object input yields no scalars and the whole value as the nested body.

zapface/integration_test.go

@@ -26,27 +26,31 @@ func startZapApp(t *testing.T) (string, func()) {
 	t.Helper()
 	app := zip.New(zip.Config{})
 
-	// /v1 handlers (mirror ai/mount.go: app.All("/v1/*", ...)).
+	// /v1 handlers (mirror ai/mount.go: app.All("/v1/*", ...)). The surface is
+	// RESTful, so these switch on METHOD AND PATH — the same pair the dispatcher
+	// now has to carry, and the reason it can no longer infer a verb from a name.
 	app.All("/v1/*", func(c *zip.Ctx) error {
-		path := c.Path()
+		path, method := c.Path(), c.Method()
 		switch {
-		case strings.HasSuffix(path, "/get-global-providers"):
+		case method == "GET" && strings.HasSuffix(path, "/ai/providers/global"):
 			return c.JSON(200, fiber.Map{
 				"status": "ok", "msg": "",
 				"data": []fiber.Map{
 					{"owner": "admin", "name": "openai", "category": "Model", "_cookie": c.Header("Cookie")},
 				},
 			})
-		case strings.HasSuffix(path, "/get-providers"):
+		case method == "GET" && strings.HasSuffix(path, "/ai/providers"):
 			return c.JSON(200, fiber.Map{
 				"status": "ok", "msg": "",
 				"data":  []fiber.Map{{"owner": c.Query("owner"), "name": "p1"}},
 				"data2": 1,
 			})
-		case strings.HasSuffix(path, "/add-provider"):
+		case method == "POST" && strings.HasSuffix(path, "/ai/providers"):
 			body := map[string]any{}
 			_ = json.Unmarshal(c.Body(), &body)
 			return c.JSON(200, fiber.Map{"status": "ok", "msg": "added " + asString(body["name"])})
+		case method == "DELETE" && strings.Contains(path, "/ai/providers/"):
+			return c.JSON(200, fiber.Map{"status": "ok", "msg": "deleted " + path})
 		default:
 			return c.JSON(404, fiber.Map{"status": "error", "msg": "not found"})
 		}
@@ -130,10 +134,10 @@ func TestEndToEndOverWebSocket(t *testing.T) {
 		}
 	}
 
-	// 1) get-global-providers -> ok, real data, cookie replayed to /v1 handler.
-	rep := call("get-global-providers", nil, 1)
+	// 1) A global listing -> ok, real data, cookie replayed to the /v1 handler.
+	rep := call("GET ai/providers/global", nil, 1)
 	if !rep.ok || rep.status != 200 {
-		t.Fatalf("get-global-providers: ok=%v status=%d err=%s", rep.ok, rep.status, rep.errorJSON)
+		t.Fatalf("GET ai/providers/global: ok=%v status=%d err=%s", rep.ok, rep.status, rep.errorJSON)
 	}
 	var providers []map[string]any
 	if err := json.Unmarshal(superJSONUnwrap(rep.result), &providers); err != nil {
@@ -146,26 +150,43 @@ func TestEndToEndOverWebSocket(t *testing.T) {
 		t.Fatalf("cookie not replayed to /v1 handler: %v", providers[0]["_cookie"])
 	}
 
-	// 2) get-providers with a query param.
-	rep = call("get-providers", map[string]any{"owner": "admin"}, 2)
+	// 2) A collection list with a query param.
+	rep = call("GET ai/providers", map[string]any{"owner": "admin"}, 2)
 	if !rep.ok {
-		t.Fatalf("get-providers !ok: %s", rep.errorJSON)
+		t.Fatalf("GET ai/providers !ok: %s", rep.errorJSON)
 	}
 	_ = json.Unmarshal(superJSONUnwrap(rep.result), &providers)
 	if providers[0]["owner"] != "admin" {
-		t.Fatalf("get-providers owner query not forwarded: %v", providers)
+		t.Fatalf("owner query not forwarded: %v", providers)
 	}
 
-	// 3) add-provider (POST body).
-	rep = call("add-provider", map[string]any{"owner": "admin", "name": "claude"}, 3)
+	// 3) Create (POST body).
+	rep = call("POST ai/providers", map[string]any{"owner": "admin", "name": "claude"}, 3)
 	if !rep.ok {
-		t.Fatalf("add-provider !ok: %s", rep.errorJSON)
+		t.Fatalf("POST ai/providers !ok: %s", rep.errorJSON)
 	}
 
-	// 4) unknown method -> /v1 404 envelope -> !ok.
-	rep = call("get-nonexistent", nil, 4)
+	// 4) A DELETE reaches the DELETE route — the case the old prefix heuristic
+	// got wrong: it had no "get-" prefix, so it would have been sent as a POST
+	// and silently landed on the create route instead of destroying anything.
+	rep = call("DELETE ai/providers/admin/openai", nil, 4)
+	if !rep.ok {
+		t.Fatalf("DELETE ai/providers/admin/openai !ok: %s", rep.errorJSON)
+	}
+
+	// 5) unknown path -> /v1 404 envelope -> !ok.
+	rep = call("GET ai/nonexistent", nil, 5)
 	if rep.ok {
-		t.Fatalf("unknown method should not be ok")
+		t.Fatalf("unknown path should not be ok")
+	}
+
+	// 6) A method with NO verb is refused outright, not guessed at.
+	rep = call("get-providers", nil, 6)
+	if rep.ok {
+		t.Fatalf("a verbless method must be refused, not inferred")
+	}
+	if !strings.Contains(rep.errorJSON, "verb is required") {
+		t.Fatalf("refusal should name the missing verb, got %s", rep.errorJSON)
 	}
 }
 

zapface/wire_test.go

@@ -36,7 +36,7 @@ func buildClientRequestBytes(method, payload string, promiseID uint32) []byte {
 // TestParseClientRequest proves the server decodes a real client frame: outer
 // rpc envelope -> inner ZapRequest -> (method, SuperJSON payload).
 func TestParseClientRequest(t *testing.T) {
-	const method = "get-providers"
+	const method = "GET ai/providers"
 	// console sends SuperJSON.stringify(input); a plain object X -> {"json":X}.
 	input := map[string]any{"owner": "admin", "store": "default", "limit": "20"}
 	innerJSON, _ := json.Marshal(input)
@@ -176,45 +176,54 @@ func TestBuildHTTPRequest(t *testing.T) {
 		wantBodyHa string // a substring expected in the body ('' = no body)
 	}{
 		{
-			name:       "get-global-providers no args",
-			method:     "get-global-providers",
+			name:       "global listing, no args",
+			method:     "GET ai/providers/global",
 			input:      nil,
 			wantMethod: "GET",
-			wantPath:   "/v1/get-global-providers",
+			wantPath:   "/v1/ai/providers/global",
 		},
 		{
-			name:       "get-provider with id",
-			method:     "get-provider",
-			input:      map[string]any{"id": "admin/openai"},
+			name:       "member read — the id is in the PATH, not a query param",
+			method:     "GET ai/providers/admin/openai",
+			input:      nil,
 			wantMethod: "GET",
-			wantPath:   "/v1/get-provider",
-			wantQuery:  map[string]string{"id": "admin/openai"},
+			wantPath:   "/v1/ai/providers/admin/openai",
 		},
 		{
-			name:       "get-providers list query",
-			method:     "get-providers",
+			name:       "collection list keeps its filters as a query",
+			method:     "GET ai/providers",
 			input:      map[string]any{"owner": "admin", "store": "default"},
 			wantMethod: "GET",
-			wantPath:   "/v1/get-providers",
+			wantPath:   "/v1/ai/providers",
 			wantQuery:  map[string]string{"owner": "admin", "store": "default"},
 		},
 		{
-			name:       "update-provider id+resource",
-			method:     "update-provider",
-			input:      map[string]any{"id": "admin/openai", "provider": map[string]any{"owner": "admin", "name": "openai", "type": "OpenAI"}},
-			wantMethod: "POST",
-			wantPath:   "/v1/update-provider",
-			wantQuery:  map[string]string{"id": "admin/openai"},
+			// The verb travels with the call. Under the old prefix heuristic this
+			// same request would have been sent as a POST.
+			name:       "member update is a PATCH with the resource as the body",
+			method:     "PATCH ai/providers/admin/openai",
+			input:      map[string]any{"provider": map[string]any{"owner": "admin", "name": "openai", "type": "OpenAI"}},
+			wantMethod: "PATCH",
+			wantPath:   "/v1/ai/providers/admin/openai",
 			wantBodyHa: `"type":"OpenAI"`,
 		},
 		{
-			name:       "add-provider bare resource",
-			method:     "add-provider",
+			name:       "create posts the bare resource to the collection",
+			method:     "POST ai/providers",
 			input:      map[string]any{"owner": "admin", "name": "openai", "type": "OpenAI"},
 			wantMethod: "POST",
-			wantPath:   "/v1/add-provider",
+			wantPath:   "/v1/ai/providers",
 			wantBodyHa: `"name":"openai"`,
 		},
+		{
+			// A DELETE carries no body — and under the old heuristic it would have
+			// been a POST, i.e. a destroy sent as a write to the wrong route.
+			name:       "member delete is a DELETE with no body",
+			method:     "DELETE ai/providers/admin/openai",
+			input:      nil,
+			wantMethod: "DELETE",
+			wantPath:   "/v1/ai/providers/admin/openai",
+		},
 	}
 
 	for _, tc := range cases {