llmgw/internal/control/adminapi/api.go

// Package adminapi is the control plane's management surface, which `ctl`
// consumes (plan §11, phase 3).
//
// It is mounted on the admin listener rather than the client-facing one. The
// two have different audiences: one serves inference to whoever holds an API
// key, the other lets a caller cordon nodes and read the cluster's shape.
package adminapi

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"slices"
	"strings"
	"time"

	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
	"gitea.mynodes.xyz/ilker/llmgw/internal/bench"
	"gitea.mynodes.xyz/ilker/llmgw/internal/control"
	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
	"gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrics"
	"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
	"gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink"
	"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
	"gitea.mynodes.xyz/ilker/llmgw/internal/control/topology"
	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)

// API serves the management endpoints.
type API struct {
	models   Models
	capacity Capacity
	usage    Usage
	ctl      *control.Controller
}

// New builds the API over a controller.
func New(ctl *control.Controller) *API { return &API{ctl: ctl} }

// Register mounts the endpoints on a mux under /api/v1.
func (a *API) Register(mux *http.ServeMux) {
	mux.HandleFunc("GET /api/v1/summary", a.summary)
	// Adding a model is a write, and the only one this API has. It lives here
	// rather than on the client-facing listener for the same reason the rest of
	// it does: reaching this port already means holding the control plane.
	mux.HandleFunc("POST /api/v1/models", a.addModel)
	mux.HandleFunc("DELETE /api/v1/models/{name}", a.removeModel)
	mux.HandleFunc("POST /api/v1/models/{name}/scale", a.scaleModel)
	mux.HandleFunc("PUT /api/v1/models/{name}/capacity", a.putCapacity)
	mux.HandleFunc("GET /api/v1/capacity", a.listCapacity)
	mux.HandleFunc("GET /api/v1/usage", a.usageByScope)
	mux.HandleFunc("GET /api/v1/nodes", a.listNodes)
	mux.HandleFunc("GET /api/v1/fabrics", func(w http.ResponseWriter, _ *http.Request) {
		writeJSON(w, http.StatusOK, a.Fabrics())
	})
	mux.HandleFunc("GET /api/v1/nvlink", func(w http.ResponseWriter, _ *http.Request) {
		writeJSON(w, http.StatusOK, a.NVLink())
	})
	mux.HandleFunc("GET /api/v1/fabrics/{id}/topology", func(w http.ResponseWriter, r *http.Request) {
		t, ok := a.Topology(r.PathValue("id"))
		if !ok {
			http.NotFound(w, r)
			return
		}
		writeJSON(w, http.StatusOK, t)
	})
	mux.HandleFunc("GET /api/v1/models", a.listModels)
	mux.HandleFunc("GET /api/v1/replicas", a.listReplicas)
	// Listing is a read and deleting is a separate verb, so a mistyped
	// command cannot remove anything.
	mux.HandleFunc("GET /api/v1/weights/orphans", a.listOrphans)
	mux.HandleFunc("POST /api/v1/weights/collect", a.collectWeights)
	mux.HandleFunc("GET /api/v1/explain-placement", a.explainPlacement)
	mux.HandleFunc("POST /api/v1/nodes/{id}/state", a.setNodeState)
	mux.HandleFunc("DELETE /api/v1/nodes/{id}", a.removeNode)
	mux.HandleFunc("POST /api/v1/fabric-test", a.fabricTest)
	mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
	mux.HandleFunc("POST /api/v1/restart", a.restart)
}

// --- responses ---

// NodeView is a node as `ctl nodes` shows it.
type NodeView struct {
	ID       string            `json:"id"`
	Hostname string            `json:"hostname,omitempty"`
	MgmtAddr string            `json:"mgmt_addr,omitempty"`
	State    string            `json:"state"`
	Labels   map[string]string `json:"labels,omitempty"`

	GPUTotal int `json:"gpu_total"`
	GPUFree  int `json:"gpu_free"`
	// MemoryGB is the machine's RAM, 0 when the agent did not report it. It is
	// what decides whether a CPU profile can be placed here, so it belongs
	// where someone looks when a model will not schedule.
	MemoryGB int `json:"memory_gb,omitempty"`
	// Engines are the inference engines installed here, as the agent found
	// them. Absent means the agent is older than the probe and said nothing,
	// which is not the same as having none — placement treats it that way too,
	// or a control plane upgrade would empty a fleet.
	Engines []string `json:"engines,omitempty"`
	// Provision is where installing each wanted engine stands. An engine is in
	// Engines only once its installation here is ready.
	Provision []inventory.EngineInstall `json:"provision,omitempty"`
	// OS is the operating system and architecture, as the agent read them.
	OS string `json:"os,omitempty"`
	// Warnings are facts about the machine that make it slow or unusable
	// without making it unhealthy: a VM with no vector instructions runs every
	// model, at a fifth of the speed.
	Warnings []string `json:"warnings,omitempty"`

	// Models are the weights this node can see. Absent means it was not asked
	// to look, which is not the same as holding none.
	Models []string `json:"models_on_disk,omitempty"`
	// DiskFreeGB and DiskTotalGB are the filesystems holding those weights,
	// summed. Zero means the agent did not report any, which an older agent
	// and a node with no model directories both do — so it reads as unknown
	// rather than as a full disk.
	//
	// Summed because an operator asking "can this node take another model"
	// wants one number; the per-filesystem detail is in the node's own record
	// for anyone who needs it. Shared filesystems are left out of the sum: the
	// space on a filer is not this node's to spend, and adding it would let a
	// fleet of twenty nodes each believe it has the same free terabyte.
	DiskFreeGB  int      `json:"disk_free_gb,omitempty"`
	DiskTotalGB int      `json:"disk_total_gb,omitempty"`
	Classes     []string `json:"gpu_classes,omitempty"`
	Fabrics     []string `json:"fabrics,omitempty"`

	// Unhealthy lists GPUs kept out of placement and why, which is usually the
	// answer when a node looks idle but nothing will schedule onto it.
	Unhealthy []string `json:"unhealthy_gpus,omitempty"`

	LastSeenAgo  string `json:"last_seen_ago"`
	AgentVersion string `json:"agent_version,omitempty"`
	// Features are the requests the agent answers beyond the usual:
	// "fabric-test". Absent is an agent too old to say.
	Features []string `json:"features,omitempty"`
	Replicas int      `json:"replicas"`

	// Load is how busy the machine is, absent when its agent has not
	// measured it lately — which is unknown, not idle.
	Load *LoadView `json:"load,omitempty"`
	// IBPorts are the node's InfiniBand and RoCE ports at the last heartbeat.
	IBPorts []IBPortView `json:"ib_ports,omitempty"`
	// GPUs are the cards, with how they are wired: NVLink peers, the
	// adapter nearest each, and the NVLink domain each has joined.
	GPUs []GPUView `json:"gpus,omitempty"`
	// NVLinkDomain is the multi-node domain every card has joined, if any.
	NVLinkDomain string `json:"nvlink_domain,omitempty"`
	IMEX         string `json:"imex,omitempty"`
	IMEXChannels int    `json:"imex_channels,omitempty"`
}

// GPUView is one card as a node's page shows it.
type GPUView struct {
	Index       int    `json:"index"`
	Name        string `json:"name"`
	NUMA        int    `json:"numa"`
	NVLinkPeers []int  `json:"nvlink_peers,omitempty"`
	// NearestNIC is the adapter closest to the card, and how close.
	NearestNIC  string `json:"nearest_nic,omitempty"`
	NICPath     string `json:"nic_path,omitempty"`
	FabricState string `json:"fabric_state,omitempty"`
	Domain      string `json:"domain,omitempty"`
	Bandwidth   string `json:"bandwidth,omitempty"`
}

// IBPortView is one port as a page shows it.
type IBPortView struct {
	HCA       string `json:"hca"`
	Port      int    `json:"port"`
	State     string `json:"state"`
	PhysState string `json:"phys_state,omitempty"`
	RateGbps  int    `json:"rate_gbps"`
	LinkLayer string `json:"link_layer,omitempty"`
	Active    bool   `json:"active"`
	// Fabric is the declared fabric this port's adapter carries, if any.
	Fabric string `json:"fabric,omitempty"`
	// Physical is the fabric the port is physically on: its subnet manager,
	// or its RoCE network. SMError says why that is not known.
	Physical string `json:"physical,omitempty"`
	SMError  string `json:"sm_error,omitempty"`
	// Errors are the error counters that are not zero, by name.
	Errors  map[string]uint64 `json:"errors,omitempty"`
	TxBytes uint64            `json:"tx_bytes,omitempty"`
	RxBytes uint64            `json:"rx_bytes,omitempty"`
}

// LoadView is a node's CPU and memory in use, at its last heartbeat.
type LoadView struct {
	CPUPercent float64 `json:"cpu_percent"`
	CPUs       int     `json:"cpus,omitempty"`
	Load1      float64 `json:"load1"`
	Load5      float64 `json:"load5"`
	Load15     float64 `json:"load15"`
	MemUsed    int64   `json:"memory_used_bytes"`
	MemTotal   int64   `json:"memory_total_bytes"`
	// MemPercent is used over used plus available.
	MemPercent float64 `json:"memory_percent"`
	AgeSec     int64   `json:"age_seconds"`
}

// loadFresh is how old a measurement may be and still be shown as current.
const loadFresh = time.Minute

func loadView(l *inventory.Load, now time.Time) *LoadView {
	if l == nil || now.Sub(l.At) > loadFresh {
		return nil
	}
	v := &LoadView{
		CPUPercent: l.CPUPercent, CPUs: l.CPUCount,
		Load1: l.Load1, Load5: l.Load5, Load15: l.Load15,
		MemUsed: l.MemUsedBytes, MemTotal: l.MemUsedBytes + l.MemAvailBytes,
		AgeSec: int64(now.Sub(l.At).Seconds()),
	}
	if v.MemTotal > 0 {
		v.MemPercent = float64(v.MemUsed) * 100 / float64(v.MemTotal)
	}
	return v
}

// ModelView is a model as `ctl models` shows it.
type ModelView struct {
	Name string `json:"name"`
	// Description is what the model is for, when somebody said.
	Description string   `json:"description,omitempty"`
	Aliases     []string `json:"aliases,omitempty"`
	Source      string   `json:"source"`
	Min         int      `json:"min_replicas"`
	Max         int      `json:"max_replicas"`
	Ready       int      `json:"ready_replicas"`
	Total       int      `json:"total_replicas"`
	Fallbacks   []string `json:"fallbacks,omitempty"`
	Profiles    []string `json:"profiles"`
}

// ReplicaView is a replica as `ctl replicas` shows it.
type ReplicaView struct {
	ID string `json:"id"`
	// EngineKey is the key this replica's engine demands, for `ctl bench`
	// and anyone else holding the admin listener who needs to reach an engine
	// directly. The client-facing pages never render it.
	EngineKey string `json:"engine_key,omitempty"`
	// Restart is "queued" or "replacing" while an operator's restart of this
	// replica is under way.
	Restart string `json:"restart,omitempty"`
	Model   string `json:"model"`
	Profile string `json:"profile"`
	// Stale is true when the profile has been edited since this replica was
	// placed, so it is running a definition the catalog no longer holds.
	//
	// Reported rather than acted on. Moving a replica that is serving costs
	// more than the imperfection, and the operator who made the edit is the
	// one who should decide whether it is worth a restart.
	Stale    bool   `json:"stale,omitempty"`
	State    string `json:"state"`
	Pool     string `json:"pool,omitempty"`
	Fabric   string `json:"fabric,omitempty"`
	Endpoint string `json:"endpoint,omitempty"`
	AgeSec   int64  `json:"age_seconds"`
	// IdleSec is how long since a request was routed here. Negative means the
	// router has not reported any use of this replica, which is not the same
	// as idle — it is what a replica looks like before anything has asked for
	// it, and after a router restart.
	IdleSec int64       `json:"idle_seconds"`
	Err     string      `json:"error,omitempty"`
	Shards  []ShardView `json:"shards"`
}

// ShardView is one shard of a replica.
type ShardView struct {
	ID    string `json:"id"`
	Node  string `json:"node"`
	Rank  int    `json:"rank"`
	GPUs  []int  `json:"gpus"`
	State string `json:"state"`
	Err   string `json:"error,omitempty"`
	// Fetch is the download this shard is waiting on, absent when it is not
	// waiting on one. It is what tells a replica pulling a hundred gigabytes
	// apart from one that is wedged — which look identical without it.
	Fetch *FetchView `json:"fetch,omitempty"`
	// CPUCores and MemoryBytes are the engine's use of its node, from the
	// node's last heartbeat; zero when not measured.
	CPUCores    float64 `json:"cpu_cores,omitempty"`
	MemoryBytes int64   `json:"memory_bytes,omitempty"`
}

// FetchView is a download in flight.
type FetchView struct {
	Source    string `json:"source"`
	BytesDone int64  `json:"bytes_done"`
	// BytesTotal is 0 when the server did not say how large the file is.
	BytesTotal int64 `json:"bytes_total,omitempty"`
	// Percent is absent rather than zero when the total is unknown: an
	// invented percentage that stops moving is worse than a byte count that
	// keeps rising.
	Percent *int `json:"percent,omitempty"`
}

// ExplainView is why a model could not be placed.
type ExplainView struct {
	Model string `json:"model"`
	// Placed is true when the model has replicas, in which case there is
	// nothing to explain.
	Placed bool `json:"placed"`
	// Rejections is empty when placement was never attempted, which is itself
	// the answer: the model is satisfied, or the reconciler has not run yet.
	Rejections []RejectionView `json:"rejections,omitempty"`
}

// RejectionView is one rejected candidate.
type RejectionView struct {
	Profile string `json:"profile"`
	Pool    string `json:"pool,omitempty"`
	Node    string `json:"node,omitempty"`
	Reason  string `json:"reason"`
}

// --- handlers ---

func (a *API) listNodes(w http.ResponseWriter, _ *http.Request) {
	writeJSON(w, http.StatusOK, a.Nodes())
}

// Nodes builds the node view.
//
// Exported because the dashboard renders the same data server-side. Going back
// out through HTTP to fetch what is already in this process would add a way for
// the two to disagree.
func (a *API) Nodes() []NodeView {
	replicasOn := map[string]int{}
	for _, rep := range a.ctl.Reconciler().Replicas() {
		for _, n := range rep.NodeIDs() {
			replicasOn[n]++
		}
	}

	inv := a.ctl.Inventory()
	out := make([]NodeView, 0)
	for _, n := range inv.Nodes() {
		v := NodeView{
			ID: n.ID, Hostname: n.Hostname, MgmtAddr: n.MgmtAddr,
			State: n.State.String(), Labels: n.Labels,
			GPUTotal: len(n.GPUs), GPUFree: len(inv.FreeGPUs(n.ID)),
			MemoryGB: n.MemoryTotalMB / 1024,
			Fabrics:  n.FabricNames(), AgentVersion: n.AgentVersion, Features: n.Features,
			Replicas: replicasOn[n.ID],
		}
		if free, total := n.LocalDisk(); total > 0 {
			v.DiskFreeGB = int(free / (1 << 30))
			v.DiskTotalGB = int(total / (1 << 30))
		}
		for name := range n.EngineVersions {
			v.Engines = append(v.Engines, name)
		}
		slices.Sort(v.Engines)
		v.Provision = n.Provision
		if h := n.Host; h.OSID != "" {
			v.OS = strings.TrimSpace(h.OSID + " " + h.OSVersion + " " + h.Arch)
			if h.Arch == "amd64" && !h.HasCPUFlag("avx2") {
				v.Warnings = append(v.Warnings, "no AVX2: inference runs several times slower "+
					"(a VM needs CPU type \"host\")")
			}
		}
		for _, w := range n.Weights {
			v.Models = append(v.Models, w.Path)
		}
		seen := map[string]struct{}{}
		for _, g := range n.GPUs {
			if _, dup := seen[g.Class()]; !dup {
				seen[g.Class()] = struct{}{}
				v.Classes = append(v.Classes, g.Class())
			}
			if !g.Health.OK() {
				v.Unhealthy = append(v.Unhealthy,
					fmt.Sprintf("gpu %d: %s", g.Index, g.Health.Problems[0]))
			}
		}
		if !n.LastSeen.IsZero() {
			v.LastSeenAgo = time.Since(n.LastSeen).Round(time.Second).String()
		}
		v.Load = loadView(n.Load, time.Now())
		v.NVLinkDomain, v.IMEX, v.IMEXChannels = n.NVLinkDomain(), n.IMEX.State, n.IMEX.Channels
		for _, g := range n.GPUs {
			gv := GPUView{Index: g.Index, Name: g.Name, NUMA: g.NUMANode, NVLinkPeers: g.NVLinkPeers}
			for hca, rel := range g.NICPaths {
				if gv.NearestNIC == "" || discovery.PathRank(rel) < discovery.PathRank(gv.NICPath) ||
					(discovery.PathRank(rel) == discovery.PathRank(gv.NICPath) && hca < gv.NearestNIC) {
					gv.NearestNIC, gv.NICPath = hca, rel
				}
			}
			if f := g.Fabric; f != nil {
				gv.FabricState, gv.Domain, gv.Bandwidth = f.State, f.Domain(), f.Bandwidth
			}
			v.GPUs = append(v.GPUs, gv)
		}
		fabricOf := map[string]string{}
		for _, f := range n.Fabrics {
			for _, h := range f.HCAs {
				fabricOf[h] = f.Fabric
			}
		}
		for _, p := range n.IBPorts {
			pv := IBPortView{HCA: p.HCA, Port: p.Port, State: p.State, PhysState: p.PhysState,
				RateGbps: p.RateGbps, LinkLayer: p.LinkLayer, Active: p.Active(), Fabric: fabricOf[p.HCA],
				TxBytes: p.Counters["port_xmit_data"], RxBytes: p.Counters["port_rcv_data"], SMError: p.SMError}
			switch {
			case p.RoCE() && p.Network != "":
				pv.Physical = "RoCE " + p.Network
			case p.SMGUID != "":
				pv.Physical = fmt.Sprintf("SM %s (LID %d)", p.SMGUID, p.SMLID)
			}
			for name, c := range p.Counters {
				if c > 0 && name != "port_xmit_data" && name != "port_rcv_data" {
					if pv.Errors == nil {
						pv.Errors = map[string]uint64{}
					}
					pv.Errors[name] = c
				}
			}
			v.IBPorts = append(v.IBPorts, pv)
		}
		out = append(out, v)
	}
	return out
}

func (a *API) listModels(w http.ResponseWriter, _ *http.Request) {
	writeJSON(w, http.StatusOK, a.Models())
}

// Models builds the catalog view.
func (a *API) Models() []ModelView {
	ready, total := map[string]int{}, map[string]int{}
	for _, rep := range a.ctl.Reconciler().Replicas() {
		total[rep.Model]++
		if rep.State == reconciler.Ready {
			ready[rep.Model]++
		}
	}

	out := make([]ModelView, 0)
	for _, m := range a.ctl.Catalog().Models() {
		v := ModelView{
			Name: m.Name, Description: m.Description, Aliases: m.Aliases, Source: m.Source,
			Min: m.Scaling.Min, Max: m.Scaling.Max,
			Ready: ready[m.Name], Total: total[m.Name],
			Fallbacks: m.Fallbacks,
		}
		for _, p := range m.Profiles {
			v.Profiles = append(v.Profiles, p.Name)
		}
		out = append(out, v)
	}
	return out
}

func (a *API) listReplicas(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, a.Replicas(r.URL.Query().Get("model")))
}

// Replicas builds the replica view, optionally filtered to one model.
func (a *API) Replicas(model string) []ReplicaView {
	out := make([]ReplicaView, 0)
	loads := map[string]*inventory.Load{}
	for _, n := range a.ctl.Inventory().Nodes() {
		if n.Load != nil && time.Since(n.Load.At) <= loadFresh {
			loads[n.ID] = n.Load
		}
	}
	for _, rep := range a.ctl.Reconciler().Replicas() {
		if model != "" && rep.Model != model {
			continue
		}
		v := ReplicaView{
			ID: rep.ID, Model: rep.Model, Profile: rep.Profile,
			State: rep.State.String(), Pool: rep.Pool, Fabric: rep.Fabric,
			Endpoint: rep.Endpoint, Err: rep.Err,
			EngineKey: a.ctl.EngineKey(rep.ID),
			Restart:   a.ctl.Reconciler().RestartState(rep.ID),
			Stale:     a.profileChanged(rep.Model, rep.Profile, rep.ProfileFP),
			AgeSec:    int64(time.Since(rep.CreatedAt).Seconds()),
			IdleSec:   -1,
		}
		if rep.LastUsedAt.IsZero() {
			v.IdleSec = -1
		} else {
			v.IdleSec = int64(time.Since(rep.LastUsedAt).Seconds())
		}
		for _, s := range rep.Shards {
			sv := ShardView{
				ID: s.ID, Node: s.NodeID, Rank: s.Rank,
				GPUs: s.GPUIndices, State: s.State.String(), Err: s.Err,
			}
			if f := s.Fetch; f != nil {
				sv.Fetch = &FetchView{
					Source: f.Source, BytesDone: f.BytesDone, BytesTotal: f.BytesTotal,
				}
				if p, ok := f.Percent(); ok {
					sv.Fetch.Percent = &p
				}
			}
			if l := loads[s.NodeID]; l != nil {
				if use, ok := l.Shards[s.ID]; ok {
					sv.CPUCores, sv.MemoryBytes = use.CPUCores, use.RSSBytes
				}
			}
			v.Shards = append(v.Shards, sv)
		}
		out = append(out, v)
	}
	return out
}

// Models is the service that owns catalog changes, when there is one.
//
// Optional because a control plane with no store cannot persist a model, and
// pretending otherwise would accept a model that vanishes on restart.
type Models interface {
	Add(ctx context.Context, m catalog.Model, by string) error
	Remove(ctx context.Context, name string) error
	Scale(ctx context.Context, name string, s catalog.Scaling, by string) (catalog.Model, error)
}

// Usage reports what each group of the tenancy tree spent. The store
// implements it.
type Usage interface {
	UsageByScope(ctx context.Context, scope store.Scope, since time.Time) ([]store.ScopeUsage, error)
}

// SetUsage supplies the usage reporting source.
func (a *API) SetUsage(u Usage) { a.usage = u }

// ScopeUsageView is one group's spend, with its quota when it has one.
type ScopeUsageView struct {
	Scope string `json:"scope"`
	ID    string `json:"id"`
	Name  string `json:"name"`

	Keys       int64 `json:"keys"`
	Requests   int64 `json:"requests"`
	Prompt     int64 `json:"prompt_tokens"`
	Completion int64 `json:"completion_tokens"`
	CostMicros int64 `json:"cost_micros"`
	Errors     int64 `json:"errors"`

	LastAt time.Time `json:"last_at,omitzero"`
	// Budget is what this level sets, absent when it sets none. A group bounded
	// only by its parent has no budget of its own, and reporting the parent's
	// against each child would read as though each had that much.
	Budget       *int64 `json:"budget_micros,omitempty"`
	BudgetPeriod string `json:"budget_period,omitempty"`
	// Fraction is spend over budget, absent for the same reason.
	Fraction *float64 `json:"fraction,omitempty"`
}

func (a *API) usageByScope(w http.ResponseWriter, r *http.Request) {
	if a.usage == nil {
		writeError(w, http.StatusNotImplemented, "this control plane has no store, so nothing is recorded")
		return
	}
	scope := store.Scope(r.URL.Query().Get("by"))
	if scope == "" {
		scope = store.ScopeTeam
	}
	since := 24 * time.Hour
	if v := r.URL.Query().Get("since"); v != "" {
		d, err := time.ParseDuration(v)
		if err != nil || d <= 0 {
			writeError(w, http.StatusBadRequest, "since must be a positive duration, like 24h")
			return
		}
		since = d
	}

	rows, err := a.usage.UsageByScope(r.Context(), scope, time.Now().Add(-since))
	if err != nil {
		// The only failure a caller can act on is naming a level that is not
		// one, which is what this mostly is.
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	out := make([]ScopeUsageView, 0, len(rows))
	for _, u := range rows {
		v := ScopeUsageView{
			Scope: string(u.Scope), ID: u.ID, Name: u.Name,
			Keys: u.Keys, Requests: u.Requests, Prompt: u.Prompt,
			Completion: u.Completion, CostMicros: u.CostMicros, Errors: u.Errors,
			LastAt: u.LastAt, Budget: u.Budget, BudgetPeriod: string(u.BudgetPeriod),
		}
		if f, ok := u.Spent(); ok {
			v.Fraction = &f
		}
		out = append(out, v)
	}
	writeJSON(w, http.StatusOK, out)
}

// Capacity is where measured capacity is kept. The store implements it.
type Capacity interface {
	PutCapacity(ctx context.Context, c store.StoredCapacity) error
	Capacities(ctx context.Context) ([]store.StoredCapacity, error)
}

// SetCapacity supplies somewhere to keep benchmark results.
func (a *API) SetCapacity(c Capacity) { a.capacity = c }

// putCapacity records what a replica of a model was measured to serve.
//
// A PUT rather than a POST: re-running a benchmark replaces what was known
// about that model on that profile. It is a better answer to the same question,
// not another data point.
func (a *API) putCapacity(w http.ResponseWriter, r *http.Request) {
	if a.capacity == nil {
		writeError(w, http.StatusNotImplemented,
			"this control plane has no store, so a measurement recorded here would not survive a restart")
		return
	}
	var c bench.Capacity
	if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&c); err != nil {
		writeError(w, http.StatusBadRequest, "the measurement is not valid JSON: "+err.Error())
		return
	}
	c.Model = r.PathValue("name")
	if c.Profile == "" {
		writeError(w, http.StatusBadRequest,
			"a measurement needs the profile it was taken on: the same model on CPU and on a GPU are different numbers")
		return
	}

	// The model has to exist now, which catches a typo in the name — a
	// measurement filed under a misspelling is worse than none, because it
	// looks like the model was never benchmarked. It is not deleted when the
	// model is, though: a benchmark costs real GPU time and outliving the
	// catalog entry is the point.
	m, ok := a.ctl.Catalog().Resolve(c.Model)
	if !ok {
		writeError(w, http.StatusNotFound, "no model named "+c.Model+" is in the catalog")
		return
	}
	if !slices.ContainsFunc(m.Profiles, func(p catalog.Profile) bool { return p.Name == c.Profile }) {
		writeError(w, http.StatusBadRequest,
			"model "+c.Model+" has no profile named "+c.Profile)
		return
	}

	body, err := json.Marshal(c)
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	if err := a.capacity.PutCapacity(r.Context(), store.StoredCapacity{
		Model: c.Model, Profile: c.Profile, Measurement: body,
		MeasuredAt: c.MeasuredAt, MeasuredBy: r.Header.Get("X-Llmgw-Actor"),
	}); err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, c)
}

// CapacityView is one stored measurement, with who recorded it.
type CapacityView struct {
	bench.Capacity
	MeasuredBy string `json:"measured_by,omitempty"`
}

func (a *API) listCapacity(w http.ResponseWriter, r *http.Request) {
	if a.capacity == nil {
		writeJSON(w, http.StatusOK, []CapacityView{})
		return
	}
	rows, err := a.capacity.Capacities(r.Context())
	if err != nil {
		writeError(w, http.StatusInternalServerError, err.Error())
		return
	}
	out := make([]CapacityView, 0, len(rows))
	for _, row := range rows {
		var c bench.Capacity
		if err := json.Unmarshal(row.Measurement, &c); err != nil {
			// Skipped rather than failing the listing: one row written by an
			// older shape must not hide every other measurement.
			continue
		}
		out = append(out, CapacityView{Capacity: c, MeasuredBy: row.MeasuredBy})
	}
	writeJSON(w, http.StatusOK, out)
}

// SetModels supplies the catalog-changing service.
func (a *API) SetModels(m Models) { a.models = m }

func (a *API) addModel(w http.ResponseWriter, r *http.Request) {
	if a.models == nil {
		writeError(w, http.StatusNotImplemented,
			"this control plane has no store, so a model added here would not survive a restart")
		return
	}
	var m catalog.Model
	if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&m); err != nil {
		writeError(w, http.StatusBadRequest, "the model definition is not valid JSON: "+err.Error())
		return
	}
	if err := a.models.Add(r.Context(), m, r.Header.Get("X-Llmgw-Actor")); err != nil {
		// Every failure here is the definition being wrong, which is the
		// caller's to fix — a collision, a missing fallback, a profile that
		// cannot be placed. None of them are server faults.
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, map[string]string{"added": m.Name})
}

// ScaleRequest is how many replicas a model should run.
type ScaleRequest struct {
	Min int `json:"min"`
	// Max omitted follows Min, which is what the catalog does for a model that
	// declares only a minimum.
	Max         int           `json:"max,omitempty"`
	IdleTimeout time.Duration `json:"idle_timeout,omitempty"`
}

// ScaleResponse reports the scaling that is now in force.
//
// What is in force rather than what was asked for: the catalog fills in a max
// that was left out, and a caller that echoed its own request would print a
// number the cluster is not using.
type ScaleResponse struct {
	Model       string        `json:"model"`
	Min         int           `json:"min"`
	Max         int           `json:"max"`
	IdleTimeout time.Duration `json:"idle_timeout,omitempty"`
	OnDemand    bool          `json:"on_demand"`
}

func (a *API) scaleModel(w http.ResponseWriter, r *http.Request) {
	if a.models == nil {
		writeError(w, http.StatusNotImplemented, "this control plane has no store")
		return
	}
	var req ScaleRequest
	if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "the scaling request is not valid JSON: "+err.Error())
		return
	}
	if req.Min < 0 || req.Max < 0 {
		writeError(w, http.StatusBadRequest, "a replica count cannot be negative")
		return
	}
	name := r.PathValue("name")
	m, err := a.models.Scale(r.Context(), name,
		catalog.Scaling{Min: req.Min, Max: req.Max, IdleTimeout: req.IdleTimeout},
		r.Header.Get("X-Llmgw-Actor"))
	if err != nil {
		// Like addModel: every refusal here is the request being wrong — a
		// model declared in configuration, one that was never added, a max
		// below a min. None are server faults.
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, ScaleResponse{
		Model: m.Name, Min: m.Scaling.Min, Max: m.Scaling.Max,
		IdleTimeout: m.Scaling.IdleTimeout, OnDemand: m.Scaling.OnDemand(),
	})
}

func (a *API) removeModel(w http.ResponseWriter, r *http.Request) {
	if a.models == nil {
		writeError(w, http.StatusNotImplemented, "this control plane has no store")
		return
	}
	name := r.PathValue("name")
	if err := a.models.Remove(r.Context(), name); err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}
	writeJSON(w, http.StatusOK, map[string]string{"removed": name})
}

func (a *API) summary(w http.ResponseWriter, _ *http.Request) {
	writeJSON(w, http.StatusOK, a.Summary())
}

func (a *API) explainPlacement(w http.ResponseWriter, r *http.Request) {
	model := r.URL.Query().Get("model")
	if model == "" {
		writeError(w, http.StatusBadRequest, "model is required")
		return
	}
	if _, ok := a.ctl.Catalog().Resolve(model); !ok {
		writeError(w, http.StatusNotFound, fmt.Sprintf("model %q is not in the catalog", model))
		return
	}

	writeJSON(w, http.StatusOK, a.Explain(model))
}

// Explain builds the placement explanation for one model.
//
// It does not check that the model exists; the handler above does that so it
// can answer 404, while the dashboard only ever asks about models it just read
// out of the catalog.
func (a *API) Explain(model string) ExplainView {
	v := ExplainView{Model: model}
	for _, rep := range a.ctl.Reconciler().Replicas() {
		if rep.Model == model && rep.State.Counts() {
			v.Placed = true
			break
		}
	}
	if res, ok := a.ctl.Reconciler().Explain(model); ok {
		for _, rej := range res.Rejections {
			v.Rejections = append(v.Rejections, RejectionView{
				Profile: rej.Profile, Pool: rej.Pool, Node: rej.Node, Reason: rej.Reason,
			})
		}
	}
	return v
}

// setNodeState implements cordon, uncordon and drain.
//
// One endpoint rather than three verbs, because they are the same operation
// with different targets and the difference matters: cordon stops new
// placement, drain also moves existing work off.
// removeNode forgets a node that has gone for good: a machine retired or
// rebuilt under another name, which otherwise stays listed as lost until the
// control plane restarts.
//
// Only one that is not there: lost, or unreachable with nothing placed on
// it. A connected node removed here would register again on its next
// heartbeat, and one with replicas would take them out of the reconciler's
// sight while they may still be running.
func (a *API) removeNode(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	inv := a.ctl.Inventory()
	n, ok := inv.Get(id)
	if !ok {
		writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", id))
		return
	}
	switch {
	case n.State == inventory.Lost:
	case n.State == inventory.Unreachable && len(inv.Allocations(id)) == 0:
	default:
		writeError(w, http.StatusConflict, fmt.Sprintf(
			"%s is %s: stop its agent (systemctl stop llmgw-agent) and remove it once it is lost, a minute later", id, n.State))
		return
	}
	inv.Remove(id)
	a.ctl.Reconciler().Notify()
	writeJSON(w, http.StatusOK, map[string]string{"id": id, "removed": "true"})
}

func (a *API) setNodeState(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")

	var body struct {
		State string `json:"state"`
	}
	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
		writeError(w, http.StatusBadRequest, err.Error())
		return
	}

	var st inventory.NodeState
	switch body.State {
	case "ready", "uncordon":
		st = inventory.Ready
	case "cordoned", "cordon":
		st = inventory.Cordoned
	case "draining", "drain":
		st = inventory.Draining
	default:
		writeError(w, http.StatusBadRequest,
			fmt.Sprintf("unknown state %q (want ready, cordoned or draining)", body.State))
		return
	}

	if !a.ctl.Inventory().SetState(id, st) {
		writeError(w, http.StatusNotFound, fmt.Sprintf("no node %q", id))
		return
	}
	// Act on it now rather than at the next tick: an operator draining a node
	// is usually about to do something to it.
	a.ctl.Reconciler().Notify()

	writeJSON(w, http.StatusOK, map[string]string{"id": id, "state": st.String()})
}

func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	_ = json.NewEncoder(w).Encode(v)
}

func writeError(w http.ResponseWriter, code int, msg string) {
	writeJSON(w, code, map[string]string{"error": msg})
}

// profileChanged reports whether a replica is running a profile the catalog no
// longer holds.
//
// A replica with no recorded fingerprint is not stale: that is every replica
// adopted from an agent older than the field, and calling those stale would
// mark a whole fleet as wrong on the first upgrade. Nor is one whose profile
// has been deleted outright — the model or the profile going away is a
// different thing, and the replica's own state already says it.
func (a *API) profileChanged(model, profile, was string) bool {
	if was == "" {
		return false
	}
	m, ok := a.ctl.Catalog().Resolve(model)
	if !ok {
		return false
	}
	for _, p := range m.Profiles {
		if p.Name == profile {
			return p.Fingerprint() != was
		}
	}
	return false
}

// Fabrics is the physical fabrics the adapters are on, against the names
// the nodes gave them.
func (a *API) Fabrics() fabrics.Report {
	inv := a.ctl.Inventory()
	return fabrics.Analyze(inv.Nodes(), inv.SMChanges(), time.Now())
}

// TopologyView is one fabric's map, with the cluster's own adapters on it
// live: their state, rate and traffic from the last heartbeat.
type TopologyView struct {
	ID        string          `json:"id"`
	Key       string          `json:"key"`
	Label     string          `json:"label"`
	Declared  []string        `json:"declared"`
	ScannedAt time.Time       `json:"scanned_at"`
	ScannedBy string          `json:"scanned_by,omitempty"`
	TookMs    int64           `json:"took_ms"`
	Err       string          `json:"error,omitempty"`
	Map       *topology.View  `json:"map,omitempty"`
	Live      map[string]Live `json:"live"`
	// Problems are the fabric's own, from the analysis of its identity and
	// subnet managers.
	Problems []fabrics.Problem `json:"problems,omitempty"`
	SM       *fabrics.SM       `json:"sm,omitempty"`
	SMs      []fabrics.SM      `json:"sms,omitempty"`
}

// Live is one of the cluster's adapters as its node last reported it.
type Live struct {
	Node   string            `json:"node"`
	HCA    string            `json:"hca"`
	State  string            `json:"state"`
	Phys   string            `json:"phys_state"`
	Gbps   int               `json:"gbps"`
	TxBps  float64           `json:"tx_bps"`
	RxBps  float64           `json:"rx_bps"`
	Errors map[string]uint64 `json:"errors,omitempty"`
}

// FabricID is a physical fabric's key as an address can carry it.
func FabricID(key string) string {
	return strings.NewReplacer(":", "-", "/", "_").Replace(key)
}

// Topologies lists every physical fabric with what is known of its map.
func (a *API) Topologies() []TopologyView {
	rep := a.Fabrics()
	scans := a.ctl.Inventory().Scans()
	out := make([]TopologyView, 0, len(rep.Physical))
	for _, ph := range rep.Physical {
		t := TopologyView{ID: FabricID(ph.Key), Key: ph.Key, Label: ph.Label, Declared: ph.Declared}
		if sc, ok := scans[ph.Key]; ok {
			t.ScannedAt, t.ScannedBy, t.TookMs, t.Err = sc.At, sc.Node+" "+sc.Port, sc.Took.Milliseconds(), sc.Err
		}
		out = append(out, t)
	}
	return out
}

// Topology is one fabric's map, by the id Topologies gives it.
func (a *API) Topology(id string) (TopologyView, bool) {
	rep := a.Fabrics()
	var ph *fabrics.Physical
	for i := range rep.Physical {
		if FabricID(rep.Physical[i].Key) == id {
			ph = &rep.Physical[i]
		}
	}
	if ph == nil {
		return TopologyView{}, false
	}
	t := TopologyView{ID: id, Key: ph.Key, Label: ph.Label, Declared: ph.Declared, Live: map[string]Live{},
		SM: ph.SM, SMs: ph.SMs}
	for _, p := range rep.Problems {
		for _, d := range ph.Declared {
			if p.Fabric == d {
				t.Problems = append(t.Problems, p)
				break
			}
		}
	}
	hosts := map[string]topology.Host{}
	for _, n := range a.ctl.Inventory().Nodes() {
		for _, p := range n.IBPorts {
			if p.NodeGUID == "" {
				continue
			}
			hosts[p.NodeGUID] = topology.Host{Node: n.ID, HCA: p.HCA}
			l := Live{Node: n.ID, HCA: p.HCA, State: p.State, Phys: p.PhysState, Gbps: p.RateGbps,
				TxBps: p.TxBps, RxBps: p.RxBps}
			for name, c := range p.Counters {
				if c > 0 && name != "port_xmit_data" && name != "port_rcv_data" {
					if l.Errors == nil {
						l.Errors = map[string]uint64{}
					}
					l.Errors[name] = c
				}
			}
			t.Live[p.NodeGUID] = l
		}
	}
	sc, ok := a.ctl.Inventory().Scans()[ph.Key]
	if !ok {
		return t, true
	}
	t.ScannedAt, t.ScannedBy, t.TookMs, t.Err = sc.At, sc.Node+" "+sc.Port, sc.Took.Milliseconds(), sc.Err
	if sc.Output != "" {
		f, err := topology.Parse(sc.Output)
		if err != nil {
			t.Err = "the map could not be read: " + err.Error()
			return t, true
		}
		v := topology.Build(f, hosts)
		t.Map = &v
	}
	return t, true
}

// NVLink is the cluster's multi-node NVLink domains, and what keeps any of
// their nodes out.
func (a *API) NVLink() nvlink.Report {
	return nvlink.Analyze(a.ctl.Inventory().Nodes())
}

// RestartRequest names what to restart: replicas by id, every replica of some
// models, or — All — every ready replica in the cluster.
type RestartRequest struct {
	Replicas []string `json:"replicas,omitempty"`
	Models   []string `json:"models,omitempty"`
	All      bool     `json:"all,omitempty"`
}

// RestartResult lists the replicas queued.
type RestartResult struct {
	Queued []string `json:"queued"`
}

func (a *API) restart(w http.ResponseWriter, r *http.Request) {
	var req RestartRequest
	if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "the restart request is not valid JSON: "+err.Error())
		return
	}
	ids := slices.Clone(req.Replicas)
	for _, rep := range a.ctl.Reconciler().ReadyReplicas() {
		if req.All || slices.Contains(req.Models, rep.Model) {
			ids = append(ids, rep.ID)
		}
	}
	if len(ids) == 0 {
		writeError(w, http.StatusBadRequest, "nothing ready matches; name replicas, models, or all")
		return
	}
	if err := a.ctl.Reconciler().Restart(ids...); err != nil {
		writeError(w, http.StatusNotFound, err.Error())
		return
	}
	writeJSON(w, http.StatusAccepted, RestartResult{Queued: ids})
}