Reuse sorted owner buckets within each system-grouping pass

Build complete hostname/IP/form owner indexes once in priority order, visit only strictly better owners and reject once two distinct qualifying roots prove ambiguity. Preserve exact/short/IP evidence precedence, metadata freshness, exclusions and all resources. Bind the immediate-parent algorithm and enumerate targets independently; exercise real provider grouping and connected Infrastructure output.

Contract-Neutral: Pass-local representation and cost only; complete inventory, strict better-priority and ambiguity rules, evidence ordering, identity, public schemas, observations and alert policy are unchanged and covered by exhaustive and connected reference controls.
Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/unified-resources.md

@@ -5220,6 +5220,17 @@ Shared grouping may attach `qnap` to `qnap.local` when one surface reports the
 short hostname and another reports the FQDN, but it must not collapse two
 distinct fully-qualified hosts that merely share the same short prefix across
 different domains.
+Fallback owner buckets are pass-local priority-ordered slices, rebuilt from
+each complete group snapshot. Queries visit only strictly better-priority
+owners; a second distinct qualifying root still rejects attachment even when
+its priority or evidence kind differs. The unique root keeps lexical exact-host
+evidence before full short/FQDN evidence before lexical exact-IP evidence.
+Comparable/short-form keys only prune candidates: the full hostname equivalence
+check remains authoritative. No index survives a grouping pass or caller, and
+no resource, reporting surface or ambiguous identity is discarded. Exhaustive
+all-group target/evidence enumeration, immediate-parent content comparisons,
+multi-pass identity changes and connected provider/Infrastructure controls
+verify this unchanged grouping contract.
 That same projection contract now also owns structured replacement selectors
 and detailed previews. Shared callers may serialize source-native selector
 fields such as hostname, host URL, machine or agent identity, and source-owned
@@ -5922,4 +5933,3 @@ above the hosts table without its filter toolbar. Nothing renders when
 vCenter reports no signals. The Overview's own model, workload snapshot and
 navigation facets still come from the Overview query; the Health tab keeps
 the filterable table.
-

internal/monitoring/connected_infrastructure_test.go

@@ -400,3 +400,29 @@ func TestBuildConnectedInfrastructure_ResolvesLegacyAgentUpgradePlatform(t *test
 		})
 	}
 }
+
+func TestBuildConnectedInfrastructure_AmbiguousFallbackKeepsEverySurface(t *testing.T) {
+	resources := []unifiedresources.Resource{
+		{ID: "platform-a", Type: unifiedresources.ResourceTypeAgent, Name: "Platform A", Proxmox: &unifiedresources.ProxmoxData{SourceID: "pve-a", NodeName: "tower.lab"}},
+		{ID: "platform-b", Type: unifiedresources.ResourceTypeAgent, Name: "Platform B", Proxmox: &unifiedresources.ProxmoxData{SourceID: "pve-b", NodeName: "tower.lab"}},
+		{ID: "host", Type: unifiedresources.ResourceTypeAgent, Name: "Host", Agent: &unifiedresources.AgentData{AgentID: "host-agent", Hostname: "tower"}},
+	}
+	items := buildConnectedInfrastructure(resources, models.StateSnapshot{})
+	if len(items) != 3 {
+		t.Fatalf("ambiguous hostname attached to arbitrary platform: %d items", len(items))
+	}
+	surfaces := map[string]bool{}
+	for _, item := range items {
+		for _, surface := range item.Surfaces {
+			if surfaces[surface.ID] {
+				t.Fatalf("duplicated surface %q", surface.ID)
+			}
+			surfaces[surface.ID] = true
+		}
+	}
+	for _, id := range []string{"proxmox:pve-a", "proxmox:pve-b", "agent:host-agent"} {
+		if !surfaces[id] {
+			t.Fatalf("ambiguity dropped surface %q (got %v)", id, surfaces)
+		}
+	}
+}

internal/unifiedresources/top_level_owner_bench_test.go

@@ -0,0 +1,78 @@
+package unifiedresources
+
+import "testing"
+
+var benchmarkTopLevelOwnerResolver TopLevelSystemResolver
+var benchmarkTopLevelOwnerAttachments int
+
+// Fixed synthetic topologies. Both variants include all index construction;
+// resolver comparisons also include the matcher, all passes and full output.
+// These are not installed fleet CPU/RSS or release-performance thresholds.
+func BenchmarkTopLevelOwnerIndexResolver(b *testing.B) {
+	for _, fixture := range topLevelOwnerIndexFixtures() {
+		b.Run(fixture.name, func(b *testing.B) {
+			for _, method := range []struct {
+				name    string
+				resolve func([]Resource) TopLevelSystemResolver
+			}{{"before", resolveTopLevelSystemsBeforeOwnerIndex}, {"indexed", ResolveTopLevelSystems}} {
+				b.Run(method.name, func(b *testing.B) {
+					b.ReportAllocs()
+					for i := 0; i < b.N; i++ {
+						benchmarkTopLevelOwnerResolver = method.resolve(fixture.resources)
+					}
+				})
+			}
+		})
+	}
+}
+
+func BenchmarkTopLevelOwnerIndexPass(b *testing.B) {
+	for _, fixture := range topLevelOwnerIndexFixtures() {
+		nodes := make([]topLevelSystemNode, 0, len(fixture.resources))
+		parents := make([]int, len(fixture.resources))
+		for i := range fixture.resources {
+			nodes = append(nodes, buildTopLevelSystemNode(&fixture.resources[i]))
+			parents[i] = i
+		}
+		groups := buildTopLevelSystemResolvedGroups(nodes, parents, nil)
+		best := int(^uint(0) >> 1)
+		for _, group := range groups {
+			if group.priority < best {
+				best = group.priority
+			}
+		}
+		b.Run(fixture.name, func(b *testing.B) {
+			for _, method := range []string{"before", "indexed"} {
+				b.Run(method, func(b *testing.B) {
+					b.ReportAllocs()
+					for i := 0; i < b.N; i++ {
+						attached := 0
+						if method == "before" {
+							hosts, ips := buildTopLevelSystemFallbackOwners(groups)
+							forms := buildTopLevelSystemHostFormOwners(groups)
+							for root, group := range groups {
+								if !group.attachByHost || group.priority <= best {
+									continue
+								}
+								if _, ok := uniqueBetterTopLevelSystemTargetBeforeOwnerIndex(root, group, hosts, ips, forms, groups); ok {
+									attached++
+								}
+							}
+						} else {
+							index := buildTopLevelSystemFallbackIndex(groups)
+							for root, group := range groups {
+								if !group.attachByHost || group.priority <= best {
+									continue
+								}
+								if _, ok := uniqueBetterTopLevelSystemTarget(root, group, index, groups); ok {
+									attached++
+								}
+							}
+						}
+						benchmarkTopLevelOwnerAttachments = attached
+					}
+				})
+			}
+		})
+	}
+}

internal/unifiedresources/top_level_owner_index_test.go

@@ -0,0 +1,325 @@
+package unifiedresources
+
+import (
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"reflect"
+	"sort"
+	"testing"
+)
+
+// Enumerate every better group, without an owner/form index, candidate pruning
+// or early ambiguity return. This oracle intentionally does more work than the
+// implementation and checks the whole target/evidence, not just group counts.
+func exhaustiveTopLevelFallbackTarget(root int, group topLevelSystemResolvedGroup, groups map[int]topLevelSystemResolvedGroup) (topLevelSystemFallbackTarget, bool) {
+	roots := make([]int, 0, len(groups))
+	for candidate := range groups {
+		if candidate != root && groups[candidate].priority < group.priority {
+			roots = append(roots, candidate)
+		}
+	}
+	sort.Ints(roots)
+	targets := make(map[int]topLevelSystemGroupingEvidence)
+	for _, host := range ownerOracleOrderedStrings(group.exactHosts) {
+		for _, candidate := range roots {
+			if _, present := groups[candidate].exactHosts[host]; present {
+				if _, already := targets[candidate]; !already {
+					targets[candidate] = topLevelSystemAttachmentEvidence(group, groups[candidate], "exact-host-attachment", "exact-host", host)
+				}
+			}
+		}
+	}
+	for _, candidate := range roots {
+		if _, already := targets[candidate]; already {
+			continue
+		}
+		left, right := ownerOracleOrderedStrings(group.exactHosts), ownerOracleOrderedStrings(groups[candidate].exactHosts)
+		matched := false
+		for _, a := range left {
+			for _, b := range right {
+				if a != b && HostnamesEquivalent(a, b) {
+					targets[candidate] = topLevelSystemAttachmentEvidence(group, groups[candidate], "hostname-form-attachment", "short-hostname", topLevelSystemHostMatchValue(a, b))
+					matched = true
+					break
+				}
+			}
+			if matched {
+				break
+			}
+		}
+	}
+	for _, ip := range ownerOracleOrderedStrings(group.exactIPs) {
+		for _, candidate := range roots {
+			if _, present := groups[candidate].exactIPs[ip]; present {
+				if _, already := targets[candidate]; !already {
+					targets[candidate] = topLevelSystemAttachmentEvidence(group, groups[candidate], "exact-ip-attachment", "exact-ip", ip)
+				}
+			}
+		}
+	}
+	if len(targets) != 1 {
+		return topLevelSystemFallbackTarget{}, false
+	}
+	for candidate, evidence := range targets {
+		return topLevelSystemFallbackTarget{root: candidate, evidence: evidence}, true
+	}
+	return topLevelSystemFallbackTarget{}, false
+}
+
+func ownerOracleOrderedStrings(values map[string]struct{}) []string {
+	result := make([]string, 0, len(values))
+	for value := range values {
+		result = append(result, value)
+	}
+	sort.Strings(result)
+	return result
+}
+
+func ownerTestSet(values ...string) map[string]struct{} {
+	set := make(map[string]struct{}, len(values))
+	for _, value := range values {
+		set[value] = struct{}{}
+	}
+	return set
+}
+
+func ownerTestGroup(priority int, hosts, ips []string) topLevelSystemResolvedGroup {
+	resource := &Resource{ID: fmt.Sprintf("priority-%d", priority), Type: ResourceTypeAgent}
+	return topLevelSystemResolvedGroup{
+		priority: priority, attachByHost: len(hosts)+len(ips) > 0,
+		resources: []*Resource{resource}, exactHosts: ownerTestSet(hosts...), exactIPs: ownerTestSet(ips...),
+	}
+}
+
+func assertOwnerTargetMatchesOracle(t *testing.T, root int, groups map[int]topLevelSystemResolvedGroup, index topLevelSystemFallbackIndex) {
+	t.Helper()
+	got, gotOK := uniqueBetterTopLevelSystemTarget(root, groups[root], index, groups)
+	want, wantOK := exhaustiveTopLevelFallbackTarget(root, groups[root], groups)
+	if gotOK != wantOK || got != want {
+		t.Fatalf("fallback differs from exhaustive oracle: root=%d groups=%#v got=%+v/%t want=%+v/%t", root, groups, got, gotOK, want, wantOK)
+	}
+	hosts, ips := buildTopLevelSystemFallbackOwners(groups)
+	forms := buildTopLevelSystemHostFormOwners(groups)
+	before, beforeOK := uniqueBetterTopLevelSystemTargetBeforeOwnerIndex(root, groups[root], hosts, ips, forms, groups)
+	if beforeOK != wantOK || before != want {
+		t.Fatalf("immediate-parent reference differs from oracle: got=%+v/%t want=%+v/%t", before, beforeOK, want, wantOK)
+	}
+}
+
+func TestTopLevelOwnerIndexExhaustive(t *testing.T) {
+	hostSets := [][]string{nil, {"alpha"}, {"alpha.lab"}, {"alpha.other"}, {"beta"}, {"alpha", "alpha.lab", "beta"}}
+	ipSets := [][]string{nil, {"192.0.2.8"}}
+	var variants []topLevelSystemResolvedGroup
+	for priority := 0; priority < 3; priority++ {
+		for _, hosts := range hostSets {
+			for _, ips := range ipSets {
+				variants = append(variants, ownerTestGroup(priority, hosts, ips))
+			}
+		}
+	}
+	checked := 0
+	for _, source := range variants {
+		for _, left := range variants {
+			for _, right := range variants {
+				// Sparse roots, including zero, must not be treated as slice positions
+				// or as an unset selected target.
+				groups := map[int]topLevelSystemResolvedGroup{0: left, 7: source, 42: right}
+				assertOwnerTargetMatchesOracle(t, 7, groups, buildTopLevelSystemFallbackIndex(groups))
+				checked++
+			}
+		}
+	}
+	if checked != 46656 {
+		t.Fatalf("exhaustive fixture lost coverage: %d", checked)
+	}
+	t.Logf("checked %d complete priority/hostname/IP target-and-evidence combinations", checked)
+}
+
+func TestTopLevelOwnerIndexRandomized(t *testing.T) {
+	rng := rand.New(rand.NewSource(2199))
+	names := []string{"alpha", "alpha.lab", "alpha.other", " ALPHA.LAB. ", "beta", "beta.lab", "beta.other", "192.0.2.8", "", "localhost", "foo-bar", "foo-bar.example"}
+	ips := []string{"192.0.2.8", "192.0.2.9", "2001:db8::8"}
+	for trial := 0; trial < 256; trial++ {
+		groups := make(map[int]topLevelSystemResolvedGroup)
+		for i := 0; i < 16; i++ {
+			var hosts, addresses []string
+			for j := 0; j < rng.Intn(5); j++ {
+				hosts = append(hosts, names[rng.Intn(len(names))])
+			}
+			for j := 0; j < rng.Intn(3); j++ {
+				addresses = append(addresses, ips[rng.Intn(len(ips))])
+			}
+			groups[i*17] = ownerTestGroup(rng.Intn(6), hosts, addresses)
+		}
+		index := buildTopLevelSystemFallbackIndex(groups)
+		for root := range groups {
+			assertOwnerTargetMatchesOracle(t, root, groups, index)
+		}
+	}
+}
+
+func TestTopLevelOwnerIndexBucketsAreOrderedAndDeduplicated(t *testing.T) {
+	groups := map[int]topLevelSystemResolvedGroup{
+		1: ownerTestGroup(2, []string{"alpha", "alpha.lab", "alpha.other"}, []string{"192.0.2.8"}),
+		9: ownerTestGroup(0, []string{"alpha", "alpha.lab"}, []string{"192.0.2.8"}),
+		4: ownerTestGroup(1, []string{"alpha", "alpha.other"}, []string{"192.0.2.8"}),
+		3: ownerTestGroup(1, []string{"alpha", "alpha.lab"}, []string{"192.0.2.8"}),
+	}
+	index := buildTopLevelSystemFallbackIndex(groups)
+	for name, bucket := range map[string][]int{"host": index.hosts["alpha"], "ip": index.ips["192.0.2.8"], "form": index.hostForms["alpha"]} {
+		if !reflect.DeepEqual(bucket, []int{9, 3, 4, 1}) {
+			t.Fatalf("%s owners not priority/root ordered and deduplicated: %v", name, bucket)
+		}
+	}
+}
+
+func TestTopLevelOwnerIndexEvidenceAndAmbiguity(t *testing.T) {
+	tests := []struct {
+		name                string
+		source              topLevelSystemResolvedGroup
+		targets             []topLevelSystemResolvedGroup
+		wantOK              bool
+		wantKind, wantValue string
+	}{
+		{"exact evidence precedes short and IP for same target", ownerTestGroup(3, []string{"alpha.lab", "alpha", "beta"}, []string{"192.0.2.8"}), []topLevelSystemResolvedGroup{ownerTestGroup(0, []string{"alpha.lab", "alpha", "beta"}, []string{"192.0.2.8"})}, true, "exact-host-attachment", "alpha"},
+		{"sorted short form evidence", ownerTestGroup(3, []string{"alpha", "beta"}, nil), []topLevelSystemResolvedGroup{ownerTestGroup(0, []string{"beta.lab", "alpha.lab"}, nil)}, true, "hostname-form-attachment", "alpha.lab"},
+		{"sorted IP evidence", ownerTestGroup(3, nil, []string{"192.0.2.9", "192.0.2.8"}), []topLevelSystemResolvedGroup{ownerTestGroup(0, nil, []string{"192.0.2.8", "192.0.2.9"})}, true, "exact-ip-attachment", "192.0.2.8"},
+		{"exact and IP disagree", ownerTestGroup(3, []string{"alpha"}, []string{"192.0.2.8"}), []topLevelSystemResolvedGroup{ownerTestGroup(0, []string{"alpha"}, nil), ownerTestGroup(1, nil, []string{"192.0.2.8"})}, false, "", ""},
+		{"different priorities do not choose best ambiguous owner", ownerTestGroup(3, []string{"alpha"}, nil), []topLevelSystemResolvedGroup{ownerTestGroup(0, []string{"alpha.lab"}, nil), ownerTestGroup(1, []string{"alpha.other"}, nil)}, false, "", ""},
+		{"distinct FQDNs sharing a short name are not equivalent", ownerTestGroup(3, []string{"alpha.lab"}, nil), []topLevelSystemResolvedGroup{ownerTestGroup(0, []string{"alpha.other"}, nil)}, false, "", ""},
+		{"equal priority cannot attach", ownerTestGroup(1, []string{"alpha"}, nil), []topLevelSystemResolvedGroup{ownerTestGroup(1, []string{"alpha"}, nil)}, false, "", ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			groups := map[int]topLevelSystemResolvedGroup{42: tc.source}
+			for i, target := range tc.targets {
+				groups[i] = target
+			}
+			index := buildTopLevelSystemFallbackIndex(groups)
+			assertOwnerTargetMatchesOracle(t, 42, groups, index)
+			got, ok := uniqueBetterTopLevelSystemTarget(42, tc.source, index, groups)
+			if ok != tc.wantOK || got.evidence.kind != tc.wantKind || got.evidence.value != tc.wantValue {
+				t.Fatalf("wrong target/evidence: %+v/%t", got, ok)
+			}
+		})
+	}
+}
+
+func TestTopLevelOwnerIndexResolverPreservesEveryResourceAndRecord(t *testing.T) {
+	for _, fixture := range topLevelOwnerIndexFixtures() {
+		t.Run(fixture.name, func(t *testing.T) {
+			got, want := ResolveTopLevelSystems(fixture.resources), resolveTopLevelSystemsBeforeOwnerIndex(fixture.resources)
+			if !reflect.DeepEqual(got.resourceToGroup, want.resourceToGroup) {
+				t.Fatal("complete resource-to-group mapping changed")
+			}
+			members := 0
+			for _, group := range got.groups {
+				members += len(group.resources)
+			}
+			if members != len(fixture.resources) {
+				t.Fatalf("inventory lost members: %d/%d", members, len(fixture.resources))
+			}
+			actual, err := json.Marshal(got.records())
+			if err != nil {
+				t.Fatal(err)
+			}
+			reference, err := json.Marshal(want.records())
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(actual) != string(reference) {
+				t.Fatal("ordered records, IDs, surfaces or grouping explanations changed")
+			}
+			t.Logf("retained %d complete resources in %d groups with byte-equal records", members, got.Count())
+		})
+	}
+}
+
+type topLevelOwnerIndexFixture struct {
+	name      string
+	resources []Resource
+}
+
+func topLevelOwnerIndexFixtures() []topLevelOwnerIndexFixture {
+	fixtures := []topLevelOwnerIndexFixture{{name: "empty"}, {name: "one", resources: []Resource{topLevelTestAgent("agent", "tower.lab", "machine", "agent")}}}
+	for _, name := range []string{"spread-1220", "shared-exact-1220", "shared-short-1220", "ambiguous-1220", "different-fqdns-200"} {
+		count, nodes := 1220, 1
+		switch name {
+		case "spread-1220":
+			nodes = 20
+		case "ambiguous-1220":
+			nodes = 2
+		case "different-fqdns-200":
+			count, nodes = 200, 100
+		}
+		resources := make([]Resource, 0, count)
+		for i := 0; i < nodes; i++ {
+			host := "tower.lab"
+			switch name {
+			case "spread-1220":
+				host = fmt.Sprintf("node-%d.lab", i)
+			case "different-fqdns-200":
+				host = fmt.Sprintf("tower.domain-%d", i)
+			}
+			resources = append(resources, topLevelTestProxmoxNode(fmt.Sprintf("node-%d", i), host, fmt.Sprintf("pve-%d", i), "https://"+host+":8006"))
+		}
+		for i := nodes; i < count; i++ {
+			host := "tower.lab"
+			switch name {
+			case "spread-1220":
+				host = fmt.Sprintf("node-%d.lab", i%nodes)
+			case "shared-short-1220":
+				host = "tower"
+			case "different-fqdns-200":
+				host = fmt.Sprintf("tower.other-%d", i)
+			}
+			resources = append(resources, topLevelTestAgent(fmt.Sprintf("agent-%d", i), host, fmt.Sprintf("machine-%d", i), fmt.Sprintf("agent-id-%d", i)))
+		}
+		fixtures = append(fixtures, topLevelOwnerIndexFixture{name: name, resources: resources})
+	}
+	return fixtures
+}
+
+// Test-only bridges let the external package exercise actual provider adapters
+// and mock graphs without importing mock (which itself uses this package).
+func CheckTopLevelOwnerIndexResourcesForTest(t testing.TB, resources []Resource) {
+	t.Helper()
+	got, want := ResolveTopLevelSystems(resources), resolveTopLevelSystemsBeforeOwnerIndex(resources)
+	if !reflect.DeepEqual(got.resourceToGroup, want.resourceToGroup) {
+		t.Fatal("connected resource-to-group mapping changed")
+	}
+	actual, err := json.Marshal(got.records())
+	if err != nil {
+		t.Fatal(err)
+	}
+	reference, err := json.Marshal(want.records())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(actual) != string(reference) {
+		t.Fatal("connected ordered records or grouping explanation changed")
+	}
+	members := 0
+	for _, group := range got.groups {
+		members += len(group.resources)
+	}
+	if members != len(resources) {
+		t.Fatalf("connected inventory lost members: %d/%d", members, len(resources))
+	}
+	t.Logf("connected grouping retains %d resources in %d groups with byte-equal records", members, got.Count())
+}
+
+func BenchmarkTopLevelOwnerIndexResourcesForTest(b *testing.B, resources []Resource) {
+	for _, method := range []struct {
+		name    string
+		resolve func([]Resource) TopLevelSystemResolver
+	}{{"before", resolveTopLevelSystemsBeforeOwnerIndex}, {"indexed", ResolveTopLevelSystems}} {
+		b.Run(method.name, func(b *testing.B) {
+			b.ReportAllocs()
+			for i := 0; i < b.N; i++ {
+				benchmarkTopLevelOwnerResolver = method.resolve(resources)
+			}
+		})
+	}
+}

internal/unifiedresources/top_level_owner_integration_test.go

@@ -0,0 +1,62 @@
+package unifiedresources_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/mock"
+	unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+)
+
+func connectedOwnerIndexResources(t testing.TB) []unified.Resource {
+	t.Helper()
+	previousEnabled, previousConfig := mock.IsMockEnabled(), mock.GetConfig()
+	t.Cleanup(func() {
+		_ = mock.SetEnabled(false)
+		mock.SetMockConfig(previousConfig)
+		if previousEnabled {
+			if err := mock.SetEnabled(true); err != nil {
+				t.Error(err)
+			}
+		}
+	})
+	if err := mock.SetEnabled(false); err != nil {
+		t.Fatal(err)
+	}
+	cfg := mock.DefaultConfig
+	cfg.UpdateInterval = 5 * time.Minute
+	mock.SetMockConfig(cfg)
+	if err := mock.SetEnabled(true); err != nil {
+		t.Fatal(err)
+	}
+	graph := mock.CurrentFixtureGraph()
+	registry := unified.NewRegistry(nil)
+	registry.IngestSnapshot(graph.State)
+	for _, source := range []unified.DataSource{unified.SourceTrueNAS, unified.SourceVMware, unified.SourceAvailability} {
+		registry.IngestRecords(source, graph.SupplementalRecords(source))
+	}
+	resources := registry.List()
+	if len(resources) < 1000 {
+		t.Fatalf("connected fixture lost estate: %d resources", len(resources))
+	}
+	providers := map[unified.DataSource]bool{}
+	for _, resource := range resources {
+		for _, source := range resource.Sources {
+			providers[source] = true
+		}
+	}
+	if len(providers) < 9 {
+		t.Fatalf("connected fixture lost providers: %v", providers)
+	}
+	return resources
+}
+
+func TestTopLevelOwnerIndexConnectedDemo(t *testing.T) {
+	resources := connectedOwnerIndexResources(t)
+	unified.CheckTopLevelOwnerIndexResourcesForTest(t, resources)
+}
+
+func BenchmarkTopLevelOwnerIndexConnectedDemo(b *testing.B) {
+	resources := connectedOwnerIndexResources(b)
+	unified.BenchmarkTopLevelOwnerIndexResourcesForTest(b, resources)
+}

internal/unifiedresources/top_level_owner_reference_test.go

@@ -0,0 +1,334 @@
+package unifiedresources
+
+import (
+	"sort"
+	"strings"
+)
+
+// Exact production grouping before pass-local owner indexes at
+// 70c71ad5b8baa519a6ae5ab255a78d9d5187300e. Keep the immediate-parent
+// algorithm for cost comparisons, not the older broad matcher reference.
+
+func resolveTopLevelSystemsBeforeOwnerIndex(resources []Resource) TopLevelSystemResolver {
+	if len(resources) == 0 {
+		return TopLevelSystemResolver{
+			groups:          nil,
+			resourceToGroup: map[string]string{},
+		}
+	}
+
+	nodes := make([]topLevelSystemNode, 0, len(resources))
+	for i := range resources {
+		node := buildTopLevelSystemNode(&resources[i])
+		if node.resource == nil {
+			continue
+		}
+		nodes = append(nodes, node)
+	}
+	if len(nodes) == 0 {
+		return TopLevelSystemResolver{
+			groups:          nil,
+			resourceToGroup: map[string]string{},
+		}
+	}
+
+	parent := make([]int, len(nodes))
+	for i := range parent {
+		parent[i] = i
+	}
+	groupEvidence := make(map[int][]topLevelSystemGroupingEvidence, len(nodes))
+
+	find := func(id int) int {
+		for parent[id] != id {
+			parent[id] = parent[parent[id]]
+			id = parent[id]
+		}
+		return id
+	}
+
+	union := func(left, right int, evidence topLevelSystemGroupingEvidence) {
+		leftRoot := find(left)
+		rightRoot := find(right)
+		if leftRoot == rightRoot {
+			return
+		}
+		if existing := groupEvidence[leftRoot]; len(existing) > 0 {
+			groupEvidence[rightRoot] = append(groupEvidence[rightRoot], existing...)
+			delete(groupEvidence, leftRoot)
+		}
+		parent[leftRoot] = rightRoot
+		if evidence.kind != "" {
+			groupEvidence[rightRoot] = append(groupEvidence[rightRoot], evidence)
+		}
+	}
+
+	strongIDOwners := make(map[string]int)
+	for i := range nodes {
+		for _, strongID := range topLevelSystemOrderedStrongIDs(nodes[i].strongIDs) {
+			if existing, ok := strongIDOwners[strongID]; ok {
+				union(i, existing, topLevelSystemStrongIDEvidence(nodes[i].resource, nodes[existing].resource, strongID))
+				continue
+			}
+			strongIDOwners[strongID] = i
+		}
+	}
+
+	matcher := NewIdentityMatcher()
+	nodeIDByMatcherID := make(map[string]int, len(nodes))
+	for i := range nodes {
+		matcherID := topLevelSystemMatcherID(i)
+		nodeIDByMatcherID[matcherID] = i
+		matcher.Add(matcherID, nodes[i].identity)
+	}
+	for i := range nodes {
+		matches := matcher.findCandidatesAtLeast(nodes[i].identity, HighConfidenceThreshold)
+		for _, match := range matches {
+			if match.Confidence < HighConfidenceThreshold {
+				continue
+			}
+			existingIndex, ok := nodeIDByMatcherID[match.ID]
+			if !ok || existingIndex == i {
+				continue
+			}
+			union(i, existingIndex, topLevelSystemIdentityMatchEvidence(nodes[i].resource, nodes[existingIndex].resource, match.Reason))
+		}
+	}
+
+	for {
+		initialGroups := buildTopLevelSystemResolvedGroups(nodes, parent, groupEvidence)
+		bestPriority := int(^uint(0) >> 1)
+		for _, group := range initialGroups {
+			if group.priority < bestPriority {
+				bestPriority = group.priority
+			}
+		}
+		hostOwners, ipOwners := buildTopLevelSystemFallbackOwners(initialGroups)
+		hostFormOwners := buildTopLevelSystemHostFormOwners(initialGroups)
+		attached := false
+
+		for groupRoot, group := range initialGroups {
+			if !group.attachByHost || group.priority <= bestPriority {
+				continue
+			}
+			target, ok := uniqueBetterTopLevelSystemTargetBeforeOwnerIndex(groupRoot, group, hostOwners, ipOwners, hostFormOwners, initialGroups)
+			if !ok {
+				continue
+			}
+			union(groupRoot, target.root, target.evidence)
+			attached = true
+		}
+
+		if !attached {
+			finalGroups := buildTopLevelSystemResolvedGroups(nodes, parent, groupEvidence)
+			orderedGroups := make([]topLevelSystemResolvedGroup, 0, len(finalGroups))
+			for _, group := range finalGroups {
+				orderedGroups = append(orderedGroups, group)
+			}
+			sort.Slice(orderedGroups, func(i, j int) bool {
+				if orderedGroups[i].priority != orderedGroups[j].priority {
+					return orderedGroups[i].priority < orderedGroups[j].priority
+				}
+				if len(orderedGroups[i].resources) == 0 || len(orderedGroups[j].resources) == 0 {
+					return orderedGroups[i].id < orderedGroups[j].id
+				}
+				left := monitoredSystemDisplayName(orderedGroups[i].resources, preferredMonitoredSystemResource(orderedGroups[i].resources))
+				right := monitoredSystemDisplayName(orderedGroups[j].resources, preferredMonitoredSystemResource(orderedGroups[j].resources))
+				if left == right {
+					return orderedGroups[i].id < orderedGroups[j].id
+				}
+				return left < right
+			})
+
+			resourceToGroup := make(map[string]string, len(nodes))
+			for _, group := range orderedGroups {
+				for _, resource := range group.resources {
+					if resource == nil || strings.TrimSpace(resource.ID) == "" {
+						continue
+					}
+					resourceToGroup[strings.TrimSpace(resource.ID)] = group.id
+				}
+			}
+
+			return TopLevelSystemResolver{
+				groups:          orderedGroups,
+				resourceToGroup: resourceToGroup,
+			}
+		}
+	}
+}
+
+func buildTopLevelSystemFallbackOwners(
+	groups map[int]topLevelSystemResolvedGroup,
+) (map[string]map[int]struct{}, map[string]map[int]struct{}) {
+	hostOwners := make(map[string]map[int]struct{})
+	ipOwners := make(map[string]map[int]struct{})
+	for groupRoot, group := range groups {
+		for host := range group.exactHosts {
+			bucket := hostOwners[host]
+			if bucket == nil {
+				bucket = make(map[int]struct{})
+				hostOwners[host] = bucket
+			}
+			bucket[groupRoot] = struct{}{}
+		}
+		for ip := range group.exactIPs {
+			bucket := ipOwners[ip]
+			if bucket == nil {
+				bucket = make(map[int]struct{})
+				ipOwners[ip] = bucket
+			}
+			bucket[groupRoot] = struct{}{}
+		}
+	}
+	return hostOwners, ipOwners
+}
+
+func buildTopLevelSystemHostFormOwners(
+	groups map[int]topLevelSystemResolvedGroup,
+) map[string]map[int]struct{} {
+	owners := make(map[string]map[int]struct{})
+	add := func(key string, root int) {
+		if key == "" {
+			return
+		}
+		bucket := owners[key]
+		if bucket == nil {
+			bucket = make(map[int]struct{})
+			owners[key] = bucket
+		}
+		bucket[root] = struct{}{}
+	}
+	for groupRoot, group := range groups {
+		for host := range group.exactHosts {
+			comparable := normalizeComparableHostname(host)
+			if comparable == "" {
+				continue
+			}
+			add(comparable, groupRoot)
+			if short := NormalizeHostname(comparable); short != "" && short != comparable {
+				add(short, groupRoot)
+			}
+		}
+	}
+	return owners
+}
+
+func uniqueBetterTopLevelSystemTargetBeforeOwnerIndex(
+	groupRoot int,
+	group topLevelSystemResolvedGroup,
+	hostOwners map[string]map[int]struct{},
+	ipOwners map[string]map[int]struct{},
+	hostFormOwners map[string]map[int]struct{},
+	groups map[int]topLevelSystemResolvedGroup,
+) (topLevelSystemFallbackTarget, bool) {
+	targets := make(map[int]topLevelSystemGroupingEvidence)
+	candidateRoots := make(map[int]struct{})
+	for host := range group.exactHosts {
+		comparable := normalizeComparableHostname(host)
+		if comparable == "" {
+			continue
+		}
+		for root := range hostFormOwners[comparable] {
+			if root != groupRoot && groups[root].priority < group.priority {
+				candidateRoots[root] = struct{}{}
+			}
+		}
+		if short := NormalizeHostname(comparable); short != "" && short != comparable {
+			for root := range hostFormOwners[short] {
+				if root != groupRoot && groups[root].priority < group.priority {
+					candidateRoots[root] = struct{}{}
+				}
+			}
+		}
+	}
+
+	for _, host := range topLevelSystemSortedSet(group.exactHosts) {
+		for _, targetRoot := range topLevelSystemBetterRoots(hostOwners[host], groups, group.priority) {
+			if targetRoot == groupRoot {
+				continue
+			}
+			if groups[targetRoot].priority >= group.priority {
+				continue
+			}
+			if _, ok := targets[targetRoot]; !ok {
+				targets[targetRoot] = topLevelSystemAttachmentEvidence(
+					group,
+					groups[targetRoot],
+					"exact-host-attachment",
+					"exact-host",
+					host,
+				)
+			}
+		}
+	}
+	for _, targetRoot := range topLevelSystemSortedRoots(candidateRoots) {
+		if targetRoot == groupRoot {
+			continue
+		}
+		if groups[targetRoot].priority >= group.priority {
+			continue
+		}
+		if _, ok := targets[targetRoot]; ok {
+			continue
+		}
+		if value, ok := topLevelSystemShortFormHostMatchValue(group.exactHosts, groups[targetRoot].exactHosts); ok {
+			targets[targetRoot] = topLevelSystemAttachmentEvidence(
+				group,
+				groups[targetRoot],
+				"hostname-form-attachment",
+				"short-hostname",
+				value,
+			)
+		}
+	}
+	for _, ip := range topLevelSystemSortedSet(group.exactIPs) {
+		for _, targetRoot := range topLevelSystemBetterRoots(ipOwners[ip], groups, group.priority) {
+			if targetRoot == groupRoot {
+				continue
+			}
+			if groups[targetRoot].priority >= group.priority {
+				continue
+			}
+			if _, ok := targets[targetRoot]; !ok {
+				targets[targetRoot] = topLevelSystemAttachmentEvidence(
+					group,
+					groups[targetRoot],
+					"exact-ip-attachment",
+					"exact-ip",
+					ip,
+				)
+			}
+		}
+	}
+
+	if len(targets) != 1 {
+		return topLevelSystemFallbackTarget{}, false
+	}
+	for targetRoot, evidence := range targets {
+		return topLevelSystemFallbackTarget{
+			root:     targetRoot,
+			evidence: evidence,
+		}, true
+	}
+	return topLevelSystemFallbackTarget{}, false
+}
+
+func topLevelSystemSortedRoots(values map[int]struct{}) []int {
+	out := make([]int, 0, len(values))
+	for value := range values {
+		out = append(out, value)
+	}
+	sort.Ints(out)
+	return out
+}
+
+func topLevelSystemBetterRoots(values map[int]struct{}, groups map[int]topLevelSystemResolvedGroup, priority int) []int {
+	var out []int
+	for root := range values {
+		if groups[root].priority < priority {
+			out = append(out, root)
+		}
+	}
+	sort.Ints(out)
+	return out
+}

internal/unifiedresources/top_level_systems.go

@@ -143,15 +143,14 @@ func ResolveTopLevelSystems(resources []Resource) TopLevelSystemResolver {
 				bestPriority = group.priority
 			}
 		}
-		hostOwners, ipOwners := buildTopLevelSystemFallbackOwners(initialGroups)
-		hostFormOwners := buildTopLevelSystemHostFormOwners(initialGroups)
+		ownerIndex := buildTopLevelSystemFallbackIndex(initialGroups)
 		attached := false
 
 		for groupRoot, group := range initialGroups {
 			if !group.attachByHost || group.priority <= bestPriority {
 				continue
 			}
-			target, ok := uniqueBetterTopLevelSystemTarget(groupRoot, group, hostOwners, ipOwners, hostFormOwners, initialGroups)
+			target, ok := uniqueBetterTopLevelSystemTarget(groupRoot, group, ownerIndex, initialGroups)
 			if !ok {
 				continue
 			}
@@ -407,167 +406,146 @@ func topLevelSystemMatcherID(index int) string {
 	return "top-level-system:" + strconv.Itoa(index)
 }
 
-func buildTopLevelSystemFallbackOwners(
-	groups map[int]topLevelSystemResolvedGroup,
-) (map[string]map[int]struct{}, map[string]map[int]struct{}) {
-	hostOwners := make(map[string]map[int]struct{})
-	ipOwners := make(map[string]map[int]struct{})
-	for groupRoot, group := range groups {
-		for host := range group.exactHosts {
-			bucket := hostOwners[host]
-			if bucket == nil {
-				bucket = make(map[int]struct{})
-				hostOwners[host] = bucket
-			}
-			bucket[groupRoot] = struct{}{}
-		}
-		for ip := range group.exactIPs {
-			bucket := ipOwners[ip]
-			if bucket == nil {
-				bucket = make(map[int]struct{})
-				ipOwners[ip] = bucket
-			}
-			bucket[groupRoot] = struct{}{}
-		}
-	}
-	return hostOwners, ipOwners
+// A fallback pass sees an immutable group snapshot, even while union updates
+// the parents for the next pass. Order each owner bucket by priority once so
+// queries visit only better owners, without filtering and sorting it again.
+// The index is local to this pass: identity/metadata changes cannot reuse it.
+type topLevelSystemFallbackIndex struct {
+	hosts     map[string][]int
+	ips       map[string][]int
+	hostForms map[string][]int
 }
 
-// buildTopLevelSystemHostFormOwners indexes group roots by every hostname
-// form HostnamesEquivalent can match on (the comparable form and, when it
-// differs, the short form). Two hosts can only be equivalent when they share
-// at least one of these keys, so the index prunes the fallback's candidate
-// scan from every group to the groups that could possibly match. The full
-// pairwise matcher still decides each surviving candidate, so semantics are
-// unchanged.
-func buildTopLevelSystemHostFormOwners(
-	groups map[int]topLevelSystemResolvedGroup,
-) map[string]map[int]struct{} {
-	owners := make(map[string]map[int]struct{})
-	add := func(key string, root int) {
+func buildTopLevelSystemFallbackIndex(groups map[int]topLevelSystemResolvedGroup) topLevelSystemFallbackIndex {
+	index := topLevelSystemFallbackIndex{
+		hosts:     make(map[string][]int),
+		ips:       make(map[string][]int),
+		hostForms: make(map[string][]int),
+	}
+	roots := make([]int, 0, len(groups))
+	for root := range groups {
+		roots = append(roots, root)
+	}
+	sort.Slice(roots, func(i, j int) bool {
+		if groups[roots[i]].priority != groups[roots[j]].priority {
+			return groups[roots[i]].priority < groups[roots[j]].priority
+		}
+		return roots[i] < roots[j]
+	})
+	add := func(owners map[string][]int, key string, root int) {
 		if key == "" {
 			return
 		}
 		bucket := owners[key]
-		if bucket == nil {
-			bucket = make(map[int]struct{})
-			owners[key] = bucket
+		// All aliases of one root are added together, so a last-entry check
+		// deduplicates its comparable/short forms without a per-key set.
+		if len(bucket) == 0 || bucket[len(bucket)-1] != root {
+			owners[key] = append(bucket, root)
 		}
-		bucket[root] = struct{}{}
 	}
-	for groupRoot, group := range groups {
+	for _, root := range roots {
+		group := groups[root]
 		for host := range group.exactHosts {
+			add(index.hosts, host, root)
 			comparable := normalizeComparableHostname(host)
-			if comparable == "" {
-				continue
-			}
-			add(comparable, groupRoot)
-			if short := NormalizeHostname(comparable); short != "" && short != comparable {
-				add(short, groupRoot)
+			add(index.hostForms, comparable, root)
+			if comparable != "" {
+				add(index.hostForms, NormalizeHostname(comparable), root)
 			}
 		}
+		for ip := range group.exactIPs {
+			add(index.ips, ip, root)
+		}
 	}
-	return owners
+	return index
 }
 
 func uniqueBetterTopLevelSystemTarget(
 	groupRoot int,
 	group topLevelSystemResolvedGroup,
-	hostOwners map[string]map[int]struct{},
-	ipOwners map[string]map[int]struct{},
-	hostFormOwners map[string]map[int]struct{},
+	index topLevelSystemFallbackIndex,
 	groups map[int]topLevelSystemResolvedGroup,
 ) (topLevelSystemFallbackTarget, bool) {
-	targets := make(map[int]topLevelSystemGroupingEvidence)
-	candidateRoots := make(map[int]struct{})
-	for host := range group.exactHosts {
+	var target topLevelSystemFallbackTarget
+	found := false
+	hosts := topLevelSystemSortedSet(group.exactHosts)
+	for _, host := range hosts {
+		for _, targetRoot := range index.hosts[host] {
+			if groups[targetRoot].priority >= group.priority {
+				break
+			}
+			if targetRoot == groupRoot {
+				continue
+			}
+			if found && target.root != targetRoot {
+				// A second distinct qualifying root is ambiguity, regardless
+				// of evidence kind or priority. Never pick the first/best one.
+				return topLevelSystemFallbackTarget{}, false
+			}
+			if !found {
+				target = topLevelSystemFallbackTarget{root: targetRoot, evidence: topLevelSystemAttachmentEvidence(
+					group, groups[targetRoot], "exact-host-attachment", "exact-host", host,
+				)}
+				found = true
+			}
+		}
+	}
+	// The form index is only a candidate filter. Distinct FQDNs sharing a
+	// short name still need the full pairwise hostname-equivalence check.
+	seen := make(map[int]struct{})
+	for _, host := range hosts {
 		comparable := normalizeComparableHostname(host)
 		if comparable == "" {
 			continue
 		}
-		for root := range hostFormOwners[comparable] {
-			if root != groupRoot && groups[root].priority < group.priority {
-				candidateRoots[root] = struct{}{}
+		forms := [2]string{comparable, NormalizeHostname(comparable)}
+		for i, form := range forms {
+			if form == "" || (i == 1 && form == comparable) {
+				continue
 			}
-		}
-		if short := NormalizeHostname(comparable); short != "" && short != comparable {
-			for root := range hostFormOwners[short] {
-				if root != groupRoot && groups[root].priority < group.priority {
-					candidateRoots[root] = struct{}{}
+			for _, targetRoot := range index.hostForms[form] {
+				if groups[targetRoot].priority >= group.priority {
+					break
+				}
+				if targetRoot == groupRoot || (found && target.root == targetRoot) {
+					continue
+				}
+				if _, ok := seen[targetRoot]; ok {
+					continue
+				}
+				seen[targetRoot] = struct{}{}
+				if value, ok := topLevelSystemShortFormHostMatchValue(group.exactHosts, groups[targetRoot].exactHosts); ok {
+					if found {
+						return topLevelSystemFallbackTarget{}, false
+					}
+					target = topLevelSystemFallbackTarget{root: targetRoot, evidence: topLevelSystemAttachmentEvidence(
+						group, groups[targetRoot], "hostname-form-attachment", "short-hostname", value,
+					)}
+					found = true
 				}
 			}
 		}
 	}
-
-	for _, host := range topLevelSystemSortedSet(group.exactHosts) {
-		for _, targetRoot := range topLevelSystemBetterRoots(hostOwners[host], groups, group.priority) {
-			if targetRoot == groupRoot {
-				continue
-			}
-			if groups[targetRoot].priority >= group.priority {
-				continue
-			}
-			if _, ok := targets[targetRoot]; !ok {
-				targets[targetRoot] = topLevelSystemAttachmentEvidence(
-					group,
-					groups[targetRoot],
-					"exact-host-attachment",
-					"exact-host",
-					host,
-				)
-			}
-		}
-	}
-	for _, targetRoot := range topLevelSystemSortedRoots(candidateRoots) {
-		if targetRoot == groupRoot {
-			continue
-		}
-		if groups[targetRoot].priority >= group.priority {
-			continue
-		}
-		if _, ok := targets[targetRoot]; ok {
-			continue
-		}
-		if value, ok := topLevelSystemShortFormHostMatchValue(group.exactHosts, groups[targetRoot].exactHosts); ok {
-			targets[targetRoot] = topLevelSystemAttachmentEvidence(
-				group,
-				groups[targetRoot],
-				"hostname-form-attachment",
-				"short-hostname",
-				value,
-			)
-		}
-	}
 	for _, ip := range topLevelSystemSortedSet(group.exactIPs) {
-		for _, targetRoot := range topLevelSystemBetterRoots(ipOwners[ip], groups, group.priority) {
+		for _, targetRoot := range index.ips[ip] {
+			if groups[targetRoot].priority >= group.priority {
+				break
+			}
 			if targetRoot == groupRoot {
 				continue
 			}
-			if groups[targetRoot].priority >= group.priority {
-				continue
+			if found && target.root != targetRoot {
+				return topLevelSystemFallbackTarget{}, false
 			}
-			if _, ok := targets[targetRoot]; !ok {
-				targets[targetRoot] = topLevelSystemAttachmentEvidence(
-					group,
-					groups[targetRoot],
-					"exact-ip-attachment",
-					"exact-ip",
-					ip,
-				)
+			if !found {
+				target = topLevelSystemFallbackTarget{root: targetRoot, evidence: topLevelSystemAttachmentEvidence(
+					group, groups[targetRoot], "exact-ip-attachment", "exact-ip", ip,
+				)}
+				found = true
 			}
 		}
 	}
-
-	if len(targets) != 1 {
-		return topLevelSystemFallbackTarget{}, false
-	}
-	for targetRoot, evidence := range targets {
-		return topLevelSystemFallbackTarget{
-			root:     targetRoot,
-			evidence: evidence,
-		}, true
-	}
-	return topLevelSystemFallbackTarget{}, false
+	return target, found
 }
 
 func monitoredSystemResourceAllowsHostAttachment(resource *Resource) bool {
@@ -961,27 +939,8 @@ func topLevelSystemHumanList(values []string) string {
 	}
 }
 
-func topLevelSystemSortedRoots(values map[int]struct{}) []int {
-	out := make([]int, 0, len(values))
-	for value := range values {
-		out = append(out, value)
-	}
-	sort.Ints(out)
-	return out
-}
-
 // Discard ineligible equal/worse-priority peers before allocating and sorting
 // candidate lists. They could never win the existing attachment rule.
-func topLevelSystemBetterRoots(values map[int]struct{}, groups map[int]topLevelSystemResolvedGroup, priority int) []int {
-	var out []int
-	for root := range values {
-		if groups[root].priority < priority {
-			out = append(out, root)
-		}
-	}
-	sort.Ints(out)
-	return out
-}
 
 func topLevelSystemMaxInt(left, right int) int {
 	if left > right {

internal/unifiedresources/top_level_systems_test.go

@@ -435,3 +435,29 @@ func topLevelTestK8sCluster(id, clusterID, agentID, server string) Resource {
 		},
 	}
 }
+
+func TestResolveTopLevelSystemsRebuildsOwnerIndexAtEveryPassAndCall(t *testing.T) {
+	bridge := topLevelTestAgent("bridge", "alpha.lab", "bridge-machine", "bridge-agent")
+	bridge.Identity.IPAddresses = []string{"192.0.2.8"}
+	leaf := topLevelTestDockerHost("leaf", "beta.lab", "leaf-runtime", "")
+	leaf.Identity.IPAddresses = []string{"192.0.2.8"}
+	resources := []Resource{
+		topLevelTestProxmoxNode("platform", "alpha.lab", "pve-source", "https://alpha.lab:8006"),
+		bridge, leaf,
+		topLevelTestK8sCluster("cluster", "cluster-id", "", "https://alpha.lab:6443"),
+	}
+	// Docker has better priority than the unmerged agent, so it cannot attach
+	// by their shared IP in the first pass. Once the agent joins the platform,
+	// a fresh pass must admit the IP attachment while keeping the cluster apart.
+	resolver := ResolveTopLevelSystems(resources)
+	if resolver.Count() != 2 {
+		t.Fatalf("stale pass index or excluded cluster attachment: %d groups", resolver.Count())
+	}
+	assertTopLevelSystemGroupPairs(t, resolver, [][2]string{{"platform", "bridge"}, {"platform", "leaf"}}, [][2]string{{"platform", "cluster"}})
+	resources[2].Identity.IPAddresses = []string{"192.0.2.9"}
+	next := ResolveTopLevelSystems(resources)
+	if next.Count() != 3 {
+		t.Fatalf("index reused across identity changes: %d groups", next.Count())
+	}
+	assertTopLevelSystemGroupPairs(t, next, [][2]string{{"platform", "bridge"}}, [][2]string{{"platform", "leaf"}, {"platform", "cluster"}})
+}