Add shared long-term memory server (experimental)

.gitignore

@@ -47,3 +47,4 @@ coverage*
 crd-helm-wrapper
 cmd/vmcp/__debug_bin*
 /vmcp
+.worktrees/

cmd/thv-memory/config.go

@@ -0,0 +1,112 @@
+// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+// Package main is the entry point for the ToolHive memory MCP server.
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"gopkg.in/yaml.v3"
+)
+
+const (
+	providerOllama = "ollama"
+)
+
+// Config is the memory server configuration, loaded from memory-server.yaml.
+type Config struct {
+	Storage  StorageConfig  `yaml:"storage"`
+	Vector   VectorConfig   `yaml:"vector"`
+	Embedder EmbedderConfig `yaml:"embedder"`
+	Server   ServerConfig   `yaml:"server"`
+}
+
+// StorageConfig configures the Store backend.
+type StorageConfig struct {
+	Provider string `yaml:"provider"` // sqlite (default)
+	DSN      string `yaml:"dsn"`
+}
+
+// VectorConfig configures the VectorStore backend.
+type VectorConfig struct {
+	Provider string `yaml:"provider"` // sqlite-vec (default) | qdrant | pgvector
+	URL      string `yaml:"url"`
+}
+
+// EmbedderConfig configures the Embedder backend.
+type EmbedderConfig struct {
+	Provider string `yaml:"provider"` // ollama (default) | openai
+	URL      string `yaml:"url"`
+	Model    string `yaml:"model"`
+}
+
+// ServerConfig configures the MCP server itself.
+type ServerConfig struct {
+	Name           string `yaml:"name"`
+	Version        string `yaml:"version"`
+	Host           string `yaml:"host"`                     // default 0.0.0.0
+	Port           int    `yaml:"port"`                     // default 8080
+	LifecycleHours int    `yaml:"lifecycle_interval_hours"` // default 24
+}
+
+// LoadConfig reads and validates config from path. The path is operator-supplied
+// and expected to be a trusted config file location.
+func LoadConfig(path string) (*Config, error) {
+	// G304: path is an operator-supplied config file, not user input.
+	data, err := os.ReadFile(path) //nolint:gosec
+	if err != nil {
+		return nil, fmt.Errorf("reading config: %w", err)
+	}
+	var cfg Config
+	if err := yaml.Unmarshal(data, &cfg); err != nil {
+		return nil, fmt.Errorf("parsing config: %w", err)
+	}
+	applyStorageDefaults(&cfg)
+	applyEmbedderDefaults(&cfg)
+	applyServerDefaults(&cfg)
+	return &cfg, nil
+}
+
+func applyStorageDefaults(cfg *Config) {
+	if cfg.Storage.Provider == "" {
+		cfg.Storage.Provider = "sqlite"
+	}
+	if cfg.Storage.DSN == "" && cfg.Storage.Provider == "sqlite" {
+		cfg.Storage.DSN = "/data/memory.db"
+	}
+	if cfg.Vector.Provider == "" {
+		cfg.Vector.Provider = "sqlite-vec"
+	}
+}
+
+func applyEmbedderDefaults(cfg *Config) {
+	if cfg.Embedder.Provider == "" {
+		cfg.Embedder.Provider = providerOllama
+	}
+	if cfg.Embedder.Model == "" {
+		cfg.Embedder.Model = "nomic-embed-text"
+	}
+	if cfg.Embedder.URL == "" && cfg.Embedder.Provider == providerOllama {
+		cfg.Embedder.URL = "http://localhost:11434"
+	}
+}
+
+func applyServerDefaults(cfg *Config) {
+	if cfg.Server.Name == "" {
+		cfg.Server.Name = "toolhive-memory"
+	}
+	if cfg.Server.Version == "" {
+		cfg.Server.Version = "0.1.0"
+	}
+	if cfg.Server.Host == "" {
+		cfg.Server.Host = "0.0.0.0"
+	}
+	if cfg.Server.Port <= 0 {
+		cfg.Server.Port = 8080
+	}
+	if cfg.Server.LifecycleHours <= 0 {
+		cfg.Server.LifecycleHours = 24
+	}
+}

cmd/thv-memory/integration_test.go

@@ -0,0 +1,114 @@
+// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package main_test
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
+
+	"github.com/stacklok/toolhive/pkg/memory"
+	memorysqlite "github.com/stacklok/toolhive/pkg/memory/sqlite"
+)
+
+// fakeEmbedder returns a deterministic embedding for testing without a real model server.
+type fakeEmbedder struct{}
+
+func (*fakeEmbedder) Embed(_ context.Context, text string) ([]float32, error) {
+	v := []float32{0, 0, 0}
+	for i, c := range text {
+		if i >= 3 {
+			break
+		}
+		v[i] = float32(c) / 128.0
+	}
+	return v, nil
+}
+
+func (*fakeEmbedder) Dimensions() int { return 3 }
+
+func TestIntegration_RememberSearchForget(t *testing.T) {
+	t.Parallel()
+	dir := t.TempDir()
+	resolved, _ := filepath.EvalSymlinks(dir)
+	db, err := memorysqlite.Open(context.Background(), filepath.Join(resolved, "test.db"))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+
+	store := memorysqlite.NewStore(db)
+	vectors := memorysqlite.NewVectorStore(db)
+	svc, err := memory.NewService(store, vectors, &fakeEmbedder{}, zaptest.NewLogger(t))
+	require.NoError(t, err)
+
+	ctx := context.Background()
+
+	r, err := svc.Remember(ctx, memory.RememberInput{
+		Content: "deploy to us-east-1",
+		Type:    memory.TypeSemantic,
+		Author:  memory.AuthorHuman,
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, r.MemoryID)
+	require.Empty(t, r.Conflicts)
+
+	results, err := svc.Search(ctx, "deploy to us-east-1", nil, 5)
+	require.NoError(t, err)
+	require.NotEmpty(t, results)
+	require.Equal(t, "deploy to us-east-1", results[0].Entry.Content)
+
+	entry, err := store.Get(ctx, r.MemoryID)
+	require.NoError(t, err)
+	require.Equal(t, 1, entry.AccessCount)
+
+	require.NoError(t, store.Delete(ctx, r.MemoryID))
+	_, err = store.Get(ctx, r.MemoryID)
+	require.ErrorIs(t, err, memory.ErrNotFound)
+}
+
+func TestIntegration_ConflictDetection(t *testing.T) {
+	t.Parallel()
+	dir := t.TempDir()
+	resolved, _ := filepath.EvalSymlinks(dir)
+	db, err := memorysqlite.Open(context.Background(), filepath.Join(resolved, "test2.db"))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+
+	store := memorysqlite.NewStore(db)
+	vectors := memorysqlite.NewVectorStore(db)
+	svc, err := memory.NewService(store, vectors, &fakeEmbedder{}, zaptest.NewLogger(t))
+	require.NoError(t, err)
+
+	ctx := context.Background()
+
+	r1, err := svc.Remember(ctx, memory.RememberInput{
+		Content: "auth port 8080",
+		Type:    memory.TypeSemantic,
+		Author:  memory.AuthorHuman,
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, r1.MemoryID)
+
+	// fakeEmbedder hashes first 3 chars — "aut" maps to same vector for both,
+	// so "auth port 9090" will have cosine similarity 1.0 with "auth port 8080".
+	r2, err := svc.Remember(ctx, memory.RememberInput{
+		Content: "auth port 9090",
+		Type:    memory.TypeSemantic,
+		Author:  memory.AuthorAgent,
+	})
+	require.NoError(t, err)
+	require.Empty(t, r2.MemoryID)
+	require.NotEmpty(t, r2.Conflicts)
+
+	r3, err := svc.Remember(ctx, memory.RememberInput{
+		Content: "auth port 9090",
+		Type:    memory.TypeSemantic,
+		Author:  memory.AuthorHuman,
+		Force:   true,
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, r3.MemoryID)
+}

cmd/thv-memory/lifecycle/job.go

@@ -0,0 +1,84 @@
+// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+// Package lifecycle provides the background maintenance job for memory entries.
+package lifecycle
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/stacklok/toolhive/pkg/memory"
+)
+
+// StalenessAuditThreshold is the score above which entries are logged as audit candidates.
+const StalenessAuditThreshold = float32(0.8)
+
+// Job runs periodic maintenance on the memory store: expiring TTL'd entries
+// and recomputing trust/staleness scores.
+type Job struct {
+	store memory.Store
+	log   *zap.Logger
+}
+
+// New creates a new lifecycle Job.
+func New(store memory.Store, log *zap.Logger) *Job {
+	return &Job{store: store, log: log}
+}
+
+// Run starts the background job, ticking at the given interval until ctx is cancelled.
+func (j *Job) Run(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := j.RunOnce(ctx); err != nil {
+				j.log.Warn("lifecycle job error", zap.Error(err))
+			}
+		}
+	}
+}
+
+// RunOnce executes one maintenance pass: expire TTL'd entries, update scores.
+func (j *Job) RunOnce(ctx context.Context) error {
+	if err := j.expireEntries(ctx); err != nil {
+		return err
+	}
+	return j.recomputeScores(ctx)
+}
+
+func (j *Job) expireEntries(ctx context.Context) error {
+	expired, err := j.store.ListExpired(ctx)
+	if err != nil {
+		return err
+	}
+	for _, e := range expired {
+		if err := j.store.Archive(ctx, e.ID, memory.ArchiveReasonExpired, ""); err != nil {
+			j.log.Warn("failed to archive expired entry", zap.String("id", e.ID), zap.Error(err))
+		}
+	}
+	return nil
+}
+
+func (j *Job) recomputeScores(ctx context.Context) error {
+	entries, err := j.store.ListActive(ctx)
+	if err != nil {
+		return err
+	}
+	for _, e := range entries {
+		trust := memory.ComputeTrustScore(e)
+		staleness := memory.ComputeStalenessScore(e)
+		if err := j.store.UpdateScores(ctx, e.ID, trust, staleness); err != nil {
+			j.log.Warn("failed to update scores", zap.String("id", e.ID), zap.Error(err))
+		}
+		if staleness >= StalenessAuditThreshold {
+			j.log.Debug("high staleness entry", zap.String("id", e.ID), zap.Float32("staleness", staleness))
+		}
+	}
+	return nil
+}

cmd/thv-memory/lifecycle/job_test.go

@@ -0,0 +1,55 @@
+// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package lifecycle_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+	"go.uber.org/zap/zaptest"
+
+	"github.com/stacklok/toolhive/cmd/thv-memory/lifecycle"
+	"github.com/stacklok/toolhive/pkg/memory"
+	"github.com/stacklok/toolhive/pkg/memory/mocks"
+)
+
+func TestJob_RunOnce_ExpiresEntries(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	store := mocks.NewMockStore(ctrl)
+
+	expired := memory.Entry{
+		ID:        "mem_expired",
+		CreatedAt: time.Now().Add(-48 * time.Hour),
+	}
+	store.EXPECT().ListExpired(gomock.Any()).Return([]memory.Entry{expired}, nil)
+	store.EXPECT().Archive(gomock.Any(), "mem_expired", memory.ArchiveReasonExpired, "").Return(nil)
+	store.EXPECT().ListActive(gomock.Any()).Return(nil, nil)
+
+	job := lifecycle.New(store, zaptest.NewLogger(t))
+	err := job.RunOnce(context.Background())
+	require.NoError(t, err)
+}
+
+func TestJob_RunOnce_UpdatesScores(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	store := mocks.NewMockStore(ctrl)
+
+	entry := memory.Entry{
+		ID:        "mem_active",
+		Author:    memory.AuthorHuman,
+		CreatedAt: time.Now(),
+	}
+	store.EXPECT().ListExpired(gomock.Any()).Return(nil, nil)
+	store.EXPECT().ListActive(gomock.Any()).Return([]memory.Entry{entry}, nil)
+	store.EXPECT().UpdateScores(gomock.Any(), "mem_active", gomock.Any(), gomock.Any()).Return(nil)
+
+	job := lifecycle.New(store, zaptest.NewLogger(t))
+	err := job.RunOnce(context.Background())
+	require.NoError(t, err)
+}