From 54de7bdca738ae9d69b1856befcd4d4a60fbfe5d Mon Sep 17 00:00:00 2001
From: "pulse-triage[bot]"
 <249995291+pulse-triage[bot]@users.noreply.github.com>
Date: Mon, 5 Oct 2026 11:22:07 +0100
Subject: [PATCH 1/5] Capture fresh broadcast resources and history targets
 together

Resolve canonical source mappings once per complete projection while preserving detached resource content and target selection after host coalescing. Bind resources and targets to the same locked generation, including absence and continuity overlays, and avoid another owned outer-slice copy. No timestamp cache or native CPU/RSS claim. Retain cold, clean and dirty full-content controls and paired complete-projection measurements.

Contract-Neutral: Internal bulk-read and copy-ownership repair preserves existing public resource/history fields, source precedence, agent identity/lifecycle, tenant/revocation and safety contracts; no wire or permission delta.
Change-source: pulse-maintainer
---
 .../v6/internal/subsystems/monitoring.md      |  24 ++
 .../internal/subsystems/unified-resources.md  |  23 ++
 .../broadcast_metrics_snapshot_test.go        | 264 ++++++++++++++++++
 .../monitoring/canonical_guardrails_test.go   |   7 +-
 internal/monitoring/monitor.go                |  83 ++++--
 .../monitoring/monitor_host_agents_test.go    |   4 +
 internal/unifiedresources/monitor_adapter.go  |  12 +
 .../monitor_adapter_read_state_test.go        |  37 +++
 internal/unifiedresources/registry.go         |  31 +-
 .../registry_projection_test.go               |  99 +++++++
 internal/unifiedresources/registry_test.go    |  65 +++++
 11 files changed, 623 insertions(+), 26 deletions(-)
 create mode 100644 internal/monitoring/broadcast_metrics_snapshot_test.go
 create mode 100644 internal/unifiedresources/registry_projection_test.go

diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md
index 354ad8d5e2..fd23f74cc1 100644
--- a/docs/release-control/v6/internal/subsystems/monitoring.md
+++ b/docs/release-control/v6/internal/subsystems/monitoring.md
@@ -17,6 +17,30 @@
 
 ## Purpose
 
+### Fresh broadcast projection owns one resource/target capture
+
+Canonical live, fallback and standalone-host continuity broadcast reads prefer
+the adapter's detached bulk resource/metrics-target capture. After host
+coalescing selects presentation identity, missing history targets are attached
+from that same capture, including authoritative absence. Stores without this
+optional capability retain the previous point-resolver fallback. Only the owned
+outer coalesced slice is decorated in place; nested source payloads stay read-only.
+
+Every read still observes complete current resource data and derives live alert
+health, metadata URL edits/clears and connected/ignored infrastructure. No
+timestamp-only cache, telemetry omission, policy change or new wire field is
+introduced. All existing catalogs, discovery/history coordinates, original
+observation times, LastSeen, identity and tenant/revocation boundaries remain.
+
+`broadcast_metrics_snapshot_test.go` compares the complete served JSON with an
+independent earlier projection across the metadata-heavy nine-provider demo,
+typed-view and same-time dirty cases; it also covers captured target replacement
+and continuity/metadata/live-alert boundaries. `monitor_host_agents_test.go`
+retains its one-materialization and live-change regression. Benchmarks include
+the exact parent's complete conversion and final JSON encoding, with cold,
+list-only, typed-clean and dirty inputs; they do not establish installed CPU/RSS
+relief, native identity refresh throughput or completion of #2199.
+
 ### Report identity lookups do not rebuild unrelated Kubernetes metadata
 
 The producer's two temporary host/Docker-host admission registries use a
diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md
index 829cedf05e..1bd25916bf 100644
--- a/docs/release-control/v6/internal/subsystems/unified-resources.md
+++ b/docs/release-control/v6/internal/subsystems/unified-resources.md
@@ -23,6 +23,29 @@ and sort the complete canonical change table while startup and ingestion wait.
 
 ## Purpose
 
+### Broadcast bulk reads capture resources and history coordinates together
+
+`ListWithMetricsTargets` and the monitor adapter's matching optional read return
+detached resources and a detached target map from one locked registry generation.
+The existing source-mapping inversion runs once for this read when typed views
+are dirty; it does not build all typed views or add a retained generation cache.
+Clean typed views may supply their already-current inverse mapping. Legacy List,
+GetAll and individual target reads preserve their shapes and semantics.
+
+The target map stays separate from resources until presentation host coalescing
+selects the row identity. Missing coordinates remain missing in that capture;
+they never fall through to a later live registry. Mapping changes with unchanged
+LastSeen take effect on the next read. Returned resources, policy/identity blocks
+and target values remain caller-owned across mutations and adapter replacement.
+Projection cannot pair an earlier resource row with a later history identity.
+
+`registry_test.go`, `registry_projection_test.go` and
+`monitor_adapter_read_state_test.go` exercise full resource/point-target parity,
+same-time edits, concurrent mapping writes, detachments, empty/nil reads and
+tenant separation. Connected monitoring proof covers all nine demo providers,
+continuity overlays and complete served JSON. Paired cold/clean/dirty benchmarks
+measure whole materialization, not a new point-lookup envelope or native relief.
+
 Canonical resource History honours selected memory observation provenance and Proxmox VM filesystem deferrals instead of treating numeric carriers as fresh measurements.
 
 ### Canonical resource History cannot promote retained guest readings
diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go
new file mode 100644
index 0000000000..4e61cc2f3d
--- /dev/null
+++ b/internal/monitoring/broadcast_metrics_snapshot_test.go
@@ -0,0 +1,264 @@
+package monitoring
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/config"
+	"github.com/rcourtman/pulse-go-rewrite/internal/mock"
+	"github.com/rcourtman/pulse-go-rewrite/internal/models"
+	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+)
+
+func projectionFixtureGraph(t testing.TB) mock.FixtureGraph {
+	t.Helper()
+	previousEnabled, previousConfig := mock.IsMockEnabled(), mock.GetConfig()
+	t.Cleanup(func() {
+		mustSetMockEnabled(t, false)
+		mock.SetMockConfig(previousConfig)
+		mustSetMockEnabled(t, previousEnabled)
+	})
+	mustSetMockEnabled(t, false)
+	cfg := mock.DefaultConfig
+	cfg.UpdateInterval = 5 * time.Minute
+	mock.SetMockConfig(cfg)
+	mustSetMockEnabled(t, true)
+	graph := mock.CurrentFixtureGraph()
+	mustSetMockEnabled(t, false)
+	return graph
+}
+
+func projectionFixtureRegistry(graph mock.FixtureGraph) *unifiedresources.ResourceRegistry {
+	rr := unifiedresources.NewRegistry(nil)
+	rr.IngestSnapshot(graph.State)
+	for _, source := range []unifiedresources.DataSource{unifiedresources.SourceTrueNAS, unifiedresources.SourceVMware, unifiedresources.SourceAvailability} {
+		rr.IngestRecords(source, graph.SupplementalRecords(source))
+	}
+	return rr
+}
+
+// The original full projection, including target resolution after host
+// coalescing. Compare all wire fields/catalogues, not selected metric values.
+func legacyFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend {
+	rows := unifiedresources.CoalescePresentationHostResources(resources)
+	rows = m.applyPersistedMetadataToUnifiedResources(rows)
+	rows = unifiedresources.AttachResourceHealth(rows, resourceHealthAlerts(snapshot.ActiveAlerts), now)
+	projected, catalogs := convertResourcesForBroadcastReference(rows, resolver)
+	out := snapshot.ToFrontend()
+	out.Resources = projected
+	out.CapabilityCatalog = catalogs.capabilities
+	out.PolicyCatalog = catalogs.policies
+	out.AISafeSummaryCatalog = catalogs.aiSafeSummaries
+	out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot)
+	return out
+}
+
+func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) {
+	graph := projectionFixtureGraph(t)
+	rr := projectionFixtureRegistry(graph)
+	adapter := unifiedresources.NewMonitorAdapter(rr)
+	store := &broadcastProjectionCountingStore{MonitorAdapter: adapter}
+	m := &Monitor{resourceStore: store}
+	resources := adapter.GetAll()
+	if len(resources) < 1000 {
+		t.Fatalf("demo fixture too small: %d", len(resources))
+	}
+	sources := map[unifiedresources.DataSource]bool{}
+	for _, r := range resources {
+		for _, source := range r.Sources {
+			sources[source] = true
+		}
+	}
+	for _, source := range []unifiedresources.DataSource{unifiedresources.SourceProxmox, unifiedresources.SourceAgent, unifiedresources.SourceDocker, unifiedresources.SourcePBS, unifiedresources.SourcePMG, unifiedresources.SourceK8s, unifiedresources.SourceTrueNAS, unifiedresources.SourceVMware, unifiedresources.SourceAvailability} {
+		if !sources[source] {
+			t.Fatalf("demo omits provider %s", source)
+		}
+	}
+	assertEqual := func(stage string) {
+		t.Helper()
+		before, err := json.Marshal(adapter.GetAll())
+		if err != nil {
+			t.Fatal(err)
+		}
+		want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC())
+		got := m.buildBroadcastFrontendStateFromSnapshot(graph.State)
+		wantJSON, err := json.Marshal(want)
+		if err != nil {
+			t.Fatal(err)
+		}
+		gotJSON, err := json.Marshal(got)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(wantJSON) != string(gotJSON) {
+			t.Fatalf("%s: complete frontend/catalogue/infrastructure JSON differs", stage)
+		}
+		after, _ := json.Marshal(adapter.GetAll())
+		if string(before) != string(after) {
+			t.Fatalf("%s: projection changed source data", stage)
+		}
+		t.Logf("%s: %d complete resources, %d frontend rows, %d wire bytes equal legacy projection", stage, len(adapter.GetAll()), len(got.Resources), len(gotJSON))
+	}
+	assertEqual("cold-list-only")
+	rr.Workloads()
+	assertEqual("typed-views-clean")
+	// Same timestamps, new payload, policy and URL. No time-only invalidation.
+	resources = adapter.GetAll()
+	resources[0].Name = "changed-resource"
+	resources[0].Tags = []string{"customer-data"}
+	resources[0].CustomURL = "https://example.invalid/changed"
+	rr.IngestResources(resources)
+	assertEqual("dirty-same-time")
+}
+
+func TestBroadcastMetricsSnapshotOwnsTargetsBeforeLiveReplacement(t *testing.T) {
+	m, adapter, _ := newReadStateCloneTestMonitor(t, 4)
+	view := m.currentUnifiedStateView()
+	if view.metricsTargets == nil {
+		t.Fatal("broadcast did not capture a bulk resolver")
+	}
+	old := view.resources[0]
+	want := adapter.MetricsTargetForResource(old.ID)
+	adapter.PopulateFromSnapshot(models.EmptyStateSnapshot())
+	if got := view.metricsTargets.MetricsTargetForResource(old.ID); !reflect.DeepEqual(got, want) {
+		t.Fatal("captured resources were paired with a later generation's target")
+	}
+	if got := view.metricsTargets.MetricsTargetForResource("unknown"); got != nil {
+		t.Fatal("unknown target fell through to live data")
+	}
+	copyTarget := view.metricsTargets.MetricsTargetForResource(old.ID)
+	copyTarget.ResourceID = "caller-edit"
+	if got := view.metricsTargets.MetricsTargetForResource(old.ID); got.ResourceID == "caller-edit" {
+		t.Fatal("caller mutated the captured resolver")
+	}
+}
+
+func TestBroadcastMetricsSnapshotContinuityAndLiveOverlays(t *testing.T) {
+	m, adapter, _ := newReadStateCloneTestMonitor(t, 2)
+	now := time.Now().UTC()
+	m.hostContinuityStore = config.NewHostContinuityStore(t.TempDir(), nil)
+	entry := config.HostContinuityEntry{HostID: "absent-host", Hostname: "absent.example", MachineID: "absent-machine", LastSeen: now.Add(-time.Hour)}
+	if err := m.hostContinuityStore.Upsert(entry); err != nil {
+		t.Fatal(err)
+	}
+	view := m.unifiedStateViewWithStandaloneHostContinuity(monitorUnifiedStateView{readState: adapter})
+	if len(view.resources) != 3 || view.metricsTargets == nil {
+		t.Fatal("continuity projection omitted its resources or captured resolver")
+	}
+	var host unifiedresources.Resource
+	for _, r := range view.resources {
+		if r.Agent != nil && r.Agent.AgentID == entry.HostID {
+			host = r
+		}
+		if got, want := view.metricsTargets.MetricsTargetForResource(r.ID), broadcastMetricsTargetResolver(view.readState).MetricsTargetForResource(r.ID); !reflect.DeepEqual(got, want) {
+			t.Fatal("overlay target differs from overlay registry")
+		}
+	}
+	if host.ID == "" || host.Status == unifiedresources.StatusOnline || !host.LastSeen.Equal(entry.LastSeen) {
+		t.Fatal("continuity history was omitted or promoted to current")
+	}
+	m.hostMetadataStore = config.NewHostMetadataStore(t.TempDir(), nil)
+	for _, url := range []string{"https://example.invalid/operator", ""} {
+		if err := m.hostMetadataStore.Set(entry.HostID, &config.HostMetadata{CustomURL: url}); err != nil {
+			t.Fatal(err)
+		}
+		decorated := m.applyPersistedMetadataToUnifiedResources(view.resources)
+		for _, r := range decorated {
+			if r.ID == host.ID && r.CustomURL != url {
+				t.Fatal("operator URL change/clear was hidden by captured data")
+			}
+		}
+	}
+	critical := unifiedresources.EvaluateResourceHealth(host, []unifiedresources.ResourceHealthAlert{{ResourceID: host.ID, Level: "critical", Type: "cpu"}}, now)
+	if critical.Verdict != unifiedresources.HealthCritical {
+		t.Fatal("live alert was hidden by captured data")
+	}
+	if len(adapter.GetAll()) != 2 {
+		t.Fatal("continuity overlay wrote back to the live registry")
+	}
+}
+
+var completeProjectionSink models.StateFrontend
+var completeProjectionJSONSink []byte
+
+// Exact parent broadcast shape (one coalesce, copy-to-attach, then conversion),
+// not the older two-coalesce content oracle above. The unchanged conversion,
+// metadata/health decoration and snapshot projection are part of both costs.
+func parentFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend {
+	rows := unifiedresources.CoalescePresentationHostResources(resources)
+	for i := range rows {
+		m.applyPersistedMetadataToUnifiedResource(&rows[i])
+		health := unifiedresources.EvaluateResourceHealth(rows[i], resourceHealthAlerts(snapshot.ActiveAlerts), now)
+		rows[i].Health = &health
+	}
+	projected, catalogs := convertPresentationResourcesForBroadcast(attachBroadcastMetricsTargets(rows, resolver))
+	out := snapshot.ToFrontend()
+	out.Resources = projected
+	out.CapabilityCatalog = catalogs.capabilities
+	out.PolicyCatalog = catalogs.policies
+	out.AISafeSummaryCatalog = catalogs.aiSafeSummaries
+	out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot)
+	return out
+}
+
+func BenchmarkBroadcastMetricsSnapshot(b *testing.B) {
+	graph := projectionFixtureGraph(b)
+	for _, mode := range []string{"cold", "list-clean", "views-clean", "dirty"} {
+		for _, encode := range []bool{false, true} {
+			stage := "projection"
+			if encode {
+				stage = "json-frame"
+			}
+			for _, bulk := range []bool{false, true} {
+				path := "point-targets"
+				if bulk {
+					path = "bulk-targets"
+				}
+				b.Run(fmt.Sprintf("%s/%s/%s", mode, stage, path), func(b *testing.B) {
+					rr := projectionFixtureRegistry(graph)
+					adapter := unifiedresources.NewMonitorAdapter(rr)
+					store := &broadcastProjectionCountingStore{MonitorAdapter: adapter}
+					m := &Monitor{resourceStore: store}
+					if mode != "cold" {
+						rr.List()
+					}
+					if mode == "views-clean" {
+						rr.Workloads()
+					}
+					b.ReportAllocs()
+					b.ResetTimer()
+					for i := 0; i < b.N; i++ {
+						if mode == "cold" || mode == "dirty" {
+							b.StopTimer()
+							if mode == "cold" {
+								rr = projectionFixtureRegistry(graph)
+								adapter = unifiedresources.NewMonitorAdapter(rr)
+								store.MonitorAdapter = adapter
+							} else {
+								changed := rr.List()[0]
+								changed.Tags = []string{fmt.Sprintf("edit-%d", i)}
+								rr.IngestResources([]unifiedresources.Resource{changed})
+							}
+							b.StartTimer()
+						}
+						if bulk {
+							completeProjectionSink = m.buildBroadcastFrontendStateFromSnapshot(graph.State)
+						} else {
+							completeProjectionSink = parentFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC())
+						}
+						if encode {
+							var err error
+							completeProjectionJSONSink, err = json.Marshal(completeProjectionSink)
+							if err != nil {
+								b.Fatal(err)
+							}
+						}
+					}
+				})
+			}
+		}
+	}
+}
diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go
index 165f94a540..9e11cb4f7e 100644
--- a/internal/monitoring/canonical_guardrails_test.go
+++ b/internal/monitoring/canonical_guardrails_test.go
@@ -607,10 +607,11 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) {
 	source := string(data)
 
 	for _, snippet := range []string{
-		"metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)",
+		"metricsTargetResolver := unifiedView.metricsTargets",
+		"metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)",
 		"broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)",
-		"broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(",
-		"attachBroadcastMetricsTargets(broadcastResources, metricsTargetResolver)",
+		"broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources)",
+		"attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver)",
 		"frontendState.CapabilityCatalog = broadcastCatalogs.capabilities",
 		"frontendState.PolicyCatalog = broadcastCatalogs.policies",
 		"frontendState.AISafeSummaryCatalog = broadcastCatalogs.aiSafeSummaries",
diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go
index 6bab63084c..162d189642 100644
--- a/internal/monitoring/monitor.go
+++ b/internal/monitoring/monitor.go
@@ -4341,7 +4341,10 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS
 		}
 	}
 	unifiedView := m.currentUnifiedStateView()
-	metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)
+	metricsTargetResolver := unifiedView.metricsTargets
+	if metricsTargetResolver == nil {
+		metricsTargetResolver = broadcastMetricsTargetResolver(unifiedView.readState)
+	}
 	broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)
 	// Coalescing owns the outer slice. Decorate that one projection in place,
 	// not three full-resource copies; nested store data is still read-only.
@@ -4352,9 +4355,8 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS
 		health := unifiedresources.EvaluateResourceHealth(broadcastResources[i], healthAlerts, now)
 		broadcastResources[i].Health = &health
 	}
-	broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(
-		attachBroadcastMetricsTargets(broadcastResources, metricsTargetResolver),
-	)
+	attachBroadcastMetricsTargetsInPlace(broadcastResources, metricsTargetResolver)
+	broadcastFrontendResources, broadcastCatalogs := convertPresentationResourcesForBroadcast(broadcastResources)
 	frontendState.Resources = broadcastFrontendResources
 	frontendState.CapabilityCatalog = broadcastCatalogs.capabilities
 	frontendState.PolicyCatalog = broadcastCatalogs.policies
@@ -4871,23 +4873,50 @@ func (m *Monitor) RecentResourceChanges(resourceID string, since time.Time, limi
 }
 
 type monitorUnifiedStateView struct {
-	resources []unifiedresources.Resource
-	readState unifiedresources.ReadState
-	freshness time.Time
+	resources      []unifiedresources.Resource
+	readState      unifiedresources.ReadState
+	freshness      time.Time
+	metricsTargets MetricsTargetResourceStore
 }
 
 type unifiedResourceReadStateLister interface {
 	GetAll() []unifiedresources.Resource
 }
 
+type unifiedResourceProjectionLister interface {
+	GetAllWithMetricsTargets() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget)
+}
+
+// This resolver never falls through to the live store, including for an
+// unknown ID. A missing target belongs to this captured generation too.
+type projectionMetricsTargets map[string]unifiedresources.MetricsTarget
+
+func (targets projectionMetricsTargets) MetricsTargetForResource(id string) *unifiedresources.MetricsTarget {
+	target, ok := targets[unifiedresources.CanonicalResourceID(id)]
+	if !ok {
+		return nil
+	}
+	return &target
+}
+
+func unifiedProjectionResources(lister unifiedResourceReadStateLister) ([]unifiedresources.Resource, MetricsTargetResourceStore) {
+	if projection, ok := lister.(unifiedResourceProjectionLister); ok {
+		resources, targets := projection.GetAllWithMetricsTargets()
+		return resources, projectionMetricsTargets(targets)
+	}
+	return lister.GetAll(), nil
+}
+
 func monitorUnifiedStateViewFromSnapshot(snapshot models.StateSnapshot) monitorUnifiedStateView {
 	registry := unifiedresources.NewRegistry(nil)
 	registry.IngestSnapshot(snapshot)
 	adapter := unifiedresources.NewMonitorAdapter(registry)
+	resources, targets := unifiedProjectionResources(adapter)
 	return monitorUnifiedStateView{
-		resources: registry.List(),
-		readState: adapter,
-		freshness: snapshot.LastUpdate,
+		resources:      resources,
+		readState:      adapter,
+		freshness:      snapshot.LastUpdate,
+		metricsTargets: targets,
 	}
 }
 
@@ -4895,10 +4924,12 @@ func monitorUnifiedStateViewFromResources(resources []unifiedresources.Resource,
 	registry := unifiedresources.NewRegistry(nil)
 	registry.IngestResources(resources)
 	adapter := unifiedresources.NewMonitorAdapter(registry)
+	resources, targets := unifiedProjectionResources(adapter)
 	return monitorUnifiedStateView{
-		resources: registry.List(),
-		readState: adapter,
-		freshness: freshness,
+		resources:      resources,
+		readState:      adapter,
+		freshness:      freshness,
+		metricsTargets: targets,
 	}
 }
 
@@ -4915,8 +4946,9 @@ func (m *Monitor) unifiedStateViewWithStandaloneHostContinuity(view monitorUnifi
 		return view
 	}
 
-	resources := lister.GetAll()
+	resources, targets := unifiedProjectionResources(lister)
 	view.resources = resources
+	view.metricsTargets = targets
 	if view.freshness.IsZero() {
 		view.freshness = latestUnifiedResourceLastSeen(resources)
 	}
@@ -6479,17 +6511,24 @@ func attachBroadcastMetricsTargets(
 	}
 
 	out := make([]unifiedresources.Resource, len(resources))
-	for i, resource := range resources {
-		if resource.MetricsTarget == nil {
-			resource.MetricsTarget = validBroadcastMetricsTarget(
-				resolver.MetricsTargetForResource(resource.ID),
-			)
-		}
-		out[i] = resource
-	}
+	copy(out, resources)
+	attachBroadcastMetricsTargetsInPlace(out, resolver)
 	return out
 }
 
+// Only call with the owned outer slice of a coalesced broadcast projection.
+// Nested resource data stays read-only, and the resolver returns owned targets.
+func attachBroadcastMetricsTargetsInPlace(resources []unifiedresources.Resource, resolver MetricsTargetResourceStore) {
+	if resolver == nil {
+		return
+	}
+	for i := range resources {
+		if resources[i].MetricsTarget == nil {
+			resources[i].MetricsTarget = validBroadcastMetricsTarget(resolver.MetricsTargetForResource(resources[i].ID))
+		}
+	}
+}
+
 func validBroadcastMetricsTarget(
 	target *unifiedresources.MetricsTarget,
 ) *unifiedresources.MetricsTarget {
diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go
index 12e934c142..695d475395 100644
--- a/internal/monitoring/monitor_host_agents_test.go
+++ b/internal/monitoring/monitor_host_agents_test.go
@@ -6718,6 +6718,10 @@ func (s *broadcastProjectionCountingStore) GetAll() []unifiedresources.Resource
 	s.reads++
 	return s.MonitorAdapter.GetAll()
 }
+func (s *broadcastProjectionCountingStore) GetAllWithMetricsTargets() ([]unifiedresources.Resource, map[string]unifiedresources.MetricsTarget) {
+	s.reads++
+	return s.MonitorAdapter.GetAllWithMetricsTargets()
+}
 func (*broadcastProjectionCountingStore) TryReplaceRegistryForRead(models.StateSnapshot, time.Duration, func() map[unifiedresources.DataSource][]unifiedresources.IngestRecord) bool {
 	return false
 }
diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go
index 28eda9d110..8bac41ceaa 100644
--- a/internal/unifiedresources/monitor_adapter.go
+++ b/internal/unifiedresources/monitor_adapter.go
@@ -496,6 +496,18 @@ func (a *MonitorAdapter) GetAll() []Resource {
 	return registry.List()
 }
 
+// GetAllWithMetricsTargets materializes one complete broadcast read, binding
+// its detached target map to the same registry as its detached resource list.
+// The adapter may publish a replacement immediately after this capture; neither
+// half of the returned read then consults that newer generation.
+func (a *MonitorAdapter) GetAllWithMetricsTargets() ([]Resource, map[string]MetricsTarget) {
+	registry := a.currentRegistry()
+	if registry == nil {
+		return nil, nil
+	}
+	return registry.ListWithMetricsTargets()
+}
+
 func (a *MonitorAdapter) unifiedAIAdapter() *UnifiedAIAdapter {
 	registry := a.currentRegistry()
 	if registry == nil {
diff --git a/internal/unifiedresources/monitor_adapter_read_state_test.go b/internal/unifiedresources/monitor_adapter_read_state_test.go
index 5eb791f3cf..7eefcb5b99 100644
--- a/internal/unifiedresources/monitor_adapter_read_state_test.go
+++ b/internal/unifiedresources/monitor_adapter_read_state_test.go
@@ -971,3 +971,40 @@ func TestHostContinuityOverlayReusedWithinRegistryGeneration(t *testing.T) {
 		t.Fatal("overlay was reused past its maximum age")
 	}
 }
+
+func TestMonitorAdapterProjectionSnapshotKeepsReplacedGeneration(t *testing.T) {
+	var absent *MonitorAdapter
+	rows, targets := absent.GetAllWithMetricsTargets()
+	if rows != nil || targets != nil {
+		t.Fatal("nil adapter invented a projection")
+	}
+	adapter := NewMonitorAdapter(NewRegistry(nil))
+	snapshot := readRefreshSnapshot("before")
+	adapter.PopulateFromSnapshot(snapshot)
+	first, firstTargets := adapter.GetAllWithMetricsTargets()
+	snapshot.VMs[0].Name = "after"
+	snapshot.VMs[0].ID = "new-node:201"
+	snapshot.VMs[0].VMID = 201
+	// Observation time is deliberately unchanged.
+	adapter.PopulateFromSnapshot(snapshot)
+	second, secondTargets := adapter.GetAllWithMetricsTargets()
+	if len(first) != 1 || len(second) != 1 || first[0].Name != "before" || second[0].Name != "after" {
+		t.Fatal("replacement lost complete fresh resource content")
+	}
+	if firstTargets[first[0].ID].ResourceID == secondTargets[second[0].ID].ResourceID {
+		t.Fatal("replacement retained old history coordinates")
+	}
+	for _, r := range second {
+		if want := adapter.MetricsTargetForResource(r.ID); want == nil || *want != secondTargets[r.ID] {
+			t.Fatal("target differs from the generation's point resolver")
+		}
+	}
+	other := NewMonitorAdapter(NewRegistry(nil))
+	snapshot.VMs[0].ID = "other-tenant:501"
+	snapshot.VMs[0].Name = "other-tenant"
+	other.PopulateFromSnapshot(snapshot)
+	otherRows, otherTargets := other.GetAllWithMetricsTargets()
+	if reflect.DeepEqual(otherRows, second) || reflect.DeepEqual(otherTargets, secondTargets) {
+		t.Fatal("separate adapters shared a tenant projection")
+	}
+}
diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go
index 774f2d8930..34992538c1 100644
--- a/internal/unifiedresources/registry.go
+++ b/internal/unifiedresources/registry.go
@@ -1321,6 +1321,21 @@ func seededVMwareSourceID(resource *Resource) string {
 // the read lock and copy it without repeating identity/scope/policy derivation.
 // The typed views remain lazy, so a List-only consumer pays for no extra copy.
 func (rr *ResourceRegistry) List() []Resource {
+	resources, _ := rr.listMaterialized(false)
+	return resources
+}
+
+// ListWithMetricsTargets captures detached resources and their history targets
+// from the same registry generation. Resolve all source mappings in one pass,
+// without forcing the unrelated typed views to be built. In particular a
+// freshly ingested registry must not scan every mapping once per broadcast row.
+// Targets are separate: host coalescing must select the presentation identity
+// before a missing target is attached, just as with individual target lookups.
+func (rr *ResourceRegistry) ListWithMetricsTargets() ([]Resource, map[string]MetricsTarget) {
+	return rr.listMaterialized(true)
+}
+
+func (rr *ResourceRegistry) listMaterialized(withTargets bool) ([]Resource, map[string]MetricsTarget) {
 	for {
 		rr.mu.RLock()
 		if rr.canonicalMetadataDirty {
@@ -1332,13 +1347,27 @@ func (rr *ResourceRegistry) List() []Resource {
 			// rather than copy a generation that is no longer materialized.
 			continue
 		}
+		var sourceTargets map[string][]SourceTarget
+		var targets map[string]MetricsTarget
+		if withTargets {
+			sourceTargets = rr.cachedSourceTargets
+			if rr.viewsDirty || sourceTargets == nil {
+				sourceTargets = rr.buildSourceTargetsIndexLocked()
+			}
+			targets = make(map[string]MetricsTarget, len(rr.resources))
+		}
 		out := make([]Resource, 0, len(rr.resources))
 		for _, r := range rr.resources {
 			out = append(out, cloneMaterializedResource(r))
+			if withTargets {
+				if target := rr.metricsTargetFromSourceTargets(r, sourceTargets[r.ID]); target != nil {
+					targets[r.ID] = *target
+				}
+			}
 		}
 		rr.mu.RUnlock()
 		sortResourcesByName(out)
-		return out
+		return out, targets
 	}
 }
 
diff --git a/internal/unifiedresources/registry_projection_test.go b/internal/unifiedresources/registry_projection_test.go
new file mode 100644
index 0000000000..ca9dda8383
--- /dev/null
+++ b/internal/unifiedresources/registry_projection_test.go
@@ -0,0 +1,99 @@
+package unifiedresources
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+)
+
+// Exercise the capture while a writer changes the mapping and row together.
+// Separate List/point reads can legitimately observe different generations;
+// this paired bulk read cannot. The test never touches persistence or drivers.
+func TestRegistryProjectionSnapshotConcurrentMappingEdits(t *testing.T) {
+	rr := NewRegistry(nil)
+	rr.IngestSnapshot(benchmarkVMState(1))
+	id := rr.List()[0].ID
+	set := func(version int) {
+		rr.mu.Lock()
+		defer rr.mu.Unlock()
+		name := fmt.Sprintf("coordinate-%d", version)
+		rr.resources[id].Name = name
+		rr.bySource[SourceProxmox] = map[string]string{name: id}
+		rr.invalidateViewsLocked()
+	}
+	set(0)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 200; i++ {
+			set(i % 2)
+		}
+	}()
+	defer wg.Wait()
+	for i := 0; i < 200; i++ {
+		rows, targets := rr.ListWithMetricsTargets()
+		if len(rows) != 1 || rows[0].Name != targets[id].ResourceID {
+			t.Fatalf("mixed row/target generation: row=%q target=%q", rows[0].Name, targets[id].ResourceID)
+		}
+	}
+}
+
+var projectionResourceSink []Resource
+var projectionTargetSink map[string]MetricsTarget
+
+// Include the entire owned List and every history coordinate. These are not
+// point-lookup timings, nor a replacement for the old clone-allocation envelope.
+func BenchmarkRegistryProjectionSnapshot(b *testing.B) {
+	for _, count := range []int{0, 1, 1000} {
+		for _, mode := range []string{"cold", "list-clean", "views-clean", "dirty"} {
+			for _, bulk := range []bool{false, true} {
+				name := "point-targets"
+				if bulk {
+					name = "bulk-targets"
+				}
+				b.Run(fmt.Sprintf("%d/%s/%s", count, mode, name), func(b *testing.B) {
+					state := benchmarkVMState(count)
+					rr := NewRegistry(nil)
+					rr.IngestSnapshot(state)
+					if mode != "cold" {
+						rr.List()
+					}
+					if mode == "views-clean" {
+						rr.VMs()
+					}
+					b.ReportAllocs()
+					b.ResetTimer()
+					for i := 0; i < b.N; i++ {
+						if mode == "cold" {
+							b.StopTimer()
+							rr = NewRegistry(nil)
+							rr.IngestSnapshot(state)
+							b.StartTimer()
+						}
+						if mode == "dirty" {
+							rr.mu.Lock()
+							for _, r := range rr.resources {
+								r.Tags = []string{fmt.Sprintf("version-%d", i)}
+								break
+							}
+							rr.invalidateViewsLocked()
+							rr.mu.Unlock()
+						}
+						if bulk {
+							projectionResourceSink, projectionTargetSink = rr.ListWithMetricsTargets()
+						} else {
+							projectionResourceSink = rr.List()
+							projectionTargetSink = make(map[string]MetricsTarget, len(projectionResourceSink))
+							for _, row := range projectionResourceSink {
+								if target := rr.MetricsTarget(row.ID); target != nil {
+									projectionTargetSink[row.ID] = *target
+								}
+							}
+						}
+					}
+				})
+			}
+		}
+	}
+}
diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go
index e7490bef07..8eacc8f37b 100644
--- a/internal/unifiedresources/registry_test.go
+++ b/internal/unifiedresources/registry_test.go
@@ -6968,3 +6968,68 @@ func TestEvaluateResourceHealthKeepsStaleTelemetryBehindAWinningAlert(t *testing
 		t.Fatalf("expected only the alert reason for a fresh source, got %+v", health.Reasons)
 	}
 }
+
+// A List-only broadcast must capture history coordinates without constructing
+// every typed view, and without trusting freshness timestamps as mutation keys.
+func TestRegistryProjectionSnapshotDetachesAndTracksMappings(t *testing.T) {
+	rr := NewRegistry(nil)
+	rr.IngestSnapshot(benchmarkVMState(3))
+	resources, targets := rr.ListWithMetricsTargets()
+	if !reflect.DeepEqual(resources, rr.List()) {
+		t.Fatal("bulk read changed resource content")
+	}
+	if !rr.viewsDirty || rr.cachedVMs != nil {
+		t.Fatal("projection forced unrelated typed views to materialize")
+	}
+	if len(targets) != len(resources) {
+		t.Fatalf("targets=%d resources=%d", len(targets), len(resources))
+	}
+	for _, r := range resources {
+		if want := rr.MetricsTarget(r.ID); want == nil || *want != targets[r.ID] {
+			t.Fatalf("target for %s differs from point resolver", r.ID)
+		}
+	}
+	id := resources[0].ID
+	observedAt := resources[0].LastSeen
+	rr.mu.Lock()
+	for key, mapped := range rr.bySource[SourceProxmox] {
+		if mapped == id {
+			delete(rr.bySource[SourceProxmox], key)
+		}
+	}
+	rr.bySource[SourceProxmox]["new-history-coordinate"] = id
+	rr.resources[id].Tags = []string{"customer-data"}
+	rr.invalidateViewsLocked()
+	rr.mu.Unlock()
+	changed, updatedTargets := rr.ListWithMetricsTargets()
+	if updatedTargets[id].ResourceID != "new-history-coordinate" {
+		t.Fatal("same-time mapping change left an old target")
+	}
+	if targets[id].ResourceID == "new-history-coordinate" {
+		t.Fatal("later mapping edit mutated a prior target snapshot")
+	}
+	for _, r := range changed {
+		if r.ID == id {
+			if !r.LastSeen.Equal(observedAt) || r.Policy.Sensitivity != ResourceSensitivityRestricted {
+				t.Fatal("same-time metadata edit was stale")
+			}
+			r.Canonical.Aliases[0] = "caller-change"
+			r.Tags[0] = "caller-change"
+			r.Policy.Routing.Redact = append(r.Policy.Routing.Redact, ResourceRedactionPath)
+		}
+	}
+	again, againTargets := rr.ListWithMetricsTargets()
+	if !reflect.DeepEqual(againTargets, updatedTargets) {
+		t.Fatal("detached map mutation escaped")
+	}
+	for _, r := range again {
+		if r.ID == id && (r.Tags[0] != "customer-data" || r.Canonical.Aliases[0] == "caller-change") {
+			t.Fatal("caller changed live registry")
+		}
+	}
+	empty := NewRegistry(nil)
+	rows, emptyTargets := empty.ListWithMetricsTargets()
+	if len(rows) != 0 || len(emptyTargets) != 0 {
+		t.Fatal("empty projection invented resources or targets")
+	}
+}

From 32f09dc13d6b3afaa802da9dcad7bba2cd233b1f Mon Sep 17 00:00:00 2001
From: "pulse-triage[bot]"
 <249995291+pulse-triage[bot]@users.noreply.github.com>
Date: Mon, 5 Oct 2026 11:30:00 +0100
Subject: [PATCH 2/5] Keep complete projection controls faithful to the parent
 path

The independent full-state oracle retains the parent unified view watermark. The paired benchmark preprocesses active health alerts once per pass, like the exact parent, rather than adding per-row work to the baseline. Production projection and source fields are unchanged.

Change-source: pulse-maintainer
---
 .../monitoring/broadcast_metrics_snapshot_test.go     | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go
index 4e61cc2f3d..f774d14b72 100644
--- a/internal/monitoring/broadcast_metrics_snapshot_test.go
+++ b/internal/monitoring/broadcast_metrics_snapshot_test.go
@@ -84,6 +84,11 @@ func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) {
 			t.Fatal(err)
 		}
 		want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC())
+		// The parent's unified view fills a missing store watermark from the
+		// listed resources. Preserve that full-state header in the oracle too.
+		if freshness := latestUnifiedResourceLastSeen(adapter.GetAll()); !freshness.IsZero() {
+			want.LastUpdate = freshness.UnixMilli()
+		}
 		got := m.buildBroadcastFrontendStateFromSnapshot(graph.State)
 		wantJSON, err := json.Marshal(want)
 		if err != nil {
@@ -189,9 +194,10 @@ var completeProjectionJSONSink []byte
 // metadata/health decoration and snapshot projection are part of both costs.
 func parentFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot, resources []unifiedresources.Resource, resolver MetricsTargetResourceStore, now time.Time) models.StateFrontend {
 	rows := unifiedresources.CoalescePresentationHostResources(resources)
+	healthAlerts := resourceHealthAlerts(snapshot.ActiveAlerts)
 	for i := range rows {
 		m.applyPersistedMetadataToUnifiedResource(&rows[i])
-		health := unifiedresources.EvaluateResourceHealth(rows[i], resourceHealthAlerts(snapshot.ActiveAlerts), now)
+		health := unifiedresources.EvaluateResourceHealth(rows[i], healthAlerts, now)
 		rows[i].Health = &health
 	}
 	projected, catalogs := convertPresentationResourcesForBroadcast(attachBroadcastMetricsTargets(rows, resolver))
@@ -201,6 +207,9 @@ func parentFrontendProjectionForTest(m *Monitor, snapshot models.StateSnapshot,
 	out.PolicyCatalog = catalogs.policies
 	out.AISafeSummaryCatalog = catalogs.aiSafeSummaries
 	out.ConnectedInfrastructure = buildConnectedInfrastructure(rows, snapshot)
+	if freshness := latestUnifiedResourceLastSeen(resources); !freshness.IsZero() {
+		out.LastUpdate = freshness.UnixMilli()
+	}
 	return out
 }
 

From 8b12e0ee1b678dff4a251e73b6f54f1776c1dd34 Mon Sep 17 00:00:00 2001
From: "pulse-triage[bot]"
 <249995291+pulse-triage[bot]@users.noreply.github.com>
Date: Mon, 5 Oct 2026 11:33:31 +0100
Subject: [PATCH 3/5] Pin live row health and clear across captured resource
 reads

The full current-state regression now requires a critical alert to reach the same freshly projected row. Continuity controls compare a healthy VM, critical transition and clear instead of using an already-offline host as a vacuous health control.

Change-source: pulse-maintainer
---
 .../monitoring/broadcast_metrics_snapshot_test.go   | 13 ++++++++++---
 internal/monitoring/monitor_host_agents_test.go     |  4 ++++
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go
index f774d14b72..a6eef3d7af 100644
--- a/internal/monitoring/broadcast_metrics_snapshot_test.go
+++ b/internal/monitoring/broadcast_metrics_snapshot_test.go
@@ -177,9 +177,16 @@ func TestBroadcastMetricsSnapshotContinuityAndLiveOverlays(t *testing.T) {
 			}
 		}
 	}
-	critical := unifiedresources.EvaluateResourceHealth(host, []unifiedresources.ResourceHealthAlert{{ResourceID: host.ID, Level: "critical", Type: "cpu"}}, now)
-	if critical.Verdict != unifiedresources.HealthCritical {
-		t.Fatal("live alert was hidden by captured data")
+	for _, r := range view.resources {
+		if r.Type != unifiedresources.ResourceTypeVM {
+			continue
+		}
+		baseline := unifiedresources.EvaluateResourceHealth(r, nil, now)
+		critical := unifiedresources.EvaluateResourceHealth(r, []unifiedresources.ResourceHealthAlert{{ResourceID: r.ID, Level: "critical", Type: "cpu"}}, now)
+		cleared := unifiedresources.EvaluateResourceHealth(r, nil, now)
+		if baseline.Verdict != unifiedresources.HealthOK || critical.Verdict != unifiedresources.HealthCritical || !reflect.DeepEqual(cleared, baseline) {
+			t.Fatal("live alert/clear was hidden by captured data")
+		}
 	}
 	if len(adapter.GetAll()) != 2 {
 		t.Fatal("continuity overlay wrote back to the live registry")
diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go
index 695d475395..b7d3708922 100644
--- a/internal/monitoring/monitor_host_agents_test.go
+++ b/internal/monitoring/monitor_host_agents_test.go
@@ -6763,6 +6763,10 @@ func TestBroadcastProjectionListsRegistryOnceAndKeepsLiveChanges(t *testing.T) {
 	if row == nil || row.CPU.Current != 77 || len(row.Tags) != 1 || row.Tags[0] != "new-tag" {
 		t.Fatalf("mutable row was stale: %#v", row)
 	}
+	var health unifiedresources.ResourceHealth
+	if err := json.Unmarshal(row.Health, &health); err != nil || health.Verdict != unifiedresources.HealthCritical {
+		t.Fatalf("live critical alert did not reach the freshly projected row: %s (%v)", row.Health, err)
+	}
 	if first.Resources[0].CPU.Current != oldCPU {
 		t.Fatal("later projection mutated an accepted baseline")
 	}

From d206b0399097bc8e0d65ed814a826a97c7e2b0a2 Mon Sep 17 00:00:00 2001
From: "pulse-triage[bot]"
 <249995291+pulse-triage[bot]@users.noreply.github.com>
Date: Mon, 5 Oct 2026 11:51:05 +0100
Subject: [PATCH 4/5] Retain bounded full-projection mismatch evidence

The new complete-content oracle exposed a typed-view discrepancy. Keep that failed result and report the first differing JSON field, rather than accept an undifferentiated mismatch or normalise away evidence. Only the synthetic fixture failure message changes; equality assertions and production source are unchanged.

Change-source: pulse-maintainer
---
 internal/monitoring/broadcast_metrics_snapshot_test.go | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go
index a6eef3d7af..d238907642 100644
--- a/internal/monitoring/broadcast_metrics_snapshot_test.go
+++ b/internal/monitoring/broadcast_metrics_snapshot_test.go
@@ -99,7 +99,13 @@ func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) {
 			t.Fatal(err)
 		}
 		if string(wantJSON) != string(gotJSON) {
-			t.Fatalf("%s: complete frontend/catalogue/infrastructure JSON differs", stage)
+			first := 0
+			for first < len(wantJSON) && first < len(gotJSON) && wantJSON[first] == gotJSON[first] {
+				first++
+			}
+			start := max(0, first-80)
+			t.Fatalf("%s: complete frontend/catalogue/infrastructure JSON differs at byte %d: want %q, got %q (bytes %d/%d)",
+				stage, first, wantJSON[start:min(len(wantJSON), first+160)], gotJSON[start:min(len(gotJSON), first+160)], len(wantJSON), len(gotJSON))
 		}
 		after, _ := json.Marshal(adapter.GetAll())
 		if string(before) != string(after) {

From 9cc8523881fbeb61eb7a90c91ddefe565eefb328 Mon Sep 17 00:00:00 2001
From: "pulse-triage[bot]"
 <249995291+pulse-triage[bot]@users.noreply.github.com>
Date: Mon, 5 Oct 2026 12:17:19 +0100
Subject: [PATCH 5/5] Compare complete broadcast health at one evaluation
 instant

The five predeclared diagnostic cases retain a real 1m versus 2m health-age counterexample between separate wall-clock evaluations. Give the independent content oracle and production projection the same private health clock without removing any JSON field; pin minute-boundary age advancement and original LastSeen. Production samples real UTC at the unchanged post-read evaluation point.

Contract-Neutral: Internal health-clock test seam preserves the real UTC evaluation point, public resource/history fields, source precedence, agent identity/lifecycle, tenant/revocation and safety contracts; no wire or permission delta.
Change-source: pulse-maintainer
---
 .../v6/internal/subsystems/monitoring.md      |  6 +++
 .../broadcast_metrics_snapshot_test.go        | 38 ++++++++++++++++++-
 .../monitoring/canonical_guardrails_test.go   |  2 +-
 internal/monitoring/monitor.go                |  9 ++++-
 4 files changed, 51 insertions(+), 4 deletions(-)

diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md
index fd23f74cc1..0e47f9e058 100644
--- a/docs/release-control/v6/internal/subsystems/monitoring.md
+++ b/docs/release-control/v6/internal/subsystems/monitoring.md
@@ -41,6 +41,12 @@ the exact parent's complete conversion and final JSON encoding, with cold,
 list-only, typed-clean and dirty inputs; they do not establish installed CPU/RSS
 relief, native identity refresh throughput or completion of #2199.
 
+Complete-content comparisons use one health-evaluation instant through a
+private clock seam. Production still samples real UTC at the same post-read
+boundary. The minute-boundary control requires age labels to advance without
+renewing LastSeen; comparisons retain every health field instead of stripping
+the reproduced `1m`/`2m` clock difference from JSON.
+
 ### Report identity lookups do not rebuild unrelated Kubernetes metadata
 
 The producer's two temporary host/Docker-host admission registries use a
diff --git a/internal/monitoring/broadcast_metrics_snapshot_test.go b/internal/monitoring/broadcast_metrics_snapshot_test.go
index d238907642..02e18e99cc 100644
--- a/internal/monitoring/broadcast_metrics_snapshot_test.go
+++ b/internal/monitoring/broadcast_metrics_snapshot_test.go
@@ -79,17 +79,18 @@ func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) {
 	}
 	assertEqual := func(stage string) {
 		t.Helper()
+		evaluationTime := time.Now().UTC()
 		before, err := json.Marshal(adapter.GetAll())
 		if err != nil {
 			t.Fatal(err)
 		}
-		want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, time.Now().UTC())
+		want := legacyFrontendProjectionForTest(m, graph.State, adapter.GetAll(), adapter, evaluationTime)
 		// The parent's unified view fills a missing store watermark from the
 		// listed resources. Preserve that full-state header in the oracle too.
 		if freshness := latestUnifiedResourceLastSeen(adapter.GetAll()); !freshness.IsZero() {
 			want.LastUpdate = freshness.UnixMilli()
 		}
-		got := m.buildBroadcastFrontendStateFromSnapshot(graph.State)
+		got := m.buildBroadcastFrontendStateFromSnapshotWithClock(graph.State, func() time.Time { return evaluationTime })
 		wantJSON, err := json.Marshal(want)
 		if err != nil {
 			t.Fatal(err)
@@ -125,6 +126,39 @@ func TestBroadcastMetricsSnapshotConnectedContent(t *testing.T) {
 	assertEqual("dirty-same-time")
 }
 
+func TestBroadcastMetricsSnapshotHealthClockBoundary(t *testing.T) {
+	m, adapter, id := newReadStateCloneTestMonitor(t, 1)
+	rows := adapter.GetAll()
+	original := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+	rows[0].LastSeen = original
+	rows[0].SourceStatus = map[unifiedresources.DataSource]unifiedresources.SourceStatus{
+		unifiedresources.SourceProxmox: {Status: "stale", LastSeen: original},
+	}
+	rr := unifiedresources.NewRegistry(nil)
+	rr.IngestResources(rows)
+	m.resourceStore = &broadcastProjectionCountingStore{MonitorAdapter: unifiedresources.NewMonitorAdapter(rr)}
+	for _, tc := range []struct {
+		age    time.Duration
+		detail string
+	}{{59 * time.Second, "0m"}, {61 * time.Second, "1m"}} {
+		calls := 0
+		got := m.buildBroadcastFrontendStateFromSnapshotWithClock(models.EmptyStateSnapshot(), func() time.Time {
+			calls++
+			return original.Add(tc.age)
+		})
+		if calls != 1 || len(got.Resources) != 1 || got.Resources[0].ID != id || got.Resources[0].LastSeen != original.UnixMilli() {
+			t.Fatal("health clock changed capture identity/original time or was not sampled once")
+		}
+		var health unifiedresources.ResourceHealth
+		if err := json.Unmarshal(got.Resources[0].Health, &health); err != nil {
+			t.Fatal(err)
+		}
+		if health.Verdict != unifiedresources.HealthStale || len(health.Reasons) != 1 || health.Reasons[0].Code != "telemetry_stale" || health.Reasons[0].Detail != tc.detail {
+			t.Fatalf("health age must advance across the minute boundary: %+v", health)
+		}
+	}
+}
+
 func TestBroadcastMetricsSnapshotOwnsTargetsBeforeLiveReplacement(t *testing.T) {
 	m, adapter, _ := newReadStateCloneTestMonitor(t, 4)
 	view := m.currentUnifiedStateView()
diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go
index 9e11cb4f7e..f65c96057a 100644
--- a/internal/monitoring/canonical_guardrails_test.go
+++ b/internal/monitoring/canonical_guardrails_test.go
@@ -2958,7 +2958,7 @@ func TestBroadcastProjectionMatchesPreviousPipeline(t *testing.T) {
 	if !view.freshness.IsZero() {
 		want.LastUpdate = view.freshness.UnixMilli()
 	}
-	got := m.buildBroadcastFrontendStateFromSnapshot(snapshot)
+	got := m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, func() time.Time { return now })
 	wantJSON, err := json.Marshal(want)
 	if err != nil {
 		t.Fatal(err)
diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go
index 162d189642..c722fc70f0 100644
--- a/internal/monitoring/monitor.go
+++ b/internal/monitoring/monitor.go
@@ -4333,6 +4333,13 @@ func buildFrontendStateFromSnapshot(snapshot models.StateSnapshot) models.StateF
 }
 
 func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateSnapshot) models.StateFrontend {
+	return m.buildBroadcastFrontendStateFromSnapshotWithClock(snapshot, time.Now)
+}
+
+// The clock is sampled at the same post-read health-evaluation boundary as
+// ordinary broadcasts. Tests can compare complete projections at one instant
+// without mistaking a naturally advancing health-age label for lost content.
+func (m *Monitor) buildBroadcastFrontendStateFromSnapshotWithClock(snapshot models.StateSnapshot, healthClock func() time.Time) models.StateFrontend {
 	frontendState := buildFrontendStateFromSnapshot(snapshot)
 	m.updateResourceStoreForRead(snapshot)
 	if m != nil && m.alertManager != nil {
@@ -4349,7 +4356,7 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS
 	// Coalescing owns the outer slice. Decorate that one projection in place,
 	// not three full-resource copies; nested store data is still read-only.
 	healthAlerts := resourceHealthAlerts(frontendState.ActiveAlerts)
-	now := time.Now().UTC()
+	now := healthClock().UTC()
 	for i := range broadcastResources {
 		m.applyPersistedMetadataToUnifiedResource(&broadcastResources[i])
 		health := unifiedresources.EvaluateResourceHealth(broadcastResources[i], healthAlerts, now)