package main import ( "context" "database/sql" "errors" "fmt" "io" "mime" "net" "net/http" "net/http/httptest" "os" "path/filepath" "sync" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.kenn.io/agentsview/internal/config" "go.kenn.io/agentsview/internal/db" "go.kenn.io/agentsview/internal/dbtest" "go.kenn.io/agentsview/internal/server" "go.kenn.io/agentsview/internal/vector" ) // --- fake embedManager --- // fakeEmbedManager records every TryBuild call and returns scripted // (started, err) results in order, repeating the last scripted result once // the script is exhausted (default: started=true, err=nil). type fakeEmbedManager struct { mu sync.Mutex calls []vector.BuildRequest results []fakeTryBuildResult } type fakeTryBuildResult struct { started bool err error } func (f *fakeEmbedManager) TryBuild( _ context.Context, req vector.BuildRequest, ) (bool, error) { f.mu.Lock() defer f.mu.Unlock() f.calls = append(f.calls, req) idx := len(f.calls) - 1 if idx < len(f.results) { r := f.results[idx] return r.started, r.err } if len(f.results) > 0 { r := f.results[len(f.results)-1] return r.started, r.err } return true, nil } func (f *fakeEmbedManager) callCount() int { f.mu.Lock() defer f.mu.Unlock() return len(f.calls) } func (f *fakeEmbedManager) callsSnapshot() []vector.BuildRequest { f.mu.Lock() defer f.mu.Unlock() return append([]vector.BuildRequest(nil), f.calls...) } // waitForSchedulerCondition polls cond until it is true or 2s pass, failing // the test with msg otherwise. Avoids fixed sleeps that would either flake // under load or slow the suite down needlessly. func waitForSchedulerCondition(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) } func TestEmbedSchedulerBurstOfNotifyProducesExactlyOneBuild(t *testing.T) { fake := &fakeEmbedManager{} s := newEmbedScheduler(fake, 20*time.Millisecond, 0, false) ctx := t.Context() go s.Run(ctx) defer s.Stop() for range 10 { s.Notify() time.Sleep(2 * time.Millisecond) } waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "expected a build after the burst quieted") // Give any spurious extra build a chance to show up before asserting // there is exactly one. time.Sleep(60 * time.Millisecond) assert.Equal(t, 1, fake.callCount(), "a burst of Notify must collapse to one build") assert.Equal(t, []vector.BuildRequest{{}}, fake.callsSnapshot()) } // TestEmbedSchedulerIncludeAutomatedThreadsIntoBuildRequests asserts the // scheduler's configured include-automated scope rides along on every // BuildRequest it issues -- both the debounced after-sync path and the // backstop ticker -- so scheduled builds stay config-authoritative rather // than silently reverting to includeAutomated=false. func TestEmbedSchedulerIncludeAutomatedThreadsIntoBuildRequests(t *testing.T) { fake := &fakeEmbedManager{} s := newEmbedScheduler(fake, 5*time.Millisecond, 20*time.Millisecond, true) ctx := t.Context() go s.Run(ctx) defer s.Stop() s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "expected a debounced build") waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 2 }, "expected a backstop build") for _, req := range fake.callsSnapshot() { assert.True(t, req.IncludeAutomated, "every scheduler-issued BuildRequest must carry the configured scope") } } func TestEmbedSchedulerNotifyDuringRunningBuildRearmsForFollowUpPass(t *testing.T) { fake := &fakeEmbedManager{ results: []fakeTryBuildResult{ {started: false, err: nil}, // a build is already running elsewhere {started: true, err: nil}, // the follow-up pass actually runs }, } s := newEmbedScheduler(fake, 15*time.Millisecond, 0, false) ctx := t.Context() go s.Run(ctx) defer s.Stop() s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 2 }, "expected the scheduler to re-arm and retry after a dropped build") calls := fake.callsSnapshot() require.Len(t, calls, 2) assert.Equal(t, vector.BuildRequest{}, calls[0]) assert.Equal(t, vector.BuildRequest{}, calls[1]) } func TestEmbedSchedulerBackstopTickIssuesBackstopBuild(t *testing.T) { fake := &fakeEmbedManager{} // A very long debounce so only the backstop ticker can fire a build. s := newEmbedScheduler(fake, time.Hour, 20*time.Millisecond, false) ctx := t.Context() go s.Run(ctx) defer s.Stop() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "expected a backstop build") calls := fake.callsSnapshot() require.NotEmpty(t, calls) assert.True(t, calls[0].Backstop, "backstop tick must set BuildRequest.Backstop") } // TestEmbedSchedulerDroppedBackstopRetriesOnNextDebouncedBuild is the fix-4 // regression test: a backstop tick that collides with a build already // running elsewhere must not be silently dropped for a full backstop // interval (24h in production). The scheduler must remember it and fold // Backstop: true into the next debounced build request instead. func TestEmbedSchedulerDroppedBackstopRetriesOnNextDebouncedBuild(t *testing.T) { fake := &fakeEmbedManager{ results: []fakeTryBuildResult{ {started: false, err: nil}, // the backstop tick collides with a build elsewhere {started: true, err: nil}, // the following debounced build recovers it }, } // A long backstop interval relative to the debounce interval and the // test's own buffers keeps a second, unrelated backstop tick from // firing mid-test and making the call count non-deterministic. s := newEmbedScheduler(fake, 10*time.Millisecond, 500*time.Millisecond, false) ctx := t.Context() go s.Run(ctx) defer s.Stop() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "expected the backstop tick to fire and be dropped") s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 2 }, "expected the debounced build to recover the dropped backstop") time.Sleep(50 * time.Millisecond) calls := fake.callsSnapshot() require.Len(t, calls, 2, "the dropped backstop must be retried exactly once, not repeatedly") assert.True(t, calls[0].Backstop, "the original (dropped) backstop tick request") assert.True(t, calls[1].Backstop, "the debounced build must carry the pending backstop forward instead of dropping it") // Once the recovered build actually started, the pending flag must // clear: a further, unrelated debounced build must not keep carrying // Backstop: true forever. s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 3 }, "expected a further debounced build after the recovered one") time.Sleep(50 * time.Millisecond) calls = fake.callsSnapshot() require.Len(t, calls, 3) assert.False(t, calls[2].Backstop, "the recovered backstop must not leak into a later unrelated debounced build") } // TestEmbedSchedulerBackstopTickStartedButFailedKeepsPendingBackstop is the // fix-5 regression test: a backstop tick whose TryBuild call actually started // but then returned an error must not clear pendingBackstop -- the // reconciliation it carried never completed, so it must be retried on the // very next debounced build rather than silently deferred to the next // backstop interval (24h in production). func TestEmbedSchedulerBackstopTickStartedButFailedKeepsPendingBackstop(t *testing.T) { buildErr := errors.New("embeddings endpoint unreachable") fake := &fakeEmbedManager{ results: []fakeTryBuildResult{ {started: true, err: buildErr}, // the backstop tick starts but fails {started: true, err: nil}, // the following debounced build recovers it }, } s := newEmbedScheduler(fake, 10*time.Millisecond, 500*time.Millisecond, false) ctx := t.Context() go s.Run(ctx) defer s.Stop() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "expected the backstop tick to fire and fail") s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 2 }, "expected the debounced build to retry the failed backstop") time.Sleep(50 * time.Millisecond) calls := fake.callsSnapshot() require.Len(t, calls, 2, "the failed backstop must be retried exactly once, not repeatedly") assert.True(t, calls[0].Backstop, "the original (started-but-failed) backstop tick request") assert.True(t, calls[1].Backstop, "the debounced build must carry the failed backstop forward instead of dropping it") // Once the recovered build actually succeeded, the pending flag must // clear: a further, unrelated debounced build must not keep carrying // Backstop: true forever. s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 3 }, "expected a further debounced build after the recovered one") time.Sleep(50 * time.Millisecond) calls = fake.callsSnapshot() require.Len(t, calls, 3) assert.False(t, calls[2].Backstop, "the recovered backstop must not leak into a later unrelated debounced build") } // TestEmbedSchedulerDebouncedBuildStartedButFailedKeepsPendingBackstop is the // fix-5 regression test for the debounced-build path: once a dropped // backstop tick is being carried by pendingBackstop, a debounced build that // starts but then fails must not clear it either -- the same // started-but-failed rule applies on both paths that can clear the flag. func TestEmbedSchedulerDebouncedBuildStartedButFailedKeepsPendingBackstop(t *testing.T) { buildErr := errors.New("embeddings endpoint unreachable") fake := &fakeEmbedManager{ results: []fakeTryBuildResult{ {started: false, err: nil}, // the backstop tick collides with a build elsewhere {started: true, err: buildErr}, // the recovering debounced build starts but fails {started: true, err: nil}, // a further debounced build finally succeeds }, } s := newEmbedScheduler(fake, 10*time.Millisecond, 500*time.Millisecond, false) ctx := t.Context() go s.Run(ctx) defer s.Stop() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "expected the backstop tick to fire and be dropped") s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 2 }, "expected the debounced build to attempt recovering the dropped backstop") s.Notify() waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 3 }, "expected a further debounced build to retry after the recovering build failed") time.Sleep(50 * time.Millisecond) calls := fake.callsSnapshot() require.Len(t, calls, 3) assert.True(t, calls[0].Backstop, "the original (dropped) backstop tick request") assert.True(t, calls[1].Backstop, "the recovering (started-but-failed) debounced build") assert.True(t, calls[2].Backstop, "a started-but-failed build must not clear pendingBackstop: the retry must still carry it") } func TestEmbedSchedulerStopTerminatesRun(t *testing.T) { fake := &fakeEmbedManager{} s := newEmbedScheduler(fake, time.Hour, 0, false) go s.Run(context.Background()) // Stop blocks until Run has actually exited, so its returning at all // (within a generous timeout) is the proof Run terminated. stopped := make(chan struct{}) go func() { s.Stop() close(stopped) }() select { case <-stopped: case <-time.After(2 * time.Second): require.Fail(t, "Stop did not return; Run likely never terminated") } } func TestEmbedSchedulerNotifyNeverBlocksWithoutAReader(t *testing.T) { fake := &fakeEmbedManager{} s := newEmbedScheduler(fake, time.Hour, 0, false) done := make(chan struct{}) go func() { for range 100 { s.Notify() } close(done) }() select { case <-done: case <-time.After(2 * time.Second): require.Fail(t, "Notify blocked with no Run consuming dirty") } } // --- teeEmitter --- type recordingEmitter struct { mu sync.Mutex scopes []string } func (e *recordingEmitter) Emit(scope string) { e.mu.Lock() defer e.mu.Unlock() e.scopes = append(e.scopes, scope) } func (e *recordingEmitter) count() int { e.mu.Lock() defer e.mu.Unlock() return len(e.scopes) } func TestTeeEmitterAlwaysCallsPrimaryAndGatesSchedulerOnRunAfterSync(t *testing.T) { primary := &recordingEmitter{} fake := &fakeEmbedManager{} s := newEmbedScheduler(fake, 10*time.Millisecond, 0, false) ctx := t.Context() go s.Run(ctx) defer s.Stop() disabled := teeEmitter{primary: primary, scheduler: s, runAfterSync: false} disabled.Emit("sessions") assert.Equal(t, 1, primary.count()) time.Sleep(30 * time.Millisecond) assert.Equal(t, 0, fake.callCount(), "runAfterSync=false must not notify the scheduler") enabled := teeEmitter{primary: primary, scheduler: s, runAfterSync: true} enabled.Emit("sessions") assert.Equal(t, 2, primary.count()) waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "runAfterSync=true must notify the scheduler") } // TestRunRemoteHostSyncLoop_EmitsThroughTeeNotifiesScheduler is a // regression test for a bug where runServe wired startPeriodicSync's // scheduled remote-host sync path to the bare SSE broadcaster instead of // the wrapped teeEmitter, so remote-synced sessions notified SSE clients // but never reached the embed scheduler until an unrelated local sync or // the 24h backstop ran. It drives runRemoteHostSyncLoop — the function // startPeriodicSync spawns per configured remote host — with a teeEmitter // and a syncFn reporting synced sessions, and asserts the scheduler // actually receives a build trigger through that path. func TestRunRemoteHostSyncLoop_EmitsThroughTeeNotifiesScheduler(t *testing.T) { primary := &recordingEmitter{} fake := &fakeEmbedManager{} s := newEmbedScheduler(fake, 5*time.Millisecond, 0, false) schedCtx := t.Context() go s.Run(schedCtx) defer s.Stop() tee := teeEmitter{primary: primary, scheduler: s, runAfterSync: true} syncFn := func() (int, error) { return 1, nil // one session synced on this remote host } loopCtx := t.Context() done := make(chan struct{}) go runRemoteHostSyncLoop( loopCtx, "remote-host", 5*time.Millisecond, syncFn, tee, nil, done, ) defer close(done) waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 }, "a scheduled remote sync reaching the tee emitter must notify the embed scheduler") assert.NotZero(t, primary.count(), "the tee must still forward remote sync completions to the SSE broadcaster") } // --- searcherAdapter error taxonomy --- func TestTranslateSearchErrorMapsVectorErrorsToSemanticUnavailable(t *testing.T) { t.Run("no active generation", func(t *testing.T) { err := translateSearchError(vector.ErrNoActiveGeneration) assert.ErrorIs(t, err, db.ErrSemanticUnavailable) }) t.Run("building", func(t *testing.T) { err := translateSearchError(&vector.BuildingError{Percent: 62}) assert.ErrorIs(t, err, db.ErrSemanticUnavailable) assert.Contains(t, err.Error(), "index is building: 62% complete") }) t.Run("other error passes through", func(t *testing.T) { boom := errors.New("boom") assert.Same(t, boom, translateSearchError(boom)) }) t.Run("query encode failure maps to semantic transient, not unavailable", func(t *testing.T) { queryErr := &vector.QueryEncodeError{Err: errors.New("dial tcp: connection refused")} got := translateSearchError(queryErr) assert.ErrorIs(t, got, db.ErrSemanticTransient) assert.False(t, errors.Is(got, db.ErrSemanticUnavailable), "a query-time endpoint failure must not read as semantic search being disabled") assert.Contains(t, got.Error(), "connection refused") }) t.Run("query encode failure preserves the underlying cause chain", func(t *testing.T) { queryErr := &vector.QueryEncodeError{ Err: fmt.Errorf("encoding query: %w", context.Canceled), } got := translateSearchError(queryErr) assert.ErrorIs(t, got, db.ErrSemanticTransient) assert.ErrorIs(t, got, context.Canceled, "context errors must stay matchable so cancellation handling still fires") }) t.Run("mirror version mismatch maps to semantic unavailable with rebuild message", func(t *testing.T) { got := translateSearchError( fmt.Errorf("checking embedding index staleness: %w", vector.ErrMirrorVersionMismatch)) assert.ErrorIs(t, got, db.ErrSemanticUnavailable) assert.Contains(t, got.Error(), "embeddings build", "the rebuild remediation must survive translation") }) } // TestSearcherAdapterVersionMismatchedIndexReturnsSemanticUnavailable is the // adapter-level regression test for the mirror version gate: a // searcherAdapter over a read-only vectors.db written by a different mirror // schema version must return an error matching db.ErrSemanticUnavailable and // mentioning the rebuild remediation — not a raw SQL error or a wrong // staleness verdict from StaleActive querying an incompatible mirror. func TestSearcherAdapterVersionMismatchedIndexReturnsSemanticUnavailable(t *testing.T) { dataDir := t.TempDir() cfg := vectorTestConfig(dataDir) path := cfg.Vector.ResolvedDBPath(dataDir) // Create a current vectors.db, then restamp it as written by the // previous mirror schema version, simulating a file left behind by an // older agentsview build. seed, err := vector.Open(context.Background(), path, false, cfg.Vector.Embeddings.MaxInputChars) require.NoError(t, err) require.NoError(t, seed.Close()) raw, err := sql.Open("sqlite3", path) require.NoError(t, err) _, err = raw.Exec(`UPDATE vector_meta SET value = '2' WHERE key = 'mirror_schema_version'`) require.NoError(t, err) require.NoError(t, raw.Close()) ix, err := vector.Open(context.Background(), path, true, cfg.Vector.Embeddings.MaxInputChars) require.NoError(t, err, "read-only Open must succeed against a mismatched vectors.db") defer ix.Close() enc, err := newVectorEncoder(cfg.Vector.Embeddings, "") require.NoError(t, err) adapter := newSearcherAdapter(ix, enc, vectorGeneration(cfg.Vector.Embeddings)) _, err = adapter.SemanticSearch(context.Background(), "any query", 5) require.Error(t, err) assert.ErrorIs(t, err, db.ErrSemanticUnavailable, "a version-mismatched index must surface the semantic-unavailable taxonomy") assert.Contains(t, err.Error(), "embeddings build", "the error must tell the user to rebuild the index") } // --- integration: real serve/server construction path --- // vectorTestConfig returns a config.Config with a small-but-valid [vector] // section (accepted by config.Validate) pointed at an embeddings endpoint // that is never actually called in these tests — setupVectorServing only // constructs the encoder, it does not exercise it. func vectorTestConfig(dataDir string) config.Config { return config.Config{ DataDir: dataDir, DBPath: filepath.Join(dataDir, "sessions.db"), Vector: config.VectorConfig{ Enabled: true, Embeddings: config.VectorEmbeddingsConfig{ Model: "test-model", Dimension: 3, MaxInputChars: 1000, Servers: map[string]config.VectorEmbeddingsServerConfig{ "local": { Endpoint: "http://127.0.0.1:1/v1", BatchSize: 10, Concurrency: 1, Timeout: "5s", MaxRetries: 1, }, }, }, Embed: config.VectorEmbedConfig{ BackstopInterval: "24h", }, }, } } // listenLoopback opens a loopback listener on an OS-assigned port, // returning it alongside the port so a caller can set cfg.Port to it // before constructing the server: the host-check middleware validates // the Host header against cfg.Host:cfg.Port exactly, so the request's // destination port must be known up front rather than assigned by // httptest.NewServer after the fact. func listenLoopback(t *testing.T) (net.Listener, int) { t.Helper() ln, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) return ln, ln.Addr().(*net.TCPAddr).Port } // startTestServer serves handler on ln (already bound to the port cfg.Port // was set to) and returns an *httptest.Server whose URL and lifecycle // helpers work as usual. func startTestServer(t *testing.T, ln net.Listener, handler http.Handler) *httptest.Server { t.Helper() ts := httptest.NewUnstartedServer(handler) require.NoError(t, ts.Listener.Close()) ts.Listener = ln ts.Start() t.Cleanup(ts.Close) return ts } // TestServeConstructionRegistersEmbeddingsRoutesWhenVectorEnabled drives the // real setupVectorServing + server.New construction path (not a fake mux) // with a vector-enabled config and asserts the embeddings status endpoint // responds, then a sibling test rebuilds with vector disabled and asserts // it does not. func TestServeConstructionRegistersEmbeddingsRoutesWhenVectorEnabled(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) ln, port := listenLoopback(t) cfg := vectorTestConfig(dataDir) cfg.Host, cfg.Port = "127.0.0.1", port vs, err := setupVectorServing(context.Background(), cfg, database) require.NoError(t, err) require.NotNil(t, vs.Scheduler) require.NotNil(t, vs.Close) defer func() { require.NoError(t, vs.Close()) }() srv := server.New(cfg, database, nil, vs.ServerOpts...) ts := startTestServer(t, ln, srv.Handler()) resp, err := http.Get(ts.URL + "/api/v1/embeddings/status") require.NoError(t, err) defer resp.Body.Close() assert.Equal(t, http.StatusOK, resp.StatusCode) assert.Equal(t, "application/json", mediaType(t, resp.Header.Get("Content-Type"))) } // TestEmbeddingsDaemonClientBuildSucceedsThroughRealMiddleware drives a POST // build request from embeddingsDaemonClient (the same client `embeddings // build` uses to talk to a running daemon) through the server's real // middleware chain (srv.Handler(), not a bare mux), proving the CSRF guard // in internal/server/server.go's corsMiddleware does not reject it. Every // other embeddings-daemon-dispatch test in embeddings_test.go serves off an // httptest.NewServeMux() directly, bypassing that middleware entirely, which // is exactly how the missing Origin header regression went unnoticed. func TestEmbeddingsDaemonClientBuildSucceedsThroughRealMiddleware(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) ln, port := listenLoopback(t) cfg := vectorTestConfig(dataDir) cfg.Host, cfg.Port = "127.0.0.1", port vs, err := setupVectorServing(context.Background(), cfg, database) require.NoError(t, err) defer func() { require.NoError(t, vs.Close()) }() srv := server.New(cfg, database, nil, vs.ServerOpts...) ts := startTestServer(t, ln, srv.Handler()) client := embeddingsDaemonClient{baseURL: ts.URL} err = client.startBuild(context.Background(), vector.BuildRequest{}) require.NoError(t, err, "a POST build must succeed once the client sets Origin to satisfy the CSRF guard") } // TestSetupVectorServingDisablesWhenWriteLockHeld asserts a held // vectors.write.lock (simulating a concurrent direct `embeddings build`) // makes setupVectorServing degrade to a fully-disabled vectorServing after a // short retry window, rather than blocking or failing daemon startup, and // that it logs a clear warning explaining why. func TestSetupVectorServingDisablesWhenWriteLockHeld(t *testing.T) { origInterval, origTimeout := vectorsWriteLockRetryInterval, vectorsWriteLockRetryTimeout vectorsWriteLockRetryInterval = time.Millisecond vectorsWriteLockRetryTimeout = 10 * time.Millisecond t.Cleanup(func() { vectorsWriteLockRetryInterval = origInterval vectorsWriteLockRetryTimeout = origTimeout }) dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) cfg := vectorTestConfig(dataDir) held, err := tryAcquireNamedLock(dataDir, vectorsWriteLockFile) require.NoError(t, err) defer func() { require.NoError(t, held.Close()) }() logBuf := captureLogOutput(t) vs, err := setupVectorServing(context.Background(), cfg, database) require.NoError(t, err, "a held lock must degrade, not fail, daemon startup") assert.Nil(t, vs.Scheduler) assert.Nil(t, vs.Close) assert.Contains(t, logBuf.String(), "vectors.write.lock") assert.Contains(t, logBuf.String(), "disabling vector serving") // The degraded run still carries one server option: the unavailability // reason, so the embeddings routes' 501s explain the lock conflict and // the restart remedy instead of the generic "manager not available" // (the reason-to-501 threading itself is pinned in internal/server's // TestEmbeddingsUnavailableReasonReplacesGeneric501). assert.Len(t, vs.ServerOpts, 1) } // TestSetupVectorServingAcquiresAndReleasesWriteLock asserts a free // vectors.write.lock is acquired for the daemon's lifetime and released by // Close, so a second setupVectorServing call after Close succeeds rather // than finding the lock still held. func TestSetupVectorServingAcquiresAndReleasesWriteLock(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) cfg := vectorTestConfig(dataDir) vs, err := setupVectorServing(context.Background(), cfg, database) require.NoError(t, err) require.NotNil(t, vs.Close) // While vs holds the lock, a competing direct acquire must fail. _, lockErr := tryAcquireNamedLock(dataDir, vectorsWriteLockFile) require.Error(t, lockErr, "setupVectorServing must hold vectors.write.lock while running") require.NoError(t, vs.Close()) // Once released, a fresh acquire — standing in for a second // setupVectorServing call — must succeed. held, err := tryAcquireNamedLock(dataDir, vectorsWriteLockFile) require.NoError(t, err, "Close must release vectors.write.lock") require.NoError(t, held.Close()) } func TestServeConstructionKeepsEmbeddingsRoutesUnavailableWhenVectorDisabled(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) ln, port := listenLoopback(t) cfg := config.Config{DataDir: dataDir, DBPath: filepath.Join(dataDir, "sessions.db"), Host: "127.0.0.1", Port: port} vs, err := setupVectorServing(context.Background(), cfg, database) require.NoError(t, err) assert.Nil(t, vs.Scheduler) assert.Nil(t, vs.Close) assert.Empty(t, vs.ServerOpts) srv := server.New(cfg, database, nil, vs.ServerOpts...) ts := startTestServer(t, ln, srv.Handler()) resp, err := http.Get(ts.URL + "/api/v1/embeddings/status") require.NoError(t, err) defer resp.Body.Close() assert.Equal(t, http.StatusNotImplemented, resp.StatusCode, "the embeddings API should be documented and registered, but unavailable when vector is disabled") assert.Equal(t, "application/json", mediaType(t, resp.Header.Get("Content-Type"))) } // mediaType extracts the bare MIME type from a Content-Type header value, // dropping any "; charset=..." parameters, so tests can compare it exactly. func mediaType(t *testing.T, contentType string) string { t.Helper() mt, _, err := mime.ParseMediaType(contentType) require.NoError(t, err) return mt } // --- installDirectVectorSearcher (direct, non-daemon CLI path) --- func TestInstallDirectVectorSearcherNoOpWhenVectorDisabled(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) cfg := config.Config{DataDir: dataDir} closeFn := installDirectVectorSearcher(cfg, database) assert.Nil(t, closeFn) assert.False(t, database.HasSemantic()) } func TestInstallDirectVectorSearcherNoOpWhenVectorsDBMissing(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) cfg := vectorTestConfig(dataDir) closeFn := installDirectVectorSearcher(cfg, database) assert.Nil(t, closeFn) assert.False(t, database.HasSemantic(), "no searcher wired means callers see db.ErrSemanticUnavailable naturally") } func TestInstallDirectVectorSearcherWiresSearcherWhenVectorsDBExists(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) cfg := vectorTestConfig(dataDir) // Create vectors.db up front, as a prior `embeddings build` would. seed, err := vector.Open(context.Background(), cfg.Vector.ResolvedDBPath(dataDir), false, 1000) require.NoError(t, err) require.NoError(t, seed.Close()) closeFn := installDirectVectorSearcher(cfg, database) require.NotNil(t, closeFn) defer func() { assert.NoError(t, closeFn()) }() assert.True(t, database.HasSemantic(), "an existing vectors.db must wire a read-only searcher for direct CLI reads") } // TestInstallDirectVectorSearcherDegradesOnCorruptVectorsDB is a regression // test for a bug where a corrupt or incompatible vectors.db broke direct // service construction entirely, taking down unrelated commands like // `session list` that never touch the vector index. A garbage vectors.db // file must degrade to "no searcher installed" rather than propagate an // error, leaving non-semantic reads unaffected and semantic search falling // back to the standard unavailable error. func TestInstallDirectVectorSearcherDegradesOnCorruptVectorsDB(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) cfg := vectorTestConfig(dataDir) // Not a SQLite file at all — simulates corruption or an incompatible // build rather than a partially-written one. vectorsPath := cfg.Vector.ResolvedDBPath(dataDir) require.NoError(t, os.MkdirAll(filepath.Dir(vectorsPath), 0o755)) require.NoError(t, os.WriteFile(vectorsPath, []byte("not a sqlite database"), 0o644)) closeFn := installDirectVectorSearcher(cfg, database) assert.Nil(t, closeFn, "a corrupt vectors.db must not return a handle to close") assert.False(t, database.HasSemantic(), "a corrupt vectors.db must degrade to no searcher, not fail construction") _, err := database.SearchContent(context.Background(), db.ContentSearchFilter{ Pattern: "query", Mode: "semantic", Limit: 5, }) assert.ErrorIs(t, err, db.ErrSemanticUnavailable, "semantic search must return the standard unavailable error once degraded") } // TestInstallDirectVectorSearcherVersionMismatchServesRebuildRequired pins // the direct-install path's handling of a version-mismatched vectors.db: // the read-only open succeeds, so the searcher must be WIRED (not silently // dropped by the corrupt-file degradation branch) and semantic search over // HTTP must answer with the 501 rebuild-required taxonomy whose body names // the incompatible-version remediation — not the generic "not enabled" // message and not an empty page. func TestInstallDirectVectorSearcherVersionMismatchServesRebuildRequired(t *testing.T) { dataDir := t.TempDir() database := dbtest.OpenTestDBAt(t, filepath.Join(dataDir, "sessions.db")) cfg := vectorTestConfig(dataDir) path := cfg.Vector.ResolvedDBPath(dataDir) // Create a current vectors.db, then restamp it as written by the // previous mirror schema version. seed, err := vector.Open(context.Background(), path, false, cfg.Vector.Embeddings.MaxInputChars) require.NoError(t, err) require.NoError(t, seed.Close()) raw, err := sql.Open("sqlite3", path) require.NoError(t, err) _, err = raw.Exec(`UPDATE vector_meta SET value = '2' WHERE key = 'mirror_schema_version'`) require.NoError(t, err) require.NoError(t, raw.Close()) closeFn := installDirectVectorSearcher(cfg, database) require.NotNil(t, closeFn, "a version-mismatched vectors.db must wire a searcher, not silently unwire semantic search") defer func() { assert.NoError(t, closeFn()) }() require.True(t, database.HasSemantic()) ln, port := listenLoopback(t) cfg.Host, cfg.Port = "127.0.0.1", port srv := server.New(cfg, database, nil) ts := startTestServer(t, ln, srv.Handler()) req, err := http.NewRequest(http.MethodGet, ts.URL+"/api/v1/search/content?pattern=anything&mode=semantic", nil) require.NoError(t, err) req.Header.Set("X-AgentsView-Search-Intent", "semantic") resp, err := http.DefaultClient.Do(req) require.NoError(t, err) defer resp.Body.Close() body, err := io.ReadAll(resp.Body) require.NoError(t, err) assert.Equal(t, http.StatusNotImplemented, resp.StatusCode, "body: %s", body) assert.Contains(t, string(body), "incompatible version", "the error body must carry the rebuild-required remediation") assert.Contains(t, string(body), "embeddings build", "the error body must tell the user how to rebuild") }