fix: recover missing embeddings safely

This commit is contained in:
ntduc committed 2026-07-22 19:01:16 +07:00
1 parent 51a27af707
commit 876104950b
9 files changed
+914 -52

No files matched your search

+4 -2
View File
@@ -51,8 +51,10 @@ func resolveEmbeddingProvider(
}
}
// 2. Auto-detect: scan DB providers for first with settings.embedding.enabled
allProviders, err := providerStore.ListAllProviders(context.Background())
// 2. Auto-detect only from the master tenant. The selected provider receives
// cross-tenant semantic content during startup maintenance, so a tenant-owned
// endpoint must never be selected for this process-wide responsibility.
allProviders, err := providerStore.ListProviders(masterCtx)
if err != nil {
slog.Warn("failed to list providers for embedding auto-detect", "error", err)
return nil
+68 -1
View File
@@ -98,7 +98,12 @@ func captureEmbeddingRequest(t *testing.T, es *store.EmbeddingSettings) map[stri
t.Fatalf("Decode() error = %v", err)
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"data":[{"embedding":[0.1,0.2]}]}`))
_ = json.NewEncoder(w).Encode(map[string]any{
"data": []map[string]any{{
"embedding": make([]float32, store.RequiredMemoryEmbeddingDimensions),
"index": 0,
}},
})
}))
defer server.Close()
@@ -137,3 +142,65 @@ func TestBuildEmbeddingProviderIgnoresIncompatibleStoredDimensions(t *testing.T)
t.Fatalf("dimensions = %v, want fallback 1536", got)
}
}
type embeddingProviderStoreStub struct {
masterProviders []store.LLMProviderData
allProviders []store.LLMProviderData
listTenant uuid.UUID
listAllCalled bool
}
func (s *embeddingProviderStoreStub) CreateProvider(context.Context, *store.LLMProviderData) error {
return nil
}
func (s *embeddingProviderStoreStub) GetProvider(context.Context, uuid.UUID) (*store.LLMProviderData, error) {
return nil, nil
}
func (s *embeddingProviderStoreStub) GetProviderByName(context.Context, string) (*store.LLMProviderData, error) {
return nil, nil
}
func (s *embeddingProviderStoreStub) ListProviders(ctx context.Context) ([]store.LLMProviderData, error) {
s.listTenant = store.TenantIDFromContext(ctx)
return s.masterProviders, nil
}
func (s *embeddingProviderStoreStub) ListAllProviders(context.Context) ([]store.LLMProviderData, error) {
s.listAllCalled = true
return s.allProviders, nil
}
func (s *embeddingProviderStoreStub) UpdateProvider(context.Context, uuid.UUID, map[string]any) error {
return nil
}
func (s *embeddingProviderStoreStub) DeleteProvider(context.Context, uuid.UUID) error { return nil }
func TestResolveEmbeddingProviderAutoDetectUsesMasterTenantOnly(t *testing.T) {
embeddingSettings := json.RawMessage(`{"embedding":{"enabled":true}}`)
providerStore := &embeddingProviderStoreStub{
masterProviders: []store.LLMProviderData{{
TenantID: store.MasterTenantID,
Name: "master-embedding",
ProviderType: store.ProviderOpenAICompat,
APIBase: "http://master.invalid/v1",
APIKey: "master-key",
Enabled: true,
Settings: embeddingSettings,
}},
allProviders: []store.LLMProviderData{{
TenantID: uuid.New(),
Name: "tenant-controlled-endpoint",
ProviderType: store.ProviderOpenAICompat,
Enabled: true,
Settings: embeddingSettings,
}},
}
provider := resolveEmbeddingProvider(providerStore, nil, nil)
if provider == nil || provider.Name() != "master-embedding" {
t.Fatalf("resolveEmbeddingProvider() = %v, want master-embedding", provider)
}
if providerStore.listTenant != store.MasterTenantID {
t.Fatalf("ListProviders tenant = %s, want master tenant", providerStore.listTenant)
}
if providerStore.listAllCalled {
t.Fatal("ListAllProviders was called; cross-tenant auto-detect must stay disabled")
}
}
+45
View File
@@ -420,16 +420,61 @@ func setupMemoryEmbeddings(
}
// Wire embedding provider into vault store for semantic document search.
var vaultStore *pg.PGVaultStore
if pgStores.Vault != nil {
pgStores.Vault.SetEmbeddingProvider(embProvider)
slog.Info("vault embeddings enabled", "provider", embProvider.Name())
vaultStore, _ = pgStores.Vault.(*pg.PGVaultStore)
}
// V3: Wire embedding provider into episodic store for semantic search.
var episodicStore *pg.PGEpisodicStore
if pgStores.Episodic != nil {
pgStores.Episodic.SetEmbeddingProvider(embProvider)
slog.Info("episodic embeddings enabled", "provider", embProvider.Name())
episodicStore, _ = pgStores.Episodic.(*pg.PGEpisodicStore)
}
// Agent create/update embedding hooks require the provider to be wired.
var agentStore *pg.PGAgentStore
if pgAgentStore, ok := pgStores.Agents.(*pg.PGAgentStore); ok {
agentStore = pgAgentStore
agentStore.SetEmbeddingProvider(embProvider)
slog.Info("agent embeddings enabled", "provider", embProvider.Name())
}
// Recover the remaining semantic indexes sequentially to avoid a burst
// of concurrent batch requests during gateway startup. Each surface gets
// its own deadline so a large agent backlog cannot starve later stores.
go func() {
if agentStore != nil {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
if count, err := agentStore.BackfillAgentEmbeddings(bgCtx); err != nil {
slog.Warn("agent embeddings backfill failed", "error", err)
} else if count > 0 {
slog.Info("agent embeddings recovery complete", "agents_updated", count)
}
cancel()
}
if episodicStore != nil {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
if count, err := episodicStore.BackfillEpisodicEmbeddings(bgCtx); err != nil {
slog.Warn("episodic embeddings backfill failed", "error", err)
} else if count > 0 {
slog.Info("episodic embeddings backfill complete", "summaries_updated", count)
}
cancel()
}
if vaultStore != nil {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
if count, err := vaultStore.BackfillVaultEmbeddings(bgCtx); err != nil {
slog.Warn("vault embeddings backfill failed", "error", err)
} else if count > 0 {
slog.Info("vault embeddings backfill complete", "documents_updated", count)
}
cancel()
}
}()
} else {
slog.Warn("memory embeddings disabled (no API key), chunks stored without vectors")
}