fix: drop orphan knowledge graph relations (#20293)

internal/ingestion/component/knowledge_compiler/common/relations.go

@@ -0,0 +1,58 @@
+package common
+
+import "strings"
+
+// FilterOrphanRelations removes relation products whose endpoints are not
+// present in the same compilation result. Entity and relation rows are scoped
+// by variant, template, and compile kind so unrelated templates cannot satisfy
+// each other's endpoints.
+func FilterOrphanRelations(products []Product) []Product {
+	entities := make(map[string]struct{}, len(products))
+	for _, product := range products {
+		if product.Meta == nil || strings.TrimSpace(metaString(product.Meta, "kind")) != "entity" {
+			continue
+		}
+		name := normalizeRelationEndpoint(metaString(product.Meta, "name"))
+		if name != "" {
+			entities[relationScope(product)+"\x00"+name] = struct{}{}
+		}
+	}
+
+	filtered := make([]Product, 0, len(products))
+	for _, product := range products {
+		from := normalizeRelationEndpoint(metaString(product.Meta, "from"))
+		to := normalizeRelationEndpoint(metaString(product.Meta, "to"))
+		isRelation := strings.TrimSpace(metaString(product.Meta, "kind")) == "relation" || (from != "" || to != "")
+		if isRelation && (from == "" || to == "" || !hasRelationEntity(entities, product, from) || !hasRelationEntity(entities, product, to)) {
+			continue
+		}
+		filtered = append(filtered, product)
+	}
+	return filtered
+}
+
+func hasRelationEntity(entities map[string]struct{}, product Product, name string) bool {
+	_, ok := entities[relationScope(product)+"\x00"+name]
+	return ok
+}
+
+func relationScope(product Product) string {
+	compileKind := metaString(product.Meta, "compile_kwd")
+	if compileKind == "" {
+		compileKind = string(product.Variant)
+	}
+	template := strings.TrimSpace(product.TemplateID)
+	if template == "" {
+		template = strings.TrimSpace(product.Kind)
+	}
+	return string(product.Variant) + "\x00" + template + "\x00" + compileKind
+}
+
+func normalizeRelationEndpoint(value string) string {
+	return strings.ToLower(strings.Join(strings.Fields(value), " "))
+}
+
+func metaString(meta map[string]any, key string) string {
+	value, _ := meta[key].(string)
+	return value
+}

internal/ingestion/component/knowledge_compiler/common/relations_test.go

@@ -0,0 +1,30 @@
+package common
+
+import "testing"
+
+func TestFilterOrphanRelations(t *testing.T) {
+	products := []Product{
+		{Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "entity", "name": "A"}},
+		{Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "entity", "name": "B"}},
+		{Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "relation", "from": " A ", "to": "B"}},
+		{Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "relation", "from": "A", "to": "Missing"}},
+	}
+
+	filtered := FilterOrphanRelations(products)
+	if len(filtered) != 3 {
+		t.Fatalf("got %d products, want 3", len(filtered))
+	}
+	if filtered[2].Meta["to"] != "B" {
+		t.Fatalf("valid relation was not preserved: %+v", filtered[2])
+	}
+}
+
+func TestFilterOrphanRelationsScopesTemplates(t *testing.T) {
+	products := []Product{
+		{Variant: VariantStructure, TemplateID: "graph", Meta: map[string]any{"kind": "entity", "name": "A"}},
+		{Variant: VariantStructure, TemplateID: "timeline", Meta: map[string]any{"kind": "relation", "from": "A", "to": "B"}},
+	}
+	if got := FilterOrphanRelations(products); len(got) != 1 {
+		t.Fatalf("relation satisfied by another template: got %d products, want 1", len(got))
+	}
+}

internal/ingestion/component/knowledge_compiler/structure/compile.go

@@ -71,6 +71,12 @@ func extractHypergraph(ctx context.Context, deps common.Deps, cfg CompileConfig,
 		known = append(known, v)
 	}
 
+	// Relations can only reference entities from this batch. If entity extraction
+	// returned nothing, skip the second LLM call and avoid creating orphan edges.
+	if len(known) == 0 {
+		return nodes, nil, nil
+	}
+
 	if strings.TrimSpace(edgePromptTmpl) == "" {
 		return nodes, nil, nil
 	}

internal/ingestion/component/knowledge_compiler/structure/structure_test.go

@@ -388,6 +388,20 @@ func TestFillKnownNodes(t *testing.T) {
 	}
 }
 
+func TestExtractHypergraphSkipsEdgesWithoutEntities(t *testing.T) {
+	chat := &graphChat{}
+	nodes, edges, err := extractHypergraph(context.Background(), common.Deps{Chat: chat}, CompileConfig{LLMID: "llm", ParserConfig: map[string]any{"entity": map[string]any{"id": "name"}}}, "node", "## Known Entities: {known_nodes}", "unrelated text")
+	if err != nil {
+		t.Fatalf("extractHypergraph: %v", err)
+	}
+	if len(nodes) != 0 || len(edges) != 0 {
+		t.Fatalf("empty entity batch returned nodes=%v edges=%v", nodes, edges)
+	}
+	if chat.nodeCalls != 1 || chat.edgeCalls != 0 {
+		t.Fatalf("LLM calls = node:%d edge:%d, want node:1 edge:0", chat.nodeCalls, chat.edgeCalls)
+	}
+}
+
 // ---- payload helpers ----
 
 func TestPayloadChunkIDs(t *testing.T) {

internal/ingestion/knowledge_compile/consumer.go

@@ -1563,6 +1563,7 @@ func structureEntityNavLine(payloadJSON string) string {
 // dataset_structure_merger._merge_bucket but is a self-contained from-scratch
 // path (B3: the legacy unified merge never aggregated structure by name).
 func (c *Consumer) mergeStructureDataset(ctx context.Context, tenant, kb string, products []kccommon.Product) error {
+	products = kccommon.FilterOrphanRelations(products)
 	common.Info("knowledge_compile: mergeStructureDataset entry",
 		zap.String("kb_id", kb),
 		zap.Int("products", len(products)))

internal/ingestion/knowledge_compile/structure_merge_test.go

@@ -119,6 +119,10 @@ func TestMergeStructureDataset_IsolatesTemplates(t *testing.T) {
 func TestMergeStructureDataset_RelationsNotDropped(t *testing.T) {
 	c := &Consumer{writer: &fakeWriter{}}
 	products := []kccommon.Product{
+		{Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity A",
+			Meta: map[string]any{"kind": "entity", "name": "A"}},
+		{Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity B",
+			Meta: map[string]any{"kind": "entity", "name": "B"}},
 		{Variant: kccommon.VariantStructure, DocID: "d1", Content: "rel desc",
 			Meta: map[string]any{"kind": "relation", "from": "A", "to": "B", "source_chunk_ids": []string{"c1"}}},
 		{Variant: kccommon.VariantStructure, DocID: "d2", Content: "rel desc 2",
@@ -128,10 +132,19 @@ func TestMergeStructureDataset_RelationsNotDropped(t *testing.T) {
 		t.Fatalf("mergeStructureDataset: %v", err)
 	}
 	fw := c.writer.(*fakeWriter)
-	if len(fw.buckets) != 1 {
-		t.Fatalf("want 1 relation bucket, got %d", len(fw.buckets))
+	if len(fw.buckets) != 3 {
+		t.Fatalf("want 2 entity buckets and 1 relation bucket, got %d", len(fw.buckets))
+	}
+	var b *StructureBucket
+	for i := range fw.buckets {
+		if fw.buckets[i].Type == "relation" {
+			b = &fw.buckets[i]
+			break
+		}
+	}
+	if b == nil {
+		t.Fatalf("missing relation bucket: %+v", fw.buckets)
 	}
-	b := fw.buckets[0]
 	if b.Type != "relation" || b.FromEntity != "A" || b.ToEntity != "B" {
 		t.Errorf("relation bucket = %+v, want Type=relation From=A To=B", b)
 	}
@@ -143,6 +156,10 @@ func TestMergeStructureDataset_RelationsNotDropped(t *testing.T) {
 func TestMergeStructureDataset_NormalizesRelationIdentity(t *testing.T) {
 	c := &Consumer{writer: &fakeWriter{}}
 	products := []kccommon.Product{
+		{Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity A",
+			Meta: map[string]any{"kind": "entity", "name": "A"}},
+		{Variant: kccommon.VariantStructure, DocID: "d1", Content: "entity B",
+			Meta: map[string]any{"kind": "entity", "name": "B"}},
 		{Variant: kccommon.VariantStructure, DocID: "d1", Content: "first",
 			Meta: map[string]any{"kind": "relation", "from": " A  ", "to": " B ", "relation_type": " Related  ", "source_chunk_ids": []string{"c1"}}},
 		{Variant: kccommon.VariantStructure, DocID: "d2", Content: "second",
@@ -152,10 +169,20 @@ func TestMergeStructureDataset_NormalizesRelationIdentity(t *testing.T) {
 		t.Fatalf("mergeStructureDataset: %v", err)
 	}
 	buckets := c.writer.(*fakeWriter).buckets
-	if len(buckets) != 1 {
-		t.Fatalf("normalized relation identity should produce one bucket, got %d: %+v", len(buckets), buckets)
+	if len(buckets) != 3 {
+		t.Fatalf("normalized relation identity should produce two entity buckets and one relation bucket, got %d: %v", len(buckets), buckets)
 	}
-	if got := len(buckets[0].SourceDocIDs); got != 2 {
+	var relationBucket *StructureBucket
+	for i := range buckets {
+		if buckets[i].Type == "relation" {
+			relationBucket = &buckets[i]
+			break
+		}
+	}
+	if relationBucket == nil {
+		t.Fatalf("missing relation bucket: %+v", buckets)
+	}
+	if got := len(relationBucket.SourceDocIDs); got != 2 {
 		t.Fatalf("relation source docs = %v, want two docs", buckets[0].SourceDocIDs)
 	}
 }
@@ -193,6 +220,9 @@ func TestMergeStructureDataset_DescriptionFallbacks(t *testing.T) {
 		{Variant: kccommon.VariantMindmap, DocID: "d1",
 			Content: `{"name":"Root","type":"mindmap"}`,
 			Meta:    map[string]any{"kind": "entity", "name": "Root", "entity_type": "mindmap"}},
+		{Variant: kccommon.VariantMindmap, DocID: "d1",
+			Content: `{"name":"Child","type":"mindmap"}`,
+			Meta:    map[string]any{"kind": "entity", "name": "Child", "entity_type": "mindmap"}},
 		{Variant: kccommon.VariantMindmap, DocID: "d1",
 			Content: `{"from":"Root","to":"Child","type":"related"}`,
 			Meta:    map[string]any{"kind": "relation", "from": "Root", "to": "Child", "relation_type": "related"}},
@@ -201,14 +231,15 @@ func TestMergeStructureDataset_DescriptionFallbacks(t *testing.T) {
 		t.Fatalf("mergeStructureDataset: %v", err)
 	}
 	fw := c.writer.(*fakeWriter)
-	if len(fw.buckets) != 2 {
-		t.Fatalf("want entity and relation buckets, got %d", len(fw.buckets))
+	if len(fw.buckets) != 3 {
+		t.Fatalf("want two entity and one relation bucket, got %d", len(fw.buckets))
 	}
 	for _, b := range fw.buckets {
 		switch b.Type {
 		case "mindmap":
-			if b.Description != "Root" {
-				t.Errorf("entity fallback description = %q, want %q", b.Description, "Root")
+			want := b.Name
+			if b.Description != want {
+				t.Errorf("entity %q fallback description = %q, want %q", b.Name, b.Description, want)
 			}
 		case "relation":
 			if b.Description != "Root related Child" {

internal/ingestion/knowledge_compile/wiki_merge.go

@@ -175,7 +175,7 @@ func unionWikiProvenance(a, b map[string]any) map[string]any {
 	}
 	// The incoming page replaces current page metadata. Identity and creation
 	// time remain from the existing row.
-	for _, key := range []string{"summary", "kind"} {
+	for _, key := range []string{"summary", "kind", "content_md_raw"} {
 		if v, ok := b[key]; ok {
 			out[key] = v
 		}