xinmi/internal/application/repository/retriever/legacygraph/write_test.go

package legacygraph

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"slices"
	"strings"
	"testing"

	"github.com/Tencent/WeKnora/internal/types"
	"gorm.io/driver/postgres"
	"gorm.io/gorm"
)

func TestNativeIdentity(t *testing.T) {
	a, identity := nativeID("doc", "entity", "concept")
	b, _ := nativeID("other-doc", "entity", "concept")
	if a < 1<<52 || a > (1<<53)-1 || a == b || len(identity) != 64 {
		t.Fatal("invalid stable document-scoped browser-safe ID")
	}
	c, again := nativeID("doc", "entity", "concept")
	if a != c || identity != again {
		t.Fatal("unstable identity")
	}
}

func TestNativeGenericEntityIdentityIsSourceScoped(t *testing.T) {
	for _, name := range []string{"本品", "本商品", "前段未名产品", "后段未名产品", "本品（前段）", "前段 未名产品", "本 品 （前段）", "This Product"} {
		t.Run(name, func(t *testing.T) {
			first, _, err := nativeEntityIdentity("doc", &types.GraphNode{Name: name, Chunks: []string{"first"}})
			if err != nil {
				t.Fatal(err)
			}
			second, _, err := nativeEntityIdentity("doc", &types.GraphNode{Name: name, Chunks: []string{"second"}})
			if err != nil || first == second {
				t.Fatal("unnamed products from separate chunks were merged", err)
			}
			for _, chunks := range [][]string{nil, {""}, {"first", "second"}} {
				if _, _, err := nativeEntityIdentity("doc", &types.GraphNode{Name: name, Chunks: chunks}); err == nil {
					t.Fatalf("missing or ambiguous source accepted: %v", chunks)
				}
			}
		})
	}
	for _, name := range []string{"Named product", "修复凝胶", "本品专用护理套装", "未名产品研究中心"} {
		for _, chunk := range []string{"first", "second"} {
			id, _, err := nativeEntityIdentity("doc", &types.GraphNode{Name: name, Chunks: []string{chunk}})
			expected, _ := nativeID("doc", "entity", normalizedName(name))
			if err != nil || id != expected {
				t.Fatal("specific product lost its stable document identity", err)
			}
		}
	}
}

func TestNativeGenericProductsDoNotShareRelations(t *testing.T) {
	for _, name := range []string{"本品", "本商品", "前段未名产品", "后段未名产品", "本品（前段）"} {
		t.Run(name, func(t *testing.T) {
			tx, r, ctx := nativeWriteFixture(t)
			if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Evidence about trust and learning.',NULL,true)`).Error; err != nil {
				t.Fatal(err)
			}
			ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
			for _, chunkID := range []string{"chunk", "sibling"} {
				graph := testGraph(chunkID)
				graph.Node[0].Name = name
				graph.Relation[0].Node1 = name
				if err := r.AddGraph(ctx, ns, []*types.GraphData{graph}); err != nil {
					t.Fatal(err)
				}
			}
			overview, err := r.Browse(ctx, "kb", "", 0, 0)
			if err != nil || overview.Counts["knowledgeentity"] != 3 || overview.Counts["knowledgerelation"] != 2 {
				t.Fatalf("separate unnamed products conflated: %#v %v", overview, err)
			}
			for _, chunkID := range []string{"chunk", "sibling"} {
				id, _, _ := nativeEntityIdentity("doc", &types.GraphNode{Name: name, Chunks: []string{chunkID}})
				detail, err := r.Browse(ctx, "kb", "", 0, id)
				if err != nil || len(detail.Relations) != 1 || len(detail.Evidence) != 1 || detail.Evidence[0].NativeChunkID != chunkID {
					t.Fatalf("unnamed product points to another source: %#v %v", detail, err)
				}
			}
			if err := r.AddGraph(ctx, ns, []*types.GraphData{{Node: []*types.GraphNode{{Name: name, Chunks: []string{"chunk", "sibling"}}}}}); err == nil {
				t.Fatal("ambiguous multi-source write accepted")
			}
			after, err := r.Browse(ctx, "kb", "", 0, 0)
			if err != nil || after.Counts["knowledgerelation"] != 2 {
				t.Fatal("failed ambiguous write damaged committed graph", err)
			}
		})
	}
}

func nativeWriteFixture(t *testing.T) (*gorm.DB, *Repository, context.Context) {
	t.Helper()
	dsn := os.Getenv("GRAPH_TEST_DSN")
	if dsn == "" {
		t.Skip("GRAPH_TEST_DSN required for temporary-table integration")
	}
	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
	if err != nil {
		t.Fatal(err)
	}
	sqlDB, err := db.DB()
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { sqlDB.Close() })
	tx := db.Begin()
	if tx.Error != nil {
		t.Fatal(tx.Error)
	}
	t.Cleanup(func() { tx.Rollback() })
	// Temporary tables shadow live names on this transaction's connection;
	// tests never mutate application records or permanent schemas.
	for _, sql := range []string{
		`CREATE TEMP TABLE knowledge_bases(id text,tenant_id bigint,deleted_at timestamptz) ON COMMIT DROP`,
		`CREATE TEMP TABLE knowledges(id text,tenant_id bigint,knowledge_base_id text,file_name text,deleted_at timestamptz) ON COMMIT DROP`,
		`CREATE TEMP TABLE chunks(id text,tenant_id bigint,knowledge_base_id text,knowledge_id text,content text,deleted_at timestamptz,is_enabled boolean) ON COMMIT DROP`,
		`CREATE TEMP TABLE native_graph_records(tenant_id bigint,knowledge_base_id text,source_table text,legacy_id bigint,graph_version_id bigint,payload jsonb,source_snapshot_sha256 text,imported_at timestamptz DEFAULT now(),PRIMARY KEY(tenant_id,knowledge_base_id,source_table,legacy_id)) ON COMMIT DROP`,
		`CREATE TEMP TABLE native_graph_sources(tenant_id bigint,knowledge_base_id text,source_chunk_id text,source_file text,source_content_hash text,excerpt text,native_knowledge_id text,native_chunk_id text,native_content_sha256 text,mapping_state text,PRIMARY KEY(tenant_id,knowledge_base_id,source_chunk_id)) ON COMMIT DROP`,
		`INSERT INTO knowledge_bases VALUES('kb',7,NULL),('kb',8,NULL),('other-kb',7,NULL)`,
		`INSERT INTO knowledges VALUES('doc',7,'kb','source.md',NULL),('doc2',7,'kb','other.md',NULL),('doc',8,'kb','private.md',NULL)`,
		`INSERT INTO chunks VALUES('chunk',7,'kb','doc','Evidence about trust and learning.',NULL,true),('chunk2',7,'kb','doc2','Evidence about trust and learning.',NULL,true),('foreign',8,'kb','doc','Private source.',NULL,true)`,
	} {
		if err := tx.Exec(sql).Error; err != nil {
			t.Fatal(err)
		}
	}
	ctx := context.WithValue(context.Background(), types.TenantIDContextKey, uint64(7))
	return tx, NewRepository(tx, nil), ctx
}

func testGraph(chunk string) *types.GraphData {
	return &types.GraphData{Node: []*types.GraphNode{{Name: "Trust", Chunks: []string{chunk}, Attributes: []string{"concept", "Earned through reliable evidence"}}, {Name: "Learning", Chunks: []string{chunk}, Attributes: []string{"process"}}}, Relation: []*types.GraphRelation{{Node1: "Trust", Node2: "Learning", Type: "supports"}}}
}

func TestNativeGraphWriteReplayBrowseAndIsolation(t *testing.T) {
	tx, r, ctx := nativeWriteFixture(t)
	ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
	chunk := &types.Chunk{ID: "chunk", TenantID: 7, KnowledgeBaseID: "kb", KnowledgeID: "doc", Content: "Evidence about trust and learning."}
	for i := 0; i < 2; i++ {
		if err := r.AddChunkGraph(ctx, ns, chunk, testGraph("chunk")); err != nil {
			t.Fatal(err)
		}
	}
	overview, err := r.Browse(ctx, "kb", "", 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if overview.Counts["knowledgeentity"] != 2 || overview.Counts["knowledgerelation"] != 1 || overview.Counts["knowledgegraphevidence"] != 3 || overview.Counts["knowledgegraphchunkstate"] != 1 || overview.MappedSources != 1 {
		t.Fatalf("not durable/idempotent: %#v", overview)
	}
	id, _ := nativeID("doc", "entity", "trust")
	detail, err := r.Browse(ctx, "kb", "", 0, id)
	if err != nil || len(detail.Relations) != 1 || len(detail.Evidence) != 1 || detail.Evidence[0].NativeChunkID != "chunk" {
		t.Fatalf("detail missing actual evidence: %#v %v", detail, err)
	}
	found, err := r.SearchNode(ctx, ns, []string{"Trust"})
	if err != nil || len(found.Node) != 2 || len(found.Relation) != 1 {
		t.Fatalf("native RAG cannot find persisted graph: %#v %v", found, err)
	}
	complete, err := r.HasChunkGraph(ctx, chunk)
	if err != nil || !complete {
		t.Fatalf("missing checkpoint: %v", err)
	}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("foreign")}); err == nil {
		t.Fatal("cross-tenant source accepted")
	}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("chunk2")}); err == nil {
		t.Fatal("cross-document source accepted")
	}
	changed := *chunk
	changed.Content = "old model input"
	if err := r.AddChunkGraph(ctx, ns, &changed, testGraph("chunk")); err == nil {
		t.Fatal("stale model output accepted")
	}
	bad := testGraph("chunk")
	bad.Relation[0].Node2 = "invented"
	if err := r.AddChunkGraph(ctx, ns, chunk, bad); err == nil {
		t.Fatal("ungrounded relation accepted")
	}
	after, err := r.Browse(ctx, "kb", "", 0, id)
	if err != nil || len(after.Evidence) != 1 {
		t.Fatal("failed extraction destroyed committed evidence", err)
	}
	if err := r.AddGraph(ctx, types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc2"}, []*types.GraphData{testGraph("chunk2")}); err != nil {
		t.Fatal(err)
	}
	if err := tx.Exec(`INSERT INTO native_graph_records VALUES(7,'kb','knowledgeentity',1,1,'{"canonical_name":"Archived","status":"active"}',repeat('a',64),now())`).Error; err != nil {
		t.Fatal(err)
	}
	if err := r.DelGraph(ctx, []types.NameSpace{ns}); err != nil {
		t.Fatal(err)
	}
	left, err := r.Browse(ctx, "kb", "", 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	if left.Counts["knowledgeentity"] != 3 || left.Sources != 1 {
		t.Fatalf("deletion damaged other document/archive: %#v", left)
	}
	wrong := context.WithValue(ctx, types.TenantIDContextKey, uint64(8))
	if err := r.DelGraph(wrong, []types.NameSpace{{KnowledgeBase: "kb"}}); err != nil {
		t.Fatal(err)
	}
	left, err = r.Browse(ctx, "kb", "", 0, 0)
	if err != nil || left.Counts["knowledgeentity"] != 3 {
		t.Fatal("cross-tenant deletion", err)
	}
}

func TestNativeEmptyResultAndReextraction(t *testing.T) {
	tx, r, ctx := nativeWriteFixture(t)
	ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
	chunk := &types.Chunk{ID: "chunk", TenantID: 7, KnowledgeBaseID: "kb", KnowledgeID: "doc", Content: "Evidence about trust and learning."}
	if err := r.AddChunkGraph(ctx, ns, chunk, testGraph("chunk")); err != nil {
		t.Fatal(err)
	}
	if err := r.AddChunkGraph(ctx, ns, chunk, &types.GraphData{}); err == nil {
		t.Fatal("empty extraction accepted without review")
	}
	overview, err := r.Browse(ctx, "kb", "", 0, 0)
	if err != nil || overview.Counts["knowledgerelation"] == 0 || overview.Counts["knowledgegraphchunkstate"] != 1 {
		t.Fatalf("rejected result replaced prior graph: %#v %v", overview, err)
	}
	complete, err := r.HasChunkGraph(ctx, chunk)
	if err != nil || !complete {
		t.Fatal("prior verified graph was lost", err)
	}
	if err := tx.Exec(`UPDATE native_graph_records SET payload=jsonb_set(payload,'{relation_count}','0'::jsonb) WHERE source_table='knowledgegraphchunkstate'`).Error; err != nil {
		t.Fatal(err)
	}
	complete, err = r.HasChunkGraph(ctx, chunk)
	if err != nil || complete {
		t.Fatal("legacy empty checkpoint was accepted", err)
	}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{{}}); err == nil {
		t.Fatal("empty extraction without source accepted")
	}
}

func TestNativeRelationQuotationIsPersistedAndChecked(t *testing.T) {
	tx, r, ctx := nativeWriteFixture(t)
	ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
	graph := testGraph("chunk")
	graph.Relation[0].Evidence = "trust and learning"
	if err := r.AddGraph(ctx, ns, []*types.GraphData{graph}); err != nil {
		t.Fatal(err)
	}
	var quote, excerpt string
	if err := tx.Raw(`SELECT payload->>'evidence' FROM native_graph_records WHERE source_table='knowledgerelation'`).Scan(&quote).Error; err != nil {
		t.Fatal(err)
	}
	if err := tx.Raw(`SELECT payload->>'excerpt' FROM native_graph_records WHERE source_table='knowledgegraphevidence' AND payload->>'subject_type'='relation'`).Scan(&excerpt).Error; err != nil {
		t.Fatal(err)
	}
	if quote != graph.Relation[0].Evidence || excerpt != quote {
		t.Fatalf("relation quotation not retained: %q %q", quote, excerpt)
	}
	graph.Relation[0].Evidence = "invented quotation"
	if err := r.AddGraph(ctx, ns, []*types.GraphData{graph}); err == nil {
		t.Fatal("relation quotation absent from source accepted")
	}
	if err := tx.Raw(`SELECT payload->>'evidence' FROM native_graph_records WHERE source_table='knowledgerelation'`).Scan(&quote).Error; err != nil || quote != "trust and learning" {
		t.Fatal("invalid quotation damaged prior graph", err)
	}
}

func TestNativeRelationRequiresCurrentSourceEvidence(t *testing.T) {
	tx, r, ctx := nativeWriteFixture(t)
	ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
	if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source discussing trust and learning.',NULL,true)`).Error; err != nil {
		t.Fatal(err)
	}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("chunk")}); err != nil {
		t.Fatal(err)
	}
	graph := testGraph("sibling")
	graph.Relation = nil
	if err := r.AddGraph(ctx, ns, []*types.GraphData{graph}); err != nil {
		t.Fatal(err)
	}
	if err := tx.Exec(`UPDATE chunks SET content='The original relation is no longer supported.' WHERE id='chunk' AND tenant_id=7`).Error; err != nil {
		t.Fatal(err)
	}
	found, err := r.SearchNode(ctx, ns, []string{"Trust"})
	if err != nil || len(found.Node) != 2 || len(found.Relation) != 0 {
		t.Fatalf("stale relation remained available through separately supported entities: %#v %v", found, err)
	}
}

func TestNativeCopiedVersionDeletionUsesRemappedSources(t *testing.T) {
	tx, r, ctx := nativeWriteFixture(t)
	ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("chunk")}); err != nil {
		t.Fatal(err)
	}
	for _, sql := range []string{
		`INSERT INTO knowledges VALUES('copy-doc',7,'other-kb','copy.md',NULL)`,
		`INSERT INTO chunks VALUES('copy-chunk',7,'other-kb','copy-doc','Evidence about trust and learning.',NULL,true)`,
		`INSERT INTO native_graph_records SELECT tenant_id,'other-kb',source_table,legacy_id,graph_version_id,payload,source_snapshot_sha256,imported_at FROM native_graph_records WHERE tenant_id=7 AND knowledge_base_id='kb'`,
		`INSERT INTO native_graph_sources SELECT tenant_id,'other-kb',source_chunk_id,source_file,source_content_hash,excerpt,'copy-doc','copy-chunk',native_content_sha256,mapping_state FROM native_graph_sources WHERE tenant_id=7 AND knowledge_base_id='kb'`,
	} {
		if err := tx.Exec(sql).Error; err != nil {
			t.Fatal(err)
		}
	}
	if err := r.DelGraph(ctx, []types.NameSpace{{KnowledgeBase: "other-kb", Knowledge: "copy-doc"}}); err != nil {
		t.Fatal(err)
	}
	copy, err := r.Browse(ctx, "other-kb", "", 0, 0)
	if err != nil || len(copy.Counts) != 0 || copy.Sources != 0 {
		t.Fatalf("copied graph was not deleted via its current document: %#v %v", copy, err)
	}
	original, err := r.Browse(ctx, "kb", "", 0, 0)
	if err != nil || original.Counts["knowledgeentity"] != 2 || original.Sources != 1 {
		t.Fatal("copy deletion changed original graph", err)
	}
}

func nativeGraphWriteState(t *testing.T, tx *gorm.DB) string {
	t.Helper()
	var state string
	if err := tx.Raw(`SELECT jsonb_build_object(
		'records',COALESCE((SELECT jsonb_agg(to_jsonb(r) ORDER BY tenant_id,knowledge_base_id,source_table,legacy_id) FROM native_graph_records r),'[]'::jsonb),
		'sources',COALESCE((SELECT jsonb_agg(to_jsonb(s) ORDER BY tenant_id,knowledge_base_id,source_chunk_id) FROM native_graph_sources s),'[]'::jsonb))::text`).Scan(&state).Error; err != nil {
		t.Fatal(err)
	}
	return state
}

func assertNativeTrustAttributes(t *testing.T, tx *gorm.DB, want []string) string {
	t.Helper()
	id, _ := nativeID("doc", "entity", "trust")
	var record struct {
		Payload json.RawMessage
		Hash    string
		State   string
	}
	if err := tx.Raw(`SELECT payload,source_snapshot_sha256 AS hash,to_jsonb(r)::text AS state FROM native_graph_records r
		WHERE tenant_id=7 AND knowledge_base_id='kb' AND source_table='knowledgeentity' AND legacy_id=?`, id).Scan(&record).Error; err != nil {
		t.Fatal(err)
	}
	var entity struct {
		Attributes []string `json:"attributes"`
		Type       string   `json:"entity_type"`
	}
	if err := json.Unmarshal(record.Payload, &entity); err != nil {
		t.Fatal(err)
	}
	wantType := "concept"
	if len(want) > 0 {
		wantType = want[0]
	}
	if !slices.Equal(entity.Attributes, want) || entity.Type != wantType {
		t.Fatalf("unexpected aggregate attributes/type: %#v; want %v / %s", entity, want, wantType)
	}
	var payload map[string]any
	if err := json.Unmarshal(record.Payload, &payload); err != nil {
		t.Fatal(err)
	}
	encoded, err := json.Marshal(payload)
	if err != nil || record.Hash != contentHash(string(encoded)) {
		t.Fatalf("entity payload hash is stale: %s %v", record.Hash, err)
	}
	return record.State
}

func TestNativeSharedEntityAttributesFollowTheirSources(t *testing.T) {
	for _, siblingHasAttributes := range []bool{false, true} {
		t.Run(fmt.Sprintf("sibling_attributes_%t", siblingHasAttributes), func(t *testing.T) {
			tx, r, ctx := nativeWriteFixture(t)
			if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source about trust and learning.',NULL,true)`).Error; err != nil {
				t.Fatal(err)
			}
			ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
			original := testGraph("chunk")
			if err := r.AddGraph(ctx, ns, []*types.GraphData{original}); err != nil {
				t.Fatal(err)
			}
			before := assertNativeTrustAttributes(t, tx, original.Node[0].Attributes)
			sibling := testGraph("sibling")
			if !siblingHasAttributes {
				sibling.Node[0].Attributes = nil
			}
			if err := r.AddGraph(ctx, ns, []*types.GraphData{sibling}); err != nil {
				t.Fatal(err)
			}
			if after := assertNativeTrustAttributes(t, tx, original.Node[0].Attributes); after != before {
				t.Fatal("a second source changed the existing entity row")
			}
			var sourceAttributes string
			id, _ := nativeID("doc", "entity", "trust")
			if err := tx.Raw(`SELECT payload->'entity_attributes' FROM native_graph_records WHERE source_table='knowledgegraphevidence'
				AND payload->>'subject_type'='entity' AND payload->>'subject_id'=? AND payload->>'source_chunk_id'='native:sibling'`, fmt.Sprint(id)).Scan(&sourceAttributes).Error; err != nil {
				t.Fatal(err)
			}
			if !siblingHasAttributes && sourceAttributes != "[]" {
				t.Fatalf("bare source inherited aggregate attributes: %s", sourceAttributes)
			}
			original.Node[0].Attributes = nil
			if err := r.AddGraph(ctx, ns, []*types.GraphData{original}); err != nil {
				t.Fatal(err)
			}
			assertNativeTrustAttributes(t, tx, sibling.Node[0].Attributes)
			sibling.Node[0].Attributes = []string{}
			if err := r.AddGraph(ctx, ns, []*types.GraphData{sibling}); err != nil {
				t.Fatal(err)
			}
			assertNativeTrustAttributes(t, tx, nil)
		})
	}
}

func TestNativeSharedEntityAttributeConflictRollsBack(t *testing.T) {
	tx, r, ctx := nativeWriteFixture(t)
	if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source about trust and learning.',NULL,true)`).Error; err != nil {
		t.Fatal(err)
	}
	ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("chunk")}); err != nil {
		t.Fatal(err)
	}
	before := nativeGraphWriteState(t, tx)
	conflicting := testGraph("sibling")
	conflicting.Node[0].Attributes = []string{"concept", "Never earned through reliable evidence"}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{conflicting}); err == nil || !strings.Contains(err.Error(), "attributes conflict") {
		t.Fatalf("conflicting source did not require review: %v", err)
	}
	if after := nativeGraphWriteState(t, tx); after != before {
		t.Fatal("failed conflict changed records, evidence, mapping or checkpoint")
	}
}

func TestNativeSharedEntityAttributesAreRemovedWithTheirNode(t *testing.T) {
	for _, siblingHasAttributes := range []bool{false, true} {
		t.Run(fmt.Sprintf("sibling_attributes_%t", siblingHasAttributes), func(t *testing.T) {
			tx, r, ctx := nativeWriteFixture(t)
			if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source about trust and learning.',NULL,true)`).Error; err != nil {
				t.Fatal(err)
			}
			ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
			original, sibling := testGraph("chunk"), testGraph("sibling")
			if !siblingHasAttributes {
				sibling.Node[0].Attributes = nil
			}
			if err := r.AddGraph(ctx, ns, []*types.GraphData{original, sibling}); err != nil {
				t.Fatal(err)
			}
			original.Node[0].Name = "Practice"
			original.Relation[0].Node1 = "Practice"
			chunk := &types.Chunk{ID: "chunk", TenantID: 7, KnowledgeBaseID: "kb", KnowledgeID: "doc", Content: "Evidence about trust and learning."}
			if err := r.AddChunkGraph(ctx, ns, chunk, original); err != nil {
				t.Fatal(err)
			}
			assertNativeTrustAttributes(t, tx, sibling.Node[0].Attributes)
			sibling.Node[0].Name = "Practice"
			sibling.Node[0].Attributes = original.Node[0].Attributes
			sibling.Relation[0].Node1 = "Practice"
			if err := r.AddGraph(ctx, ns, []*types.GraphData{sibling}); err != nil {
				t.Fatal(err)
			}
			id, _ := nativeID("doc", "entity", "trust")
			var count int64
			if err := tx.Table("native_graph_records").Where("source_table='knowledgeentity' AND legacy_id=?", id).Count(&count).Error; err != nil || count != 0 {
				t.Fatalf("entity without surviving sources was not pruned: %d %v", count, err)
			}
		})
	}
}

func TestNativeLegacyEntityAttributeProvenanceRequiresReview(t *testing.T) {
	tx, r, ctx := nativeWriteFixture(t)
	if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source about trust and learning.',NULL,true)`).Error; err != nil {
		t.Fatal(err)
	}
	ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
	if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("chunk")}); err != nil {
		t.Fatal(err)
	}
	id, _ := nativeID("doc", "entity", "trust")
	if err := tx.Exec(`UPDATE native_graph_records SET payload=payload-'entity_attributes' WHERE source_table='knowledgegraphevidence'
		AND payload->>'subject_type'='entity' AND payload->>'subject_id'=?`, fmt.Sprint(id)).Error; err != nil {
		t.Fatal(err)
	}
	before := nativeGraphWriteState(t, tx)
	bare := testGraph("sibling")
	bare.Node[0].Attributes = nil
	if err := r.AddGraph(ctx, ns, []*types.GraphData{bare}); err == nil || !strings.Contains(err.Error(), "provenance is unavailable") {
		t.Fatalf("legacy unknown provenance treated as known empty: %v", err)
	}
	if after := nativeGraphWriteState(t, tx); after != before {
		t.Fatal("unknown provenance failure changed prior graph")
	}
	// A matching explicit source can coexist with unknown older evidence.
	if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("sibling")}); err != nil {
		t.Fatal(err)
	}
	original := testGraph("chunk")
	original.Node[0].Attributes = nil
	if err := r.AddGraph(ctx, ns, []*types.GraphData{original}); err != nil {
		t.Fatal(err)
	}
	assertNativeTrustAttributes(t, tx, testGraph("sibling").Node[0].Attributes)
	if err := r.AddGraph(ctx, ns, []*types.GraphData{bare}); err != nil {
		t.Fatal(err)
	}
	assertNativeTrustAttributes(t, tx, nil)
}

func TestNativeNullEntityAttributeEvidenceRequiresReview(t *testing.T) {
	for _, snapshot := range []string{"null", "[null]"} {
		t.Run(snapshot, func(t *testing.T) {
			tx, r, ctx := nativeWriteFixture(t)
			if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source about trust and learning.',NULL,true)`).Error; err != nil {
				t.Fatal(err)
			}
			ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
			if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("chunk")}); err != nil {
				t.Fatal(err)
			}
			id, _ := nativeID("doc", "entity", "trust")
			if err := tx.Exec(`UPDATE native_graph_records SET payload=jsonb_set(payload,'{entity_attributes}',?::jsonb)
				WHERE source_table='knowledgegraphevidence' AND payload->>'subject_type'='entity' AND payload->>'subject_id'=?`, snapshot, fmt.Sprint(id)).Error; err != nil {
				t.Fatal(err)
			}
			before := nativeGraphWriteState(t, tx)
			bare := testGraph("sibling")
			bare.Node[0].Attributes = nil
			if err := r.AddGraph(ctx, ns, []*types.GraphData{bare}); err == nil {
				t.Fatal("null attribute evidence was accepted as known empty")
			}
			if after := nativeGraphWriteState(t, tx); after != before {
				t.Fatal("null attribute evidence failure changed prior graph")
			}
		})
	}
}

func TestNativeSharedEntityIgnoresInvalidAttributeEvidence(t *testing.T) {
	for name, invalidate := range map[string]string{
		"changed source":         `UPDATE chunks SET content='Changed source.' WHERE id='chunk' AND tenant_id=7`,
		"disabled source":        `UPDATE chunks SET is_enabled=false WHERE id='chunk' AND tenant_id=7`,
		"deleted source":         `UPDATE chunks SET deleted_at=now() WHERE id='chunk' AND tenant_id=7`,
		"wrong evidence hash":    `UPDATE native_graph_records SET payload=jsonb_set(payload,'{source_content_hash}',to_jsonb(repeat('a',64))) WHERE source_table='knowledgegraphevidence' AND payload->>'source_chunk_id'='native:chunk'`,
		"wrong row hash":         `UPDATE native_graph_records SET source_snapshot_sha256=repeat('a',64) WHERE source_table='knowledgegraphevidence' AND payload->>'source_chunk_id'='native:chunk'`,
		"inactive evidence":      `UPDATE native_graph_records SET payload=jsonb_set(payload,'{status}','"inactive"'::jsonb) WHERE source_table='knowledgegraphevidence' AND payload->>'source_chunk_id'='native:chunk'`,
		"wrong origin":           `UPDATE native_graph_records SET payload=jsonb_set(payload,'{origin}','"foreign"'::jsonb) WHERE source_table='knowledgegraphevidence' AND payload->>'source_chunk_id'='native:chunk'`,
		"wrong tenant":           `UPDATE native_graph_records SET tenant_id=8 WHERE source_table='knowledgegraphevidence' AND payload->>'source_chunk_id'='native:chunk'`,
		"wrong knowledge base":   `UPDATE native_graph_records SET knowledge_base_id='other-kb' WHERE source_table='knowledgegraphevidence' AND payload->>'source_chunk_id'='native:chunk'`,
		"wrong document":         `UPDATE native_graph_records SET payload=jsonb_set(payload,'{native_knowledge_id}','"doc2"'::jsonb) WHERE source_table='knowledgegraphevidence' AND payload->>'source_chunk_id'='native:chunk'`,
		"wrong mapping document": `UPDATE native_graph_sources SET native_knowledge_id='doc2' WHERE source_chunk_id='native:chunk'`,
		"inactive mapping":       `UPDATE native_graph_sources SET mapping_state='stale' WHERE source_chunk_id='native:chunk'`,
	} {
		t.Run(name, func(t *testing.T) {
			tx, r, ctx := nativeWriteFixture(t)
			if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source about trust and learning.',NULL,true)`).Error; err != nil {
				t.Fatal(err)
			}
			ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
			if err := r.AddGraph(ctx, ns, []*types.GraphData{testGraph("chunk")}); err != nil {
				t.Fatal(err)
			}
			if err := tx.Exec(invalidate).Error; err != nil {
				t.Fatal(err)
			}
			bare := testGraph("sibling")
			bare.Node[0].Attributes = nil
			if err := r.AddGraph(ctx, ns, []*types.GraphData{bare}); err != nil {
				t.Fatal(err)
			}
			assertNativeTrustAttributes(t, tx, nil)
		})
	}
}

func TestNativeEntityAttributeBatchOrderAndDuplicates(t *testing.T) {
	for _, reverse := range []bool{false, true} {
		t.Run(fmt.Sprintf("reverse_%t", reverse), func(t *testing.T) {
			tx, r, ctx := nativeWriteFixture(t)
			if err := tx.Exec(`INSERT INTO chunks VALUES('sibling',7,'kb','doc','Another source about trust and learning.',NULL,true)`).Error; err != nil {
				t.Fatal(err)
			}
			ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
			first, second := testGraph("chunk"), testGraph("sibling")
			first.Node = append(first.Node, &types.GraphNode{Name: " trust ", Chunks: []string{"chunk", "chunk"}, Attributes: first.Node[0].Attributes})
			second.Node[0].Name = "TRUST"
			second.Node[0].Attributes = nil
			graphs := []*types.GraphData{first, second}
			if reverse {
				slices.Reverse(graphs)
			}
			if err := r.AddGraph(ctx, ns, graphs); err != nil {
				t.Fatal(err)
			}
			entityBefore := assertNativeTrustAttributes(t, tx, first.Node[0].Attributes)
			slices.Reverse(graphs)
			if err := r.AddGraph(ctx, ns, graphs); err != nil {
				t.Fatal(err)
			}
			if entityAfter := assertNativeTrustAttributes(t, tx, first.Node[0].Attributes); entityAfter != entityBefore {
				t.Fatal("batch order changed the entity payload or hash")
			}
			var count int
			if err := tx.Raw(`SELECT (payload->>'entity_count')::int FROM native_graph_records WHERE source_table='knowledgegraphchunkstate' AND payload->>'native_chunk_id'='chunk'`).Scan(&count).Error; err != nil || count != 2 {
				t.Fatalf("duplicate entity evidence inflated checkpoint: %d %v", count, err)
			}
			before := nativeGraphWriteState(t, tx)
			first.Node[2].Attributes = []string{"conflicting duplicate"}
			if err := r.AddGraph(ctx, ns, graphs); err == nil || !strings.Contains(err.Error(), "attributes conflict") {
				t.Fatalf("duplicate source attributes silently overwritten: %v", err)
			}
			if after := nativeGraphWriteState(t, tx); after != before {
				t.Fatal("failed duplicate source write changed the committed graph")
			}
		})
	}
}

func TestNativeLegacyEntityAttributeBatchOrder(t *testing.T) {
	for _, reverse := range []bool{false, true} {
		t.Run(fmt.Sprintf("reverse_%t", reverse), func(t *testing.T) {
			tx, r, ctx := nativeWriteFixture(t)
			if err := tx.Exec(`INSERT INTO chunks VALUES('bare',7,'kb','doc','A bare reference to trust.',NULL,true),('explicit',7,'kb','doc','An explicit source about trust.',NULL,true)`).Error; err != nil {
				t.Fatal(err)
			}
			ns := types.NameSpace{KnowledgeBase: "kb", Knowledge: "doc"}
			original := testGraph("chunk")
			if err := r.AddGraph(ctx, ns, []*types.GraphData{original}); err != nil {
				t.Fatal(err)
			}
			if err := tx.Exec(`UPDATE native_graph_records SET payload=payload-'entity_attributes' WHERE source_table='knowledgegraphevidence' AND payload->>'subject_type'='entity'`).Error; err != nil {
				t.Fatal(err)
			}
			bare, explicit := testGraph("bare"), testGraph("explicit")
			bare.Node[0].Attributes = nil
			graphs := []*types.GraphData{bare, explicit}
			if reverse {
				slices.Reverse(graphs)
			}
			if err := r.AddGraph(ctx, ns, graphs); err != nil {
				t.Fatal(err)
			}
			assertNativeTrustAttributes(t, tx, original.Node[0].Attributes)
			original.Node[0].Attributes = nil
			explicit.Node[0].Attributes = nil
			graphs = append(graphs, original)
			if err := r.AddGraph(ctx, ns, graphs); err != nil {
				t.Fatal(err)
			}
			assertNativeTrustAttributes(t, tx, nil)
		})
	}
}