diff --git a/internal/store/knowledge_graph_store.go b/internal/store/knowledge_graph_store.go index 4b27a794..0acb52af 100644 --- a/internal/store/knowledge_graph_store.go +++ b/internal/store/knowledge_graph_store.go @@ -96,7 +96,7 @@ type KnowledgeGraphStore interface { // Auto-merges at high similarity (>0.98 + name match), flags medium (>0.90) as candidates. DedupAfterExtraction(ctx context.Context, agentID, userID string, newEntityIDs []string) (merged int, flagged int, err error) // ScanDuplicates scans ALL entities with embeddings for duplicates (self-join). - // Flags candidates above threshold. Used for on-demand bulk scanning of existing data. + // Returns the number of newly inserted review candidates above threshold. ScanDuplicates(ctx context.Context, agentID, userID string, threshold float64, limit int) (int, error) // ListDedupCandidates returns pending dedup candidates for review. ListDedupCandidates(ctx context.Context, agentID, userID string, limit int) ([]DedupCandidate, error) diff --git a/internal/store/pg/knowledge_graph_dedup.go b/internal/store/pg/knowledge_graph_dedup.go index 35491bd5..02fe65d9 100644 --- a/internal/store/pg/knowledge_graph_dedup.go +++ b/internal/store/pg/knowledge_graph_dedup.go @@ -84,9 +84,10 @@ func (s *PGKnowledgeGraphStore) DedupAfterExtraction(ctx context.Context, agentI break // entity merged, stop checking neighbors } else if n.similarity >= dedupCandidateThreshold { // Flag as candidate for manual review - if err := s.insertDedupCandidate(ctx, aid, userID, eid, n.id, n.similarity); err != nil { + inserted, err := s.insertDedupCandidate(ctx, aid, userID, eid, n.id, n.similarity) + if err != nil { slog.Warn("kg.dedup: flag candidate failed", "error", err) - } else { + } else if inserted { flagged++ } } @@ -143,32 +144,39 @@ func (s *PGKnowledgeGraphStore) knnNeighbors(ctx context.Context, agentID uuid.U return results, nil } -func (s *PGKnowledgeGraphStore) insertDedupCandidate(ctx context.Context, agentID uuid.UUID, userID, entityAID, entityBID string, similarity float64) error { +func (s *PGKnowledgeGraphStore) insertDedupCandidate(ctx context.Context, agentID uuid.UUID, userID, entityAID, entityBID string, similarity float64) (bool, error) { // Ensure consistent ordering (smaller UUID first) to avoid duplicates if entityAID > entityBID { entityAID, entityBID = entityBID, entityAID } aID, err := parseUUID(entityAID) if err != nil { - return fmt.Errorf("insert dedup candidate: entity_a_id: %w", err) + return false, fmt.Errorf("insert dedup candidate: entity_a_id: %w", err) } bID, err := parseUUID(entityBID) if err != nil { - return fmt.Errorf("insert dedup candidate: entity_b_id: %w", err) + return false, fmt.Errorf("insert dedup candidate: entity_b_id: %w", err) } tid := tenantIDForInsert(ctx) - _, err = s.db.ExecContext(ctx, ` + res, err := s.db.ExecContext(ctx, ` INSERT INTO kg_dedup_candidates (id, tenant_id, agent_id, user_id, entity_a_id, entity_b_id, similarity, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (entity_a_id, entity_b_id) DO NOTHING`, uuid.Must(uuid.NewV7()), tid, agentID, userID, aID, bID, similarity, time.Now(), ) - return err + if err != nil { + return false, err + } + n, err := res.RowsAffected() + if err != nil { + return false, err + } + return n > 0, nil } // ScanDuplicates performs a bulk scan of ALL entities with embeddings using // a self-join to find duplicate candidates above the given threshold. -// Inserts results into kg_dedup_candidates. Returns number of candidates found. +// Inserts results into kg_dedup_candidates. Returns newly inserted candidates. func (s *PGKnowledgeGraphStore) ScanDuplicates(ctx context.Context, agentID, userID string, threshold float64, limit int) (int, error) { aid, err := parseUUID(agentID) if err != nil { @@ -245,16 +253,19 @@ func (s *PGKnowledgeGraphStore) ScanDuplicates(ctx context.Context, agentID, use slog.Warn("kg.scan_duplicates: invalid entity_b UUID from DB row", "id", bID, "error", err) continue } - if _, err := s.db.ExecContext(ctx, ` + res, err := s.db.ExecContext(ctx, ` INSERT INTO kg_dedup_candidates (id, tenant_id, agent_id, user_id, entity_a_id, entity_b_id, similarity, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (entity_a_id, entity_b_id) DO NOTHING`, uuid.Must(uuid.NewV7()), tid, aid, userID, aUUID, bUUID, sim, time.Now(), - ); err != nil { + ) + if err != nil { slog.Warn("kg.scan_duplicates: insert candidate failed", "error", err) continue } - found++ + if n, err := res.RowsAffected(); err == nil && n > 0 { + found++ + } } return found, rows.Err() diff --git a/internal/store/sqlitestore/kg_dedup.go b/internal/store/sqlitestore/kg_dedup.go index ff274b3b..151d62ad 100644 --- a/internal/store/sqlitestore/kg_dedup.go +++ b/internal/store/sqlitestore/kg_dedup.go @@ -196,9 +196,10 @@ func (s *SQLiteKnowledgeGraphStore) DedupAfterExtraction(ctx context.Context, ag break } } else if sim >= kgDedupFlagThreshold { - if flagErr := s.insertDedupCandidate(ctx, agentID, userID, newE.ID, existE.ID, sim); flagErr != nil { + inserted, flagErr := s.insertDedupCandidate(ctx, agentID, userID, newE.ID, existE.ID, sim) + if flagErr != nil { slog.Warn("kg.dedup: flag candidate failed", "error", flagErr) - } else { + } else if inserted { flagged++ } } @@ -210,7 +211,7 @@ func (s *SQLiteKnowledgeGraphStore) DedupAfterExtraction(ctx context.Context, ag // ScanDuplicates performs a bulk scan of all entities using Go-side pairwise // Jaro-Winkler. 2-pass: load distinct entity_types, then compare within each type. -// Caps per-type pool at kgDedupEntityCap. Returns number of candidates inserted. +// Caps per-type pool at kgDedupEntityCap. Returns newly inserted candidates. func (s *SQLiteKnowledgeGraphStore) ScanDuplicates(ctx context.Context, agentID, userID string, threshold float64, limit int) (int, error) { if threshold <= 0 { threshold = kgDedupFlagThreshold @@ -277,11 +278,14 @@ func (s *SQLiteKnowledgeGraphStore) ScanDuplicates(ctx context.Context, agentID, a, b := &pool[i], &pool[j] sim := kg.JaroWinkler(a.Name+" "+a.Description, b.Name+" "+b.Description) if sim >= threshold { - if insertErr := s.insertDedupCandidate(ctx, agentID, userID, a.ID, b.ID, sim); insertErr != nil { + inserted, insertErr := s.insertDedupCandidate(ctx, agentID, userID, a.ID, b.ID, sim) + if insertErr != nil { slog.Warn("kg.scan_duplicates: insert candidate failed", "error", insertErr) continue } - found++ + if inserted { + found++ + } } } } @@ -486,21 +490,28 @@ func (s *SQLiteKnowledgeGraphStore) DismissCandidate(ctx context.Context, agentI } // insertDedupCandidate inserts a dedup candidate pair (smaller ID first for dedup consistency). -func (s *SQLiteKnowledgeGraphStore) insertDedupCandidate(ctx context.Context, agentID, userID, entityAID, entityBID string, similarity float64) error { +func (s *SQLiteKnowledgeGraphStore) insertDedupCandidate(ctx context.Context, agentID, userID, entityAID, entityBID string, similarity float64) (bool, error) { if entityAID > entityBID { entityAID, entityBID = entityBID, entityAID } id := uuid.Must(uuid.NewV7()).String() tid := tenantIDForInsert(ctx).String() now := time.Now().UTC().Format(time.RFC3339Nano) - _, err := s.db.ExecContext(ctx, ` + res, err := s.db.ExecContext(ctx, ` INSERT INTO kg_dedup_candidates (id, tenant_id, agent_id, user_id, entity_a_id, entity_b_id, similarity, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(entity_a_id, entity_b_id) DO NOTHING`, id, tid, agentID, userID, entityAID, entityBID, similarity, now, ) - return err + if err != nil { + return false, err + } + n, err := res.RowsAffected() + if err != nil { + return false, err + } + return n > 0, nil } // fetchEntitiesByIDs loads a set of entities by their DB IDs within an agent scope. diff --git a/internal/store/sqlitestore/kg_dedup_test.go b/internal/store/sqlitestore/kg_dedup_test.go new file mode 100644 index 00000000..d6f1b1f2 --- /dev/null +++ b/internal/store/sqlitestore/kg_dedup_test.go @@ -0,0 +1,75 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "testing" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestSQLiteKGScanDuplicatesCountsOnlyInsertedCandidates(t *testing.T) { + db := newHookTestDB(t) + tenantID, agentID := seedHookTenantAgent(t, db) + ctx := store.WithSharedKG(sqliteTenantCtx(tenantID)) + kg := NewSQLiteKnowledgeGraphStore(db) + + entities := []*store.Entity{ + { + AgentID: agentID.String(), + UserID: "u1", + ExternalID: "alice-a", + Name: "Alice Nguyen", + EntityType: "person", + Description: "Project manager for the migration", + Confidence: 0.9, + }, + { + AgentID: agentID.String(), + UserID: "u2", + ExternalID: "alice-b", + Name: "Alice Nguyen", + EntityType: "person", + Description: "Project manager for the migration", + Confidence: 0.9, + }, + } + for _, entity := range entities { + if err := kg.UpsertEntity(ctx, entity); err != nil { + t.Fatalf("UpsertEntity: %v", err) + } + } + + aID, bID := entities[0].ID, entities[1].ID + if aID > bID { + aID, bID = bID, aID + } + if _, err := db.ExecContext(ctx, ` + INSERT INTO kg_dedup_candidates + (id, tenant_id, agent_id, user_id, entity_a_id, entity_b_id, similarity, status) + VALUES (?, ?, ?, '', ?, ?, 1.0, 'dismissed')`, + uuid.Must(uuid.NewV7()).String(), tenantID.String(), agentID.String(), aID, bID, + ); err != nil { + t.Fatalf("seed dismissed candidate: %v", err) + } + + found, err := kg.ScanDuplicates(ctx, agentID.String(), "", 0.90, 100) + if err != nil { + t.Fatalf("ScanDuplicates: %v", err) + } + if found != 0 { + t.Fatalf("ScanDuplicates found = %d, want 0 for conflict with dismissed candidate", found) + } + + if _, err := db.ExecContext(ctx, `DELETE FROM kg_dedup_candidates WHERE entity_a_id = ? AND entity_b_id = ?`, aID, bID); err != nil { + t.Fatalf("delete seeded candidate: %v", err) + } + found, err = kg.ScanDuplicates(ctx, agentID.String(), "", 0.90, 100) + if err != nil { + t.Fatalf("ScanDuplicates after delete: %v", err) + } + if found != 1 { + t.Fatalf("ScanDuplicates found = %d, want 1 newly inserted candidate", found) + } +} diff --git a/tests/integration/v3_knowledge_graph_store_test.go b/tests/integration/v3_knowledge_graph_store_test.go index d1234e5c..0aa92781 100644 --- a/tests/integration/v3_knowledge_graph_store_test.go +++ b/tests/integration/v3_knowledge_graph_store_test.go @@ -329,6 +329,70 @@ func TestStoreKG_Stats(t *testing.T) { } } +func TestStoreKG_ScanDuplicatesCountsOnlyInsertedCandidates(t *testing.T) { + db := testDB(t) + tenantID, agentID := seedTenantAgent(t, db) + ctx := store.WithSharedKG(tenantCtx(tenantID)) + s := newKGStore(t) + + aid := agentID.String() + entities := []store.Entity{ + { + ExternalID: "dupe-a", + Name: "Alice Nguyen", + EntityType: "person", + Description: "Project manager for the migration", + Confidence: 0.9, + }, + { + ExternalID: "dupe-b", + Name: "Alice Nguyen", + EntityType: "person", + Description: "Project manager for the migration", + Confidence: 0.9, + }, + } + entityIDs, err := s.IngestExtraction(ctx, aid, "", entities, nil) + if err != nil { + t.Fatalf("IngestExtraction: %v", err) + } + if len(entityIDs) != 2 { + t.Fatalf("entityIDs len = %d, want 2", len(entityIDs)) + } + + aID, bID := entityIDs[0], entityIDs[1] + if aID > bID { + aID, bID = bID, aID + } + if _, err := db.ExecContext(ctx, ` + INSERT INTO kg_dedup_candidates + (id, tenant_id, agent_id, user_id, entity_a_id, entity_b_id, similarity, status) + VALUES ($1, $2, $3, '', $4, $5, 1.0, 'dismissed')`, + uuid.Must(uuid.NewV7()), tenantID, agentID, aID, bID, + ); err != nil { + t.Fatalf("seed dismissed candidate: %v", err) + } + + found, err := s.ScanDuplicates(ctx, aid, "", 0.90, 100) + if err != nil { + t.Fatalf("ScanDuplicates: %v", err) + } + if found != 0 { + t.Fatalf("ScanDuplicates found = %d, want 0 for conflict with dismissed candidate", found) + } + + if _, err := db.ExecContext(ctx, `DELETE FROM kg_dedup_candidates WHERE entity_a_id = $1 AND entity_b_id = $2`, aID, bID); err != nil { + t.Fatalf("delete seeded candidate: %v", err) + } + found, err = s.ScanDuplicates(ctx, aid, "", 0.90, 100) + if err != nil { + t.Fatalf("ScanDuplicates after delete: %v", err) + } + if found != 1 { + t.Fatalf("ScanDuplicates found = %d, want 1 newly inserted candidate", found) + } +} + func TestStoreKG_UserScopeIsolation(t *testing.T) { db := testDB(t) tenantID, agentID := seedTenantAgent(t, db)