noema/internal/brain/manager.go

package brain

import (
	"context"
	"errors"
	"strings"
	"sync"
	"time"

	"github.com/andydixon/noema/internal/minds"
	"github.com/andydixon/noema/internal/scheduler"
	"github.com/andydixon/noema/pkg/ids"
)

// Manager owns one Runtime per active mind, wakes them on new events and
// reconciles with the minds table so enabling or disabling a mind in the UI
// takes effect without a restart.
type Manager struct {
	cfg      Config
	group    *scheduler.Group
	mu       sync.RWMutex
	runtimes map[ids.ID]*entry
}

type entry struct {
	rt     *Runtime
	cancel context.CancelFunc
	done   chan struct{}
}

// ErrNotRunning is returned for minds without a live runtime.
var ErrNotRunning = errors.New("mind runtime is not running")

// NewManager creates a manager; call Start to launch runtimes.
func NewManager(cfg Config, group *scheduler.Group) *Manager {
	return &Manager{cfg: cfg, group: group, runtimes: map[ids.ID]*entry{}}
}

// Start launches runtimes for active minds, subscribes to the event bus for
// wake-ups and reconciles every 30s.
func (m *Manager) Start(ctx context.Context) error {
	if err := m.Reconcile(ctx); err != nil {
		return err
	}
	ch, err := m.cfg.Events.Bus().Subscribe(m.group.Context(), ids.Nil, 1024)
	if err != nil {
		return err
	}
	m.group.Go("brain.wake", func(ctx context.Context) error {
		for {
			select {
			case <-ctx.Done():
				return nil
			case ev := <-ch:
				if strings.HasPrefix(ev.Source, "brain") {
					continue // our own output never re-triggers a cycle (T20)
				}
				if rt := m.Get(ev.MindID); rt != nil {
					rt.Wake()
				}
			}
		}
	})
	m.group.Every("brain.reconcile", 30*time.Second, m.Reconcile)
	return nil
}

// Reconcile starts runtimes for active minds and stops the rest.
func (m *Manager) Reconcile(ctx context.Context) error {
	list, err := m.cfg.Minds.List(ctx, true)
	if err != nil {
		return err
	}
	want := map[ids.ID]minds.Mind{}
	for _, md := range list {
		if md.Status == minds.StatusActive {
			want[md.ID] = md
		}
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	for id, e := range m.runtimes {
		if _, ok := want[id]; !ok {
			e.cancel()
			<-e.done
			delete(m.runtimes, id)
			m.cfg.Log.Info("mind runtime stopped", "mind", id)
		}
	}
	for id, md := range want {
		if _, ok := m.runtimes[id]; ok {
			continue
		}
		rt, err := NewRuntime(ctx, m.cfg, md)
		if err != nil {
			m.cfg.Log.Error("cannot start mind runtime", "mind", md.Slug, "err", err)
			continue
		}
		rctx, cancel := context.WithCancel(m.group.Context())
		e := &entry{rt: rt, cancel: cancel, done: make(chan struct{})}
		m.runtimes[id] = e
		m.group.Go("brain."+md.Slug, func(context.Context) error {
			defer close(e.done)
			return rt.Run(rctx)
		})
	}
	return nil
}

// Get returns the runtime for a mind or nil.
func (m *Manager) Get(id ids.ID) *Runtime {
	m.mu.RLock()
	defer m.mu.RUnlock()
	if e, ok := m.runtimes[id]; ok {
		return e.rt
	}
	return nil
}

// Command forwards an operator command to a mind's runtime.
func (m *Manager) Command(ctx context.Context, id ids.ID, kind string, args map[string]bool) error {
	rt := m.Get(id)
	if rt == nil {
		return ErrNotRunning
	}
	return rt.Command(ctx, kind, args)
}

// Running lists the ids of minds with live runtimes.
func (m *Manager) Running() []ids.ID {
	m.mu.RLock()
	defer m.mu.RUnlock()
	out := make([]ids.ID, 0, len(m.runtimes))
	for id := range m.runtimes {
		out = append(out, id)
	}
	return out
}