kenn-io--agentsview/internal/vector/manager_test.go

package vector

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	kitvec "go.kenn.io/kit/vector"
	"go.kenn.io/kit/vector/sqlitevec"
)

// soloEncoders wraps enc as a single-server EncoderSet, the shape most
// Manager tests need.
func soloEncoders(enc kitvec.EncodeFunc) EncoderSet {
	return EncoderSet{Default: "test", ByName: map[string]ManagedEncoder{
		"test": {Encode: enc, Settings: EncodeSettings{BatchSize: 10}},
	}}
}

// blockingEncoder returns an encoder that blocks until release is closed,
// letting tests observe a Manager while its build is still in flight.
func blockingEncoder(release <-chan struct{}) kitvec.EncodeFunc {
	return func(_ context.Context, texts []string) ([][]float32, error) {
		<-release
		out := make([][]float32, len(texts))
		for i := range texts {
			out[i] = []float32{1, 0, 0}
		}
		return out, nil
	}
}

// waitFor polls cond until it returns true or the deadline passes, failing
// the test otherwise. Used instead of a fixed sleep to avoid flakiness.
func waitFor(t *testing.T, cond func() bool, msg string) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(time.Millisecond)
	}
	require.Fail(t, "timed out waiting for condition", msg)
}

// generationIDByFingerprint looks up a generation's CLI-facing ordinal ID
// from its fingerprint, for tests that need to Activate/Retire a specific
// generation by ID.
func generationIDByFingerprint(t *testing.T, ix *Index, fp string) int64 {
	t.Helper()
	gens, err := ix.Generations(context.Background())
	require.NoError(t, err)
	for _, g := range gens {
		if g.Fingerprint == fp {
			return g.ID
		}
	}
	require.Fail(t, "generation not found for fingerprint", fp)
	return 0
}

// countingEncoder returns a working encoder that counts its calls, so
// tests can tell which EncoderSet entry a build actually used.
func countingEncoder(calls *atomic.Int64) kitvec.EncodeFunc {
	return func(_ context.Context, texts []string) ([][]float32, error) {
		calls.Add(1)
		out := make([][]float32, len(texts))
		for i := range texts {
			out[i] = []float32{1, 0, 0}
		}
		return out, nil
	}
}

func TestManagerBuildUsingSelectsNamedEncoder(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")

	var localCalls, remoteCalls atomic.Int64
	encoders := EncoderSet{Default: "local", ByName: map[string]ManagedEncoder{
		"local":  {Encode: countingEncoder(&localCalls), Settings: EncodeSettings{BatchSize: 10}},
		"remote": {Encode: countingEncoder(&remoteCalls), Settings: EncodeSettings{BatchSize: 10}},
	}}
	m := NewManager(ix, src, encoders, gen)

	started, err := m.TryBuild(context.Background(), BuildRequest{Using: "remote"})
	require.NoError(t, err)
	require.True(t, started)
	assert.Positive(t, remoteCalls.Load(), "build with Using must encode on the named server")
	assert.Zero(t, localCalls.Load(), "the default server must stay idle")

	started, err = m.TryBuild(context.Background(), BuildRequest{FullRebuild: true})
	require.NoError(t, err)
	require.True(t, started)
	assert.Positive(t, localCalls.Load(), "a build without Using must encode on the default server")
}

func TestManagerBuildUnknownUsingFailsBeforeStarting(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), gen)

	err := m.StartBuild(BuildRequest{Using: "nope"})
	require.ErrorContains(t, err, `no embeddings server named "nope"`)
	assert.ErrorIs(t, err, ErrUnknownServer,
		"callers map unknown-server errors to a client error via the sentinel")
	assert.False(t, m.Status().Running, "a failed resolve must not leave the manager running")

	started, err := m.TryBuild(context.Background(), BuildRequest{Using: "nope"})
	require.ErrorContains(t, err, `no embeddings server named "nope"`)
	assert.ErrorIs(t, err, ErrUnknownServer)
	assert.False(t, started)
}

func TestManagerStartBuildSetsRunningAndConcurrentStartReturnsErrBuildRunning(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	release := make(chan struct{})
	m := NewManager(ix, src, soloEncoders(blockingEncoder(release)), gen)

	require.NoError(t, m.StartBuild(BuildRequest{}))
	waitFor(t, func() bool { return m.Status().Running }, "build never reported running")

	err := m.StartBuild(BuildRequest{})
	assert.ErrorIs(t, err, ErrBuildRunning)

	close(release)
	waitFor(t, func() bool { return !m.Status().Running }, "build never finished")
	assert.Empty(t, m.Status().LastError)
}

func TestManagerTryBuildReturnsFalseWhileRunning(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	release := make(chan struct{})
	m := NewManager(ix, src, soloEncoders(blockingEncoder(release)), gen)

	require.NoError(t, m.StartBuild(BuildRequest{}))
	waitFor(t, func() bool { return m.Status().Running }, "build never reported running")

	started, err := m.TryBuild(context.Background(), BuildRequest{})
	assert.False(t, started, "TryBuild must drop rather than queue while running")
	assert.NoError(t, err)

	close(release)
	waitFor(t, func() bool { return !m.Status().Running }, "build never finished")
}

func TestManagerStatusTransitionsToLastResultOnCompletion(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), gen)

	require.NoError(t, m.StartBuild(BuildRequest{}))
	waitFor(t, func() bool { return !m.Status().Running }, "build never finished")

	status := m.Status()
	require.NotNil(t, status.LastResult)
	assert.Equal(t, gen.Fingerprint(), status.LastResult.Fingerprint)
	assert.True(t, status.LastResult.Activated)
	assert.Empty(t, status.LastError)
}

func TestManagerStatusSetsLastErrorOnEncoderFailure(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	failingEncoder := func(_ context.Context, _ []string) ([][]float32, error) {
		return nil, fmt.Errorf("encoder rejected input")
	}
	m := NewManager(ix, src, soloEncoders(failingEncoder), gen)

	require.NoError(t, m.StartBuild(BuildRequest{}))
	waitFor(t, func() bool { return !m.Status().Running }, "build never finished")

	status := m.Status()
	assert.Contains(t, status.LastError, "encoder rejected input")
	assert.Nil(t, status.LastResult)
}

func TestManagerStatusStampsBuildIdentityAndSpace(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), gen)

	base := time.Date(2026, 7, 11, 10, 0, 0, 0, time.UTC)
	m.now = func() time.Time { return base }

	status := m.Status()
	assert.Zero(t, status.BuildID, "no build has started yet")
	assert.Empty(t, status.StartedAt)
	assert.Equal(t, "fake-model", status.Model,
		"the configured space is reported even before any build")
	assert.Equal(t, 3, status.Dimension)

	started, err := m.TryBuild(context.Background(), BuildRequest{})
	require.NoError(t, err)
	require.True(t, started)

	status = m.Status()
	assert.Equal(t, int64(1), status.BuildID)
	assert.Equal(t, "2026-07-11T10:00:00Z", status.StartedAt)

	m.now = func() time.Time { return base.Add(time.Minute) }
	started, err = m.TryBuild(context.Background(), BuildRequest{FullRebuild: true})
	require.NoError(t, err)
	require.True(t, started)

	status = m.Status()
	assert.Equal(t, int64(2), status.BuildID,
		"each build start must get a fresh identity so pollers can tell builds apart")
	assert.Equal(t, "2026-07-11T10:01:00Z", status.StartedAt)
}

func TestManagerGenerationsDelegatesToIndex(t *testing.T) {
	ix := openTestIndex(t)
	ctx := context.Background()
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), gen)

	_, err := m.TryBuild(ctx, BuildRequest{})
	require.NoError(t, err)

	want, err := ix.Generations(ctx)
	require.NoError(t, err)
	got, err := m.Generations(ctx)
	require.NoError(t, err)
	assert.Equal(t, want, got)
}

func TestManagerActivateForceRefusalMatrix(t *testing.T) {
	ix := openTestIndex(t)
	ctx := context.Background()
	src := twoDocSource()
	genA := fakeGeneration("model-a")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), genA)

	_, err := m.TryBuild(ctx, BuildRequest{})
	require.NoError(t, err, "genA becomes active")

	genB := fakeGeneration("model-b")
	fpB, err := ix.EnsureGeneration(ctx, genB, sqlitevec.StateBuilding)
	require.NoError(t, err, "genB registered but never filled, so it has Missing > 0")
	idB := generationIDByFingerprint(t, ix, fpB)

	err = m.Activate(ctx, idB, false)
	require.Error(t, err, "refuses activation of an incompletely embedded generation")
	assert.Contains(t, err.Error(), fmt.Sprintf("generation %d still has", idB))
	assert.Contains(t, err.Error(), "use --force")

	require.NoError(t, m.Activate(ctx, idB, true), "force overrides the refusal")

	active, ok, err := ix.ActiveFingerprint(ctx)
	require.NoError(t, err)
	require.True(t, ok)
	assert.Equal(t, fpB, active, "genB is now active")

	idA := generationIDByFingerprint(t, ix, genA.Fingerprint())
	gens, err := ix.Generations(ctx)
	require.NoError(t, err)
	var stateA string
	for _, g := range gens {
		if g.ID == idA {
			stateA = g.State
		}
	}
	assert.Equal(t, string(sqlitevec.StateRetired), stateA, "activating genB retires the old active genA")
}

func TestManagerRetireRefusesActiveGenerationWithoutForce(t *testing.T) {
	ix := openTestIndex(t)
	ctx := context.Background()
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), gen)

	_, err := m.TryBuild(ctx, BuildRequest{})
	require.NoError(t, err)
	id := generationIDByFingerprint(t, ix, gen.Fingerprint())

	err = m.Retire(ctx, id, false)
	require.Error(t, err, "refuses retiring the active generation without force")

	require.NoError(t, m.Retire(ctx, id, true), "force overrides the refusal")

	gens, err := ix.Generations(ctx)
	require.NoError(t, err)
	require.Len(t, gens, 1)
	assert.Equal(t, string(sqlitevec.StateRetired), gens[0].State)
}

// countActiveGenerations returns how many generations are currently in the
// active state, the invariant concurrent Activate calls must preserve (== 1
// once any generation has been activated).
func countActiveGenerations(t *testing.T, ix *Index) int {
	t.Helper()
	gens, err := ix.Generations(context.Background())
	require.NoError(t, err)
	active := 0
	for _, g := range gens {
		if g.State == string(sqlitevec.StateActive) {
			active++
		}
	}
	return active
}

// TestManagerConcurrentActivateNeverLeavesTwoActive guards the
// retire-then-activate invariant: Activate must use the single-transaction
// activateGeneration primitive and serialize against other Activate calls,
// or two racing Activates on different generations can interleave their
// retire and activate steps and leave both generations active.
func TestManagerConcurrentActivateNeverLeavesTwoActive(t *testing.T) {
	ix := openTestIndex(t)
	ctx := context.Background()
	src := twoDocSource()
	genA := fakeGeneration("model-a")
	genB := fakeGeneration("model-b")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), genA)

	_, err := ix.Build(ctx, src, fakeBuildEncoder(), genA, BuildOptions{})
	require.NoError(t, err)
	_, err = ix.Build(ctx, src, fakeBuildEncoder(), genB, BuildOptions{})
	require.NoError(t, err, "both generations fully embedded; genB active, genA retired")

	idA := generationIDByFingerprint(t, ix, genA.Fingerprint())
	idB := generationIDByFingerprint(t, ix, genB.Fingerprint())

	for i := range 25 {
		var wg sync.WaitGroup
		var errA, errB error
		wg.Add(2)
		go func() {
			defer wg.Done()
			errA = m.Activate(ctx, idA, false)
		}()
		go func() {
			defer wg.Done()
			errB = m.Activate(ctx, idB, false)
		}()
		wg.Wait()
		require.NoError(t, errA, "iteration %d", i)
		require.NoError(t, errB, "iteration %d", i)
		require.Equal(t, 1, countActiveGenerations(t, ix),
			"iteration %d: exactly one generation must be active after racing Activates", i)
	}
}

// TestManagerStartBuildRecoversPanickedEncoder guards the daemon against a
// panic in the caller-supplied encoder (a network client): StartBuild's
// detached goroutine must recover, record the panic in LastError, and clear
// the running state instead of crashing the process.
func TestManagerStartBuildRecoversPanickedEncoder(t *testing.T) {
	ix := openTestIndex(t)
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	panickingEncoder := func(_ context.Context, _ []string) ([][]float32, error) {
		panic("encoder exploded")
	}
	m := NewManager(ix, src, soloEncoders(panickingEncoder), gen)

	require.NoError(t, m.StartBuild(BuildRequest{}))
	waitFor(t, func() bool { return !m.Status().Running }, "build never finished after panic")

	status := m.Status()
	assert.Contains(t, status.LastError, "panicked")
	assert.Contains(t, status.LastError, "encoder exploded")
	assert.Nil(t, status.LastResult)

	require.NoError(t, m.StartBuild(BuildRequest{}),
		"manager must accept a new build after a panicked one")
	waitFor(t, func() bool { return !m.Status().Running }, "second build never finished")
}

// TestManagerActivateAndRetireUnknownIDPropagateNotFound guards the HTTP
// route mapping (embeddingsActionError in internal/server): Activate and
// Retire must propagate GenerationByID's ErrGenerationNotFound unwrapped
// enough for errors.Is to still match it, rather than losing the sentinel
// on the way up.
func TestManagerActivateAndRetireUnknownIDPropagateNotFound(t *testing.T) {
	ix := openTestIndex(t)
	ctx := context.Background()
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	m := NewManager(ix, src, soloEncoders(fakeBuildEncoder()), gen)

	err := m.Activate(ctx, 999, false)
	assert.ErrorIs(t, err, ErrGenerationNotFound)

	err = m.Retire(ctx, 999, false)
	assert.ErrorIs(t, err, ErrGenerationNotFound)
}

func TestManagerActivateAndRetireRefuseWhileBuildRunning(t *testing.T) {
	ix := openTestIndex(t)
	ctx := context.Background()
	src := twoDocSource()
	gen := fakeGeneration("fake-model")
	release := make(chan struct{})
	m := NewManager(ix, src, soloEncoders(blockingEncoder(release)), gen)

	require.NoError(t, m.StartBuild(BuildRequest{}))
	waitFor(t, func() bool { return m.Status().Running }, "build never reported running")

	err := m.Activate(ctx, 1, true)
	assert.ErrorIs(t, err, ErrBuildRunning)

	err = m.Retire(ctx, 1, true)
	assert.ErrorIs(t, err, ErrBuildRunning)

	close(release)
	waitFor(t, func() bool { return !m.Status().Running }, "build never finished")
}