llmgw/internal/control/modelsvc/modelsvc.go

// Package modelsvc owns the catalog that is in force: the models declared in
// configuration, plus the ones added while the cluster is running.
//
// Two sources rather than one, deliberately. Configuration stays declarative —
// a model in the YAML is in the catalog on every start, whatever the database
// says — because that is how a team keeps its production set under review and
// in version control. The database holds what someone added afterwards, which
// is the case this exists for: trying a model should not mean a config change,
// a restart and an outage.
//
// Configuration wins every conflict, and the surprising direction is the
// dangerous one: a request for a reviewed production model must never be
// answered by something added from a laptop. A stored model that collides with
// a declared name is refused when it is added and skipped if it somehow
// survives into the database.
package modelsvc

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)

// Store is the persistence this needs.
type Store interface {
	PutModel(ctx context.Context, m store.StoredModel) error
	DeleteModel(ctx context.Context, name string) error
	Models(ctx context.Context) ([]store.StoredModel, error)
}

// Applier receives the rebuilt catalog. The controller implements it.
type Applier interface {
	ReplaceCatalog(ctx context.Context, c *catalog.Catalog)
}

// Service adds and removes models without a restart.
type Service struct {
	declared []catalog.Model
	store    Store
	apply    Applier
	log      *slog.Logger

	// check refuses a model that cannot run before it is added; see
	// SetCheck.
	check func(context.Context, catalog.Model) error

	// federated are the models federated clusters offer; see SetFederated.
	federated func() []catalog.Model

	// mu serialises rebuilds. They are rare — a person adding a model — and two
	// at once could apply catalogs in either order, leaving the one in force
	// disagreeing with the database that outlives it.
	mu sync.Mutex
}

// New builds the service around the models configuration declared.
func New(declared []catalog.Model, st Store, apply Applier, log *slog.Logger) *Service {
	if log == nil {
		log = slog.Default()
	}
	return &Service{declared: declared, store: st, apply: apply, log: log}
}

// Declared reports whether a name comes from configuration.
func (s *Service) Declared(name string) bool {
	for _, d := range s.declared {
		if d.Name == name {
			return true
		}
	}
	return false
}

// Sources maps each weights source in the catalog to the models that name it,
// so a page listing files can say which of them are in use and how they run.
func (s *Service) Sources(ctx context.Context) (map[string][]catalog.Model, error) {
	c, err := s.Load(ctx)
	if err != nil {
		return nil, err
	}
	out := map[string][]catalog.Model{}
	for _, m := range c.Models() {
		if m.Source != "" {
			out[m.Source] = append(out[m.Source], m)
		}
		for _, ad := range m.Adapters {
			out[ad.Source] = append(out[ad.Source], m)
		}
	}
	return out, nil
}

// Add puts a model in the catalog and keeps it across restarts.
//
// Validated by building the whole catalog before anything is written. A model
// that collides with an existing name, names a fallback that does not exist, or
// is malformed in any of the ways catalog.New refuses is rejected here rather
// than persisted and then skipped on every future start — which would be a
// model that looks added and never appears.
func (s *Service) Add(ctx context.Context, m catalog.Model, by string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.Declared(m.Name) {
		return fmt.Errorf("model %q is declared in the cluster's settings; edit it there (llmgw ctl config edit) rather than adding a second definition", m.Name)
	}
	// A model with no GPU is all host memory, and one that declares none
	// fits everywhere as far as placement can tell — however many of it
	// there already are. Required of what is added now; models stored before
	// this still load, and say what they are missing.
	for _, p := range m.Profiles {
		if p.CPUOnly() && p.MinMemoryGB <= 0 {
			return fmt.Errorf("model %q profile %q runs without a GPU and declares no memory (min_memory_gb): "+
				"placement could not tell when a node is full", m.Name, p.Name)
		}
	}
	if s.check != nil {
		if err := s.check(ctx, m); err != nil {
			return err
		}
	}

	stored, err := s.storedModels(ctx)
	if err != nil {
		return err
	}
	next := s.base(stored)
	for _, e := range stored {
		if e.Name == m.Name {
			continue // this call replaces it
		}
		next = append(next, e)
	}
	next = append(next, m)

	built, err := catalog.New(next)
	if err != nil {
		return fmt.Errorf("model %q would make the catalog invalid: %w", m.Name, err)
	}

	def, err := json.Marshal(m)
	if err != nil {
		return err
	}
	if err := s.store.PutModel(ctx, store.StoredModel{
		Name: m.Name, Definition: def, CreatedAt: time.Now(), CreatedBy: by,
	}); err != nil {
		return err
	}

	s.apply.ReplaceCatalog(ctx, built)
	s.log.Info("model added", "model", m.Name, "by", by)
	return nil
}

// Remove takes a model out of the catalog.
//
// A model declared in configuration cannot be removed this way: it would come
// back on the next restart, so the cluster would disagree with its own config
// file until then. Refusing and saying where to edit is the smaller surprise.
func (s *Service) Remove(ctx context.Context, name string) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.Declared(name) {
		return fmt.Errorf("model %q is declared in the cluster's settings, under models; remove it there: llmgw ctl config edit", name)
	}

	stored, err := s.storedModels(ctx)
	if err != nil {
		return err
	}
	next := s.base(stored)
	found := false
	for _, e := range stored {
		if e.Name == name {
			found = true
			continue
		}
		next = append(next, e)
	}
	if !found {
		return fmt.Errorf("model %q was not added at runtime", name)
	}

	built, err := catalog.New(next)
	if err != nil {
		// Removing a model can invalidate a catalog: another may name it as a
		// fallback. Refusing beats leaving a fallback pointing at nothing,
		// which is discovered during the outage it was meant to cover.
		return fmt.Errorf("removing %q would make the catalog invalid: %w", name, err)
	}
	if err := s.store.DeleteModel(ctx, name); err != nil {
		return err
	}

	s.apply.ReplaceCatalog(ctx, built)
	s.log.Info("model removed", "model", name)
	return nil
}

// Scale changes how many replicas a model added at runtime runs.
func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "change its scaling", func(m *catalog.Model) error {
		m.Scaling = sc
		return nil
	})
	if err == nil {
		s.log.Info("model scaled", "model", name,
			"min", final.Scaling.Min, "max", final.Scaling.Max, "by", by)
	}
	return final, err
}

// Resources sets what a model's profile reserves of a node: host memory, and
// the CPU cores a replica without a GPU computes on. An empty profile name
// sets every profile.
func (s *Service) Resources(ctx context.Context, name, profile string, memoryGB, cpuCores int, by string) (catalog.Model, error) {
	if memoryGB < 0 || cpuCores < 0 {
		return catalog.Model{}, errors.New("memory and cores cannot be negative")
	}
	final, err := s.update(ctx, name, by, "change its resources", func(m *catalog.Model) error {
		found := false
		for i := range m.Profiles {
			if profile != "" && m.Profiles[i].Name != profile {
				continue
			}
			found = true
			m.Profiles[i].MinMemoryGB = memoryGB
			m.Profiles[i].CPUCores = cpuCores
		}
		if !found {
			return fmt.Errorf("model %q has no profile %q", name, profile)
		}
		return nil
	})
	if err == nil {
		s.log.Info("model resources set", "model", name, "profile", profile,
			"memory_gb", memoryGB, "cpu_cores", cpuCores, "by", by)
	}
	return final, err
}

// SetClass sets a model's capacity class: critical, standard or
// opportunistic, or empty to go back to what its scaling implies.
func (s *Service) SetClass(ctx context.Context, name, class, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "change its class", func(m *catalog.Model) error {
		m.Class = class
		return nil
	})
	if err == nil {
		s.log.Info("model class set", "model", name, "class", final.EffectiveClass(), "by", by)
	}
	return final, err
}

// SetOwner sets the part of the tenancy tree a model's capacity counts
// against, org:ID, team:ID or project:ID; empty makes it shared.
func (s *Service) SetOwner(ctx context.Context, name, owner, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "change its owner", func(m *catalog.Model) error {
		m.Owner = owner
		return nil
	})
	if err == nil {
		s.log.Info("model owner set", "model", name, "owner", owner, "by", by)
	}
	return final, err
}

// SetAdapters replaces the LoRA adapters a model serves. Replicas load the
// new set when they next start.
func (s *Service) SetAdapters(ctx context.Context, name string, adapters []catalog.Adapter, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "change its adapters", func(m *catalog.Model) error {
		m.Adapters = adapters
		return nil
	})
	if err == nil {
		names := make([]string, 0, len(adapters))
		for _, a := range adapters {
			names = append(names, a.Name)
		}
		s.log.Info("model adapters set", "model", name, "adapters", names, "by", by)
	}
	return final, err
}

// SetSource moves a model added at runtime to other weights, recording where
// they came from. Running replicas keep what they loaded until they are
// replaced; the caller restarts them.
func (s *Service) SetSource(ctx context.Context, name, source string, origin *catalog.Origin, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "change its source", func(m *catalog.Model) error {
		m.Source, m.Origin = source, origin
		return nil
	})
	if err == nil {
		rev := ""
		if origin != nil {
			rev = origin.Revision
		}
		s.log.Info("model source set", "model", name, "source", source, "revision", rev, "by", by)
	}
	return final, err
}

// SetCanary gives a model added at runtime a canary, changes its share, or
// with nil takes it away. The candidate must already be in the catalog.
func (s *Service) SetCanary(ctx context.Context, name string, c *catalog.Canary, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "set its canary", func(m *catalog.Model) error {
		m.Canary = c
		return nil
	})
	if err == nil {
		if c == nil {
			s.log.Info("model canary removed", "model", name, "by", by)
		} else {
			s.log.Info("model canary set", "model", name, "canary", c.Model, "percent", c.Percent, "by", by)
		}
	}
	return final, err
}

// SetEval sets the quality suite a model's new weights must pass before an
// update or a promotion moves it to them; nil takes it away.
func (s *Service) SetEval(ctx context.Context, name string, g *catalog.EvalGate, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "set its eval gate", func(m *catalog.Model) error {
		m.Eval = g
		return nil
	})
	if err == nil {
		if g == nil {
			s.log.Info("model eval gate removed", "model", name, "by", by)
		} else {
			s.log.Info("model eval gate set", "model", name, "suite", g.Suite, "min_score", g.MinScore, "by", by)
		}
	}
	return final, err
}

// SetOrigin records where a model's weights came from without changing them:
// for a model added before origins were kept. A model from configuration is
// left alone, its origin worked out again when it is needed.
func (s *Service) SetOrigin(ctx context.Context, name string, origin *catalog.Origin) error {
	if s.Declared(name) {
		return nil
	}
	_, err := s.update(ctx, name, "llmgw", "record its origin", func(m *catalog.Model) error {
		m.Origin = origin
		return nil
	})
	return err
}

// update changes one model added at runtime and keeps the change.
//
// Built before anything is written, like Add: catalog.New is what applies the
// defaults and refuses what does not validate, so the model that comes back
// is the one that will run rather than the one that was asked for.
func (s *Service) update(ctx context.Context, name, by, what string, change func(*catalog.Model) error) (catalog.Model, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if s.Declared(name) {
		return catalog.Model{}, fmt.Errorf(
			"model %q is declared in the cluster's settings, under models; %s there: llmgw ctl config edit", name, what)
	}
	stored, err := s.storedModels(ctx)
	if err != nil {
		return catalog.Model{}, err
	}
	next := s.base(stored)
	found := false
	for _, e := range stored {
		if e.Name == name {
			if err := change(&e); err != nil {
				return catalog.Model{}, err
			}
			found = true
		}
		next = append(next, e)
	}
	if !found {
		return catalog.Model{}, fmt.Errorf("model %q was not added at runtime", name)
	}
	built, err := catalog.New(next)
	if err != nil {
		return catalog.Model{}, fmt.Errorf("that would make the catalog invalid: %w", err)
	}
	final, ok := built.Resolve(name)
	if !ok {
		return catalog.Model{}, fmt.Errorf("model %q vanished from the catalog it was just built into", name)
	}
	def, err := json.Marshal(final)
	if err != nil {
		return catalog.Model{}, err
	}
	if err := s.store.PutModel(ctx, store.StoredModel{
		Name: name, Definition: def, CreatedAt: time.Now(), CreatedBy: by,
	}); err != nil {
		return catalog.Model{}, err
	}
	s.apply.ReplaceCatalog(ctx, built)
	return final, nil
}

// Load builds the catalog at startup: configuration plus what was stored.
//
// A stored model that no longer validates — its engine left the build, a
// field changed meaning — is reported and skipped. Refusing to start over one
// bad row would let a model somebody added from a laptop take down the cluster,
// and the models in configuration are precisely the ones that must survive it.
func (s *Service) Load(ctx context.Context) (*catalog.Catalog, error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	return s.load(ctx)
}

// load is Load, with the lock held.
func (s *Service) load(ctx context.Context) (*catalog.Catalog, error) {
	stored, err := s.storedModels(ctx)
	if err != nil {
		return nil, err
	}

	// The federated models before the stored ones, so a stored model that
	// falls back to a peer's is checked against a catalog that has it.
	next := s.base(stored)
	for _, m := range stored {
		// Checked one at a time and against what is already accepted, so one
		// bad model cannot take a good one with it.
		candidate := append(append([]catalog.Model{}, next...), m)
		if _, err := catalog.New(candidate); err != nil {
			s.log.Warn("a stored model was skipped; it does not make a valid catalog",
				"model", m.Name, "err", err)
			continue
		}
		next = append(next, m)
	}
	return catalog.New(next)
}

// storedModels reads and decodes the runtime models, skipping any that no
// longer parse rather than failing every operation around them.
func (s *Service) storedModels(ctx context.Context) ([]catalog.Model, error) {
	if s.store == nil {
		return nil, nil
	}
	rows, err := s.store.Models(ctx)
	if err != nil {
		return nil, err
	}
	out := make([]catalog.Model, 0, len(rows))
	for _, r := range rows {
		var m catalog.Model
		if err := json.Unmarshal(r.Definition, &m); err != nil {
			s.log.Warn("a stored model could not be decoded and was skipped",
				"model", r.Name, "err", err)
			continue
		}
		out = append(out, m)
	}
	return out, nil
}

// SetFederated supplies the models federated clusters offer, which join the
// catalog after the declared ones and before those added at runtime.
//
// A function rather than a list, read at every rebuild, because the list
// changes whenever a peer's does; Refresh is how federation says it has.
func (s *Service) SetFederated(f func() []catalog.Model) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.federated = f
}

// Refresh rebuilds the catalog and puts it in force: after a peer's models
// changed. Built as at startup, so a stored model that no longer makes a
// valid catalog is skipped and said rather than taking the rest with it.
func (s *Service) Refresh(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	built, err := s.load(ctx)
	if err != nil {
		return err
	}
	s.apply.ReplaceCatalog(ctx, built)
	return nil
}

// base is the declared models, then the federated ones whose names nothing
// declared or added at runtime already uses.
//
// A peer's model is named PEER/MODEL, which a local model could also be
// called — Hugging Face ids look just like that. The local one wins, for the
// reason configuration wins over the database: a request for a model this
// cluster runs must not leave it because a peer started offering a name.
func (s *Service) base(stored []catalog.Model) []catalog.Model {
	next := append([]catalog.Model{}, s.declared...)
	if s.federated == nil {
		return next
	}
	taken := map[string]bool{}
	for _, list := range [][]catalog.Model{s.declared, stored} {
		for _, m := range list {
			taken[m.Name] = true
			for _, a := range m.Aliases {
				taken[a] = true
			}
		}
	}
	for _, f := range s.federated() {
		if taken[f.Name] {
			s.log.Warn("a federated model has the name of a model of this cluster's; not offered",
				"model", f.Name)
			continue
		}
		next = append(next, f)
	}
	return next
}

// SetFallbacks replaces the models a model added at runtime falls back to
// when it has no capacity: other models of the catalog, a federated peer's
// among them.
func (s *Service) SetFallbacks(ctx context.Context, name string, fallbacks []string, by string) (catalog.Model, error) {
	final, err := s.update(ctx, name, by, "change its fallbacks", func(m *catalog.Model) error {
		for _, f := range fallbacks {
			if f == m.Name {
				return fmt.Errorf("model %q cannot fall back to itself", name)
			}
		}
		m.Fallbacks = fallbacks
		return nil
	})
	if err == nil {
		s.log.Info("model fallbacks set", "model", name, "fallbacks", fallbacks, "by", by)
	}
	return final, err
}

// SetCheck runs before a model is added, and refuses it by returning an error.
// Used to catch weights the engine cannot load, which otherwise are placed and
// then fail on every node they are tried on.
func (s *Service) SetCheck(check func(context.Context, catalog.Model) error) { s.check = check }