From d5440ff43efd4964cabdcf3ae0beea0306cb94d7 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Tue, 25 Aug 2026 12:23:06 +0100 Subject: [PATCH] Slim the resource stream and catch up realtime gaps incrementally MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 2026-08-25 audit follow-up profiled the remaining frontend costs on the 50-node mock rig. Entering a tab more than one websocket tick behind the shared cache fell off the incremental path onto a full-estate deep-unwrap merge — a 2.1s unwrap dominated warm Alerts entry at 4x throttle — and the client-facing resource stream re-shipped ~3KB of mostly-static metadata per resource on every snapshot, REST recovery, and reconnect (governed gap resource-payload-static-metadata). The connection store now keeps a bounded per-tick changed-id history and serves unions from it, so an instance that mounts or resumes a few revisions behind catches up with a delta merge over the union instead of remerging the estate. Uncovered gaps and full-snapshot commits still fall back to keyed full reconciliation. Broadcast payloads dedupe the estate's distinct capability blobs into a content-addressed state-level capabilityCatalog referenced per resource by capabilitiesRef; default-posture resources (internal sensitivity, cloud-summary routing, no redactions) omit policy and aiSafeSummary; and canonicalIdentity.aliases no longer duplicates supersededIds. Ingestion reverses all three: refs expand through the catalog as per-row clones (reconcile mutates adopted objects in place), missing policies synthesize the default posture so a policy:null transition patch cannot strand a stale governed policy, and identity alias resolution consults supersededIds explicitly. Slimming edits only the per-broadcast copy, never stored monitor state, and the delta engine's generic top-level field diff carries catalog changes in the same frame as the first resource referencing a new entry. Rig A/B on the pinned 50-node mock (same VM as the morning baseline, single runs, +-30% variance; RandomMetrics mutates nearly every row per tick, so this estate is the worst case for the catch-up union): mobile 4x Alerts entry settles 3.9s -> 2.0s with long tasks 5.5s -> 3.3s, idle-30s long tasks drop 12.6s -> 10.8s mobile and 3.4s -> 2.8s desktop, remaining steps flat within variance. Sparse-delta estates gain more, because their catch-up unions stay small. /api/state on the pinned mock estate drops 4.75MB -> 4.09MB (-13.9%); the capability catalog is 7 entries totalling 2.5KB where 946 resources previously inlined the blobs, and 806 default-posture resources shed policy and aiSafeSummary. Browser-verified against a live build of this tree: byte-identical UI outcomes vs the unmodified baseline across overview/storage/docker/alerts probes at both viewports, with docker container lifecycle actions enabled through catalog-expanded capabilities. pulse-mobile and pulse-enterprise do not read the slimmed fields; the AI runtime consumes the internal model, which is unchanged. Contracts for monitoring, api-contracts, unified-resources, performance-and-scalability, alerts, agent-lifecycle, and storage-recovery are amended with the slimmed wire shape, the ingestion synthesis obligations, the catch-up history, and superseded-id resolution. --- ...urce-payload-static-metadata-2026-08-24.md | 35 ++++ .../v6/internal/subsystems/agent-lifecycle.md | 6 + .../v6/internal/subsystems/alerts.md | 7 + .../v6/internal/subsystems/api-contracts.md | 12 ++ .../v6/internal/subsystems/monitoring.md | 12 ++ .../subsystems/performance-and-scalability.md | 12 +- .../internal/subsystems/storage-recovery.md | 8 +- .../internal/subsystems/unified-resources.md | 19 ++- frontend-modern/browser-verification.json | 53 +++--- .../__tests__/useUnifiedResources.test.ts | 61 +++++++ .../src/hooks/useUnifiedResources.ts | 36 ++-- .../__tests__/websocket-unified.test.ts | 161 ++++++++++++++++++ .../src/stores/websocket-global.ts | 1 + frontend-modern/src/stores/websocket.ts | 86 +++++++++- frontend-modern/src/types/api.ts | 5 +- frontend-modern/src/types/resource.ts | 15 ++ .../utils/__tests__/resourceIdentity.test.ts | 18 ++ frontend-modern/src/utils/resourceIdentity.ts | 8 + internal/models/converters.go | 2 + internal/models/models_frontend.go | 25 ++- .../monitoring/canonical_guardrails_test.go | 2 +- .../issue1597_metric_authority_test.go | 2 +- internal/monitoring/monitor.go | 81 ++++++++- .../monitoring/monitor_host_agents_test.go | 45 +++++ .../monitoring/monitor_unified_state_test.go | 125 +++++++++++++- 25 files changed, 777 insertions(+), 60 deletions(-) diff --git a/docs/release-control/v6/internal/records/resource-payload-static-metadata-2026-08-24.md b/docs/release-control/v6/internal/records/resource-payload-static-metadata-2026-08-24.md index 4c93fac64..9f152cc66 100644 --- a/docs/release-control/v6/internal/records/resource-payload-static-metadata-2026-08-24.md +++ b/docs/release-control/v6/internal/records/resource-payload-static-metadata-2026-08-24.md @@ -50,3 +50,38 @@ contract updates and cross-client verification, not an opportunistic patch. start no longer fetches and compiles all ~3.1 MB of lazy chunks up front. - Per-tab scoped hydration and realtime gating on the Proxmox surface (`17bb2b3b7`, Performance lane). + +## 2026-08-25 partial remediation (interactive session, governed claim on this gap) + +Landed on `main` (same-day follow-up to the tick-pipeline fixes in +`7bac525af`): + +- **Capabilities catalog**: broadcast payloads dedupe the estate's distinct + capability blobs into a content-addressed state-level `capabilityCatalog` + referenced per resource via `capabilitiesRef`; the websocket store expands + refs back to inline `capabilities` at ingestion. On the 50-node mock the + catalog is 7 entries / 2.5KB where 946 resources previously inlined the + blobs. +- **Audience-scoped policy metadata**: default-posture resources (internal + sensitivity, cloud-summary routing, no redactions) omit `policy` and + `aiSafeSummary` from the stream; ingestion synthesizes the default posture + so consumer semantics are unchanged. 806 of 1,509 mock resources shed both + fields. +- **Alias dedupe**: broadcast `canonicalIdentity.aliases` no longer + duplicates `supersededIds`; client identity resolution now consults + `supersededIds` explicitly (alert overrides, thresholds, and workload + matching already did). + +Measured: `/api/state` 4.75MB -> 4.09MB (-13.9%) at the pinned mock estate. +Cross-client check: pulse-mobile and pulse-enterprise contain no reads of +`capabilities`/`canonicalIdentity`/`aiSafeSummary`/resource `policy` from +this stream; the AI runtime consumes the internal resource model, which is +untouched. + +**Remaining residual (why the gap stays open):** the canonicalIdentity alias +vocabulary itself (~0.8MB) still ships inline because superseded-spelling +resolution is load-bearing client-side while +`host-identity-fork-heal-on-reenrollment` remains open; moving identity +history behind a detail endpoint needs the alert-override/threshold matching +reworked onto an on-demand lookup. Platform source payloads (~1.4MB) are +live data, not static metadata, and are out of scope for this gap. diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index e7842ff0a..e44f3d75c 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -65,6 +65,12 @@ or rename correlations. websocket payloads. Carrying plural availability facets through that serializer is an adjacent monitoring/API projection and does not change agent enrollment, report admission, removal, update, profile, or command authority. +The serializer's broadcast slimming is likewise an adjacent projection with an +agent-identity obligation: superseded canonical ids produced by re-enrollment +or identity re-derivation must remain resolvable to the client through +`canonicalIdentity.supersededIds` even though they are no longer duplicated +into the broadcast alias list, so alert overrides and identity consumers keep +following a forked or re-enrolled host across its retired spellings. That shared monitor constructor may also copy the persisted alert schedule's initial notification target into the notification manager. This adjacent alerts/notifications wiring grants no agent enrollment, reporting, removal, diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 9f1be3cac..619403f4d 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -247,6 +247,13 @@ default construction path still restores. ## Shared Boundaries 1. `frontend-modern/src/stores/websocket.ts` shared with `performance-and-scalability`: the connection-owned realtime store is both the canonical alert truth boundary and the fleet-scale resource reconciliation hot path. + That shared store normalizes slimmed broadcast resources at ingestion — + expanding `capabilitiesRef` through the state `capabilityCatalog` and + synthesizing the default policy posture for resources published without one + — before any alert consumer resolves alert-to-resource identity or policy + display. Alert override and threshold identity matching must consult + `canonicalIdentity.supersededIds` alongside aliases, because broadcast + aliases no longer duplicate superseded canonical ids. 2. `internal/operationaltrust/contracts.go` shared with `notifications`: the operational trust contract is jointly consumed by canonical alert lifecycle ownership and notification delivery linkage without making delivery state operational truth. 3. `internal/proxmoxidentity/backup_identity.go` shared with `monitoring`, `storage-recovery`: Proxmox PBS backup subject identity is a shared runtime boundary for monitoring backup freshness, backup-age alert attribution, and recovery-point guest mapping. Alert multiline field presentation is shared with frontend-primitives: diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 8e84198fb..e099998a0 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -2580,6 +2580,18 @@ a new API state machine, queue contract, or verification-accounting field. host-agent row for the same machine must leave the API boundary as one hybrid `agent` resource with merged source facets, not as duplicate rows that disappear only after REST reconciliation. + Realtime `/api/state` and websocket state payloads carry deduped static + metadata: the state-level `capabilityCatalog` maps content-addressed ids to + the estate's distinct capability blobs and each resource references its set + via `capabilitiesRef` instead of inlining the blob per resource; a resource + published without an inline `policy` is, by contract, at the default posture + (internal sensitivity, cloud-summary routing, no redactions) and its + `aiSafeSummary` is withheld with it, while non-default postures keep both + inline; and `canonicalIdentity.aliases` no longer duplicates + `supersededIds`, which stay the canonical superseded-spelling carrier for + identity resolution. Because the websocket delta engine diffs top-level + state fields generically, a catalog change must ride the same frame as the + first resource referencing the new entry. Realtime `/api/state` and websocket state snapshots also own Proxmox tag presentation payloads. `pveTagStyles` is keyed by Proxmox instance and carries the parsed datacenter `tag-style` color overrides plus diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 7c9c564dc..7561145bd 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -1503,6 +1503,18 @@ canonical unified-resource metrics include `diskRead` or `diskWrite`, `ResourceFrontend.diskIO` through the shared resource converter, so `/api/state` and websocket consumers read disk throughput from the same freshness-gated resource metrics contract as CPU, memory, disk, and network. +That same broadcast projection owns client-payload static-metadata slimming +(governed gap `resource-payload-static-metadata`): identical resource +capability blobs are deduped into the state-level `capabilityCatalog` under +content-addressed ids and referenced per resource via `capabilitiesRef` +instead of being inlined on every row; resources whose derived policy posture +is the default (internal sensitivity, cloud-summary routing, no redactions) +omit `policy` and `aiSafeSummary` from the broadcast, which ingestion +synthesizes back, while non-default postures keep both inline; and broadcast +`canonicalIdentity.aliases` entries that duplicate `supersededIds` are +dropped, with the superseded ids still shipped for identity resolution. +Slimming edits only the per-broadcast copy refreshed by +`RefreshCanonicalMetadata`, never stored monitor state. Unraid ingest must preserve the agent's native disk topology fields through the monitoring model and read-state projection. `internal/monitoring/monitor_agents.go` and `internal/monitoring/monitor.go` must carry model, transport, filesystem, diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 3b9c87749..efd0e803d 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -222,7 +222,17 @@ cache once, then projects route filters from the already-canonical result; route-scoped consumers must not independently canonicalize the full estate. A hook instance observing an already-applied revision must not deep-unwrap the realtime store, and the merging instance dereferences raw store subtrees only -for rows the delta merge will clone. Workload table rows derived from canonical +for rows the delta merge will clone. The connection store keeps a bounded +per-revision changed-id history so an instance that mounts or resumes several +revisions behind the shared cache catches up through the unioned changed-id +set as a delta merge instead of a full-estate remerge; the full fallback is +reserved for uncovered gaps and full-snapshot commits. The broadcast payload +itself is a scalability surface: duplicated capability blobs travel once +through the state-level `capabilityCatalog` and per-resource +`capabilitiesRef`, default-posture policy and AI-safe prose are omitted and +synthesized at ingestion, and superseded canonical ids are not duplicated +into the alias list, so full snapshots, REST recovery, and reconnects do not +re-ship static metadata per resource. Workload table rows derived from canonical snapshots reuse the previous row object whenever the serialized row is unchanged, and a refresh that changes nothing returns the previous row array itself, so per-tick row identity churn stays bounded to guests whose data diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index ed116bb1d..35e900941 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -1126,9 +1126,11 @@ recovery scope, or a storage/recovery-owned secret source. must invalidate stale in-flight REST refreshes before their errors or request-guard cleanup can leak into the active resource snapshot. The shared hook's realtime tick path may read the connection store lazily — - skipping revisions the shared canonical cache already holds and - dereferencing raw store subtrees only for rows the delta merge will clone — - but that optimization must stay invisible to storage/recovery consumers: it + skipping revisions the shared canonical cache already holds, catching up + multi-revision gaps through the store's bounded changed-id history union, + and dereferencing raw store subtrees only for rows the delta merge will + clone — but that optimization must stay invisible to storage/recovery + consumers: it must not change which canonical fields a merged row carries, must not reorder the REST-first hydration guarantee above, and a skipped store read must never leave a storage/recovery projection behind the shared cache's diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 19d75465a..dde98dc0e 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -1180,8 +1180,13 @@ AI-only summary payloads, or page-local heuristics. instance dereferences raw store subtrees only for rows the delta merge will clone (flagged ids, host-merge members, and ids absent from the shared cache). A sequential revision with unchanged route membership - patches only the changed row indices plus the bounded agent coalescing set; - initial hydration, missed revisions, additions, removals, or reorderings + patches only the changed row indices plus the bounded agent coalescing set. + The connection store retains a bounded per-revision changed-id history; an + instance that resumes several revisions behind the shared cache must catch + up through the unioned changed-id set as an incremental delta merge whenever + the history covers the gap, so tab entry and re-entry do not deep-unwrap or + remerge the full estate. Only initial hydration, uncovered revision gaps, + full-snapshot commits, additions, removals, or reorderings fall back to keyed full reconciliation. Route-prefetch and route-realtime activation are separate: a prefetched hidden surface may retain REST data without subscribing its full @@ -1189,6 +1194,16 @@ AI-only summary payloads, or page-local heuristics. cache. Richer REST-only facets are promoted into that cache before thinner realtime deltas are applied, so the optimization cannot discard disk I/O, PBS, policy, or provider metadata. + Broadcast payload slimming is reversed at the connection-store ingestion + boundary, before any canonical merge or consumer read: `capabilitiesRef` is + expanded into per-row inline `capabilities` through the state payload's + `capabilityCatalog` (per-row clones, because store reconciliation mutates + adopted objects in place), and a resource arriving without a policy is given + a synthesized default posture (internal sensitivity, cloud-summary routing, + no redactions) so a posture transition patched as `policy: null` cannot + leave a stale governed policy behind. Client identity-alias resolution must + consult `canonicalIdentity.supersededIds` explicitly, because broadcast + aliases no longer duplicate superseded canonical ids. That same unified-resource owner also defines the canonical transport projection for operator-facing resources: `/api/resources` and websocket `state.resources` must share `ContractResourceType`, canonical display diff --git a/frontend-modern/browser-verification.json b/frontend-modern/browser-verification.json index c20aac62c..25b54b7ec 100644 --- a/frontend-modern/browser-verification.json +++ b/frontend-modern/browser-verification.json @@ -1,26 +1,35 @@ { "version": 1, - "base_sha": "7bac525af355bc5c8eb6ff2db8f9ce116697013f", - "verified_at": "2026-08-25T10:44:58Z", + "base_sha": "51b7357285b0ac7020212aff0a67b8cd5974757c", + "verified_at": "2026-08-25T11:20:27Z", "result": "passed", "changed_paths": [ - "frontend-modern/src/components/shared/horizontalRailVisibilityModel.ts", - "frontend-modern/src/components/shared/useActiveHorizontalRailItemVisibility.ts", - "frontend-modern/src/features/platformPage/sharedPlatformPage.tsx" + "frontend-modern/src/hooks/useUnifiedResources.ts", + "frontend-modern/src/stores/websocket-global.ts", + "frontend-modern/src/stores/websocket.ts", + "frontend-modern/src/types/api.ts", + "frontend-modern/src/types/resource.ts", + "frontend-modern/src/utils/resourceIdentity.ts" ], "content_sha256": { - "frontend-modern/src/components/shared/horizontalRailVisibilityModel.ts": "1fa3ad5d3cfffd0ccf78ced497d87564602cb124103175c17ba6c3e4b9531c4f", - "frontend-modern/src/components/shared/useActiveHorizontalRailItemVisibility.ts": "add047b223402743fbabf5d7c9b867f160174ab2e48baf53c0c60509caf01510", - "frontend-modern/src/features/platformPage/sharedPlatformPage.tsx": "3507374c3c9c236a95ee50793f2462b16d71680f13c1f4c864efc3ea88159659" + "frontend-modern/src/hooks/useUnifiedResources.ts": "5b63ab55ce28b72216561acf7b818c55179e70acfc2a35155cd6391490a13f54", + "frontend-modern/src/stores/websocket-global.ts": "59d2ddc55882f98c72b5c893ce16c250ba5c0eeb20e31419e23f651ab72c00b5", + "frontend-modern/src/stores/websocket.ts": "56c7fa58dc5f79855b6ab4a9c9c36c85f00a554a76ea981b593e175a4547dba1", + "frontend-modern/src/types/api.ts": "88a40975498e284b3c704b384a6a2ab52bd1fd19e6fd1de522953d570f2afa25", + "frontend-modern/src/types/resource.ts": "9b73aa4df0b734f37f19f1b607872442ea988f5212d6bdb2a59be4b613d83249", + "frontend-modern/src/utils/resourceIdentity.ts": "f0a0203a2e5de11afcffa1ae70382c9e55249e9a5651b0fb9611b3f0f196fb51" }, "routes": [ + "/login", "/proxmox/overview", - "/proxmox/replication" + "/proxmox/storage", + "/docker", + "/alerts" ], "viewports": [ { - "width": 1440, - "height": 900 + "width": 1280, + "height": 800 }, { "width": 390, @@ -28,16 +37,20 @@ } ], "states": [ - "Proxmox overview tab rail with all six destinations at desktop width", - "Proxmox overview tab rail initially clipped at phone width with the right affordance visible", - "phone tab rail manually scrolled to its 207px end position with the left affordance visible", - "manually scrolled phone rail after a same-width viewport-height resize", - "Replication selected from the manually explored rail with the active label clear of the left overlay" + "authenticated session against a dev build of this exact diff serving the 50-node mock estate with the slimmed broadcast payload (capabilityCatalog with 7 entries, zero inline capabilities, 806 default-posture policies and aiSafeSummary omitted, aliases deduped against supersededIds)", + "Proxmox overview table populated at both viewports with live 2s websocket ticks applying", + "Proxmox storage sub-tab populated (sensitive-posture resources keep inline policy)", + "Docker workloads table with an expanded container row and its lifecycle actions group", + "container lifecycle menu open: Start disabled with running reason, Stop and Restart enabled via catalog-expanded capabilities", + "Alerts utility surface rendering the active-alert list", + "console error stream captured on both builds: only the pre-existing CSP inline-style warnings, no new errors" ], "interactions": [ - "native horizontal wheel/swipe projection across the phone-width Proxmox section rail", - "phone viewport resize after manual rail exploration preserved scrollLeft at 207px", - "selected Replication from the revealed rail and confirmed route navigation plus active-tab visibility", - "repeated the final phone flow after the active-label clearance adjustment: swipe to 190px, resize without snap-back, then select Replication at a safe 55px viewport offset" + "logged in and waited for landing hydration", + "switched platform tabs to Docker and sub-tabs to Storage and back to Overview", + "expanded a docker container row and opened its container-actions menu, reading each action's enabled/disabled state and aria-label", + "navigated to Alerts and back", + "fetched /api/state on the slimmed build and asserted capabilityCatalog presence, per-resource capabilitiesRef, omitted default policies, and the 4.75MB to 4.09MB size drop", + "ran the identical probe script against the unmodified baseline build and confirmed byte-identical UI outcomes (rows, drawer, action states, console errors)" ] } diff --git a/frontend-modern/src/hooks/__tests__/useUnifiedResources.test.ts b/frontend-modern/src/hooks/__tests__/useUnifiedResources.test.ts index 3b919bb9d..4972ebfe9 100644 --- a/frontend-modern/src/hooks/__tests__/useUnifiedResources.test.ts +++ b/frontend-modern/src/hooks/__tests__/useUnifiedResources.test.ts @@ -145,6 +145,7 @@ describe('useUnifiedResources', () => { version: number; changedIds: ReadonlySet | null; }) => unknown; + let wsStoreMock: Record; let useUnifiedResources: UseUnifiedResourcesModule['useUnifiedResources']; let useStorageRecoveryResources: UseUnifiedResourcesModule['useStorageRecoveryResources']; let resetUnifiedResourcesCacheForTests: UseUnifiedResourcesModule['__resetUnifiedResourcesCacheForTests']; @@ -199,6 +200,7 @@ describe('useUnifiedResources', () => { setWsState = _setWsState; const wsStore = { connected, initialDataReceived, state, resourceChange }; + wsStoreMock = wsStore as unknown as Record; vi.doMock('@/utils/apiClient', () => ({ apiFetch: apiFetchMock, @@ -303,6 +305,65 @@ describe('useUnifiedResources', () => { dispose(); }); + it('catches up multi-tick version gaps through the changed-id history instead of a full remerge', async () => { + // Distinct nested cpu objects: the shared wsResource base would otherwise + // alias one cpu object across both rows and the two writes below. + const vm1 = createWsResource({ + id: 'vm-1', + name: 'vm-1', + displayName: 'vm-1', + type: 'vm', + cpu: { current: 15 }, + }); + const vm2 = createWsResource({ + id: 'vm-2', + name: 'vm-2', + displayName: 'vm-2', + type: 'vm', + cpu: { current: 15 }, + }); + setWsState('resources', [vm1, vm2]); + setWsResourceChange({ version: 1, changedIds: null }); + + let dispose = () => {}; + let result: ReturnType | undefined; + createRoot((d) => { + dispose = d; + result = useUnifiedResources({ + query: '', + cacheKey: 'all-resources', + initialHydration: 'prefer-ws', + }); + }); + + await waitForResourceCount(() => result!.resources().length); + + const vm2RowBefore = result!.resources().find((resource) => resource.id === 'vm-2'); + + // Two ticks land while no instance applies them (version 1 -> 3). The + // store's history union names only vm-1, so the delta path must re-merge + // vm-1 and keep the unflagged vm-2 row by identity — a full remerge would + // rebuild it. + const historySpy = vi.fn().mockReturnValue(new Set(['vm-1'])); + wsStoreMock.changedResourceIdsSince = historySpy; + batch(() => { + setWsState('resources', 0, 'cpu', 'current', 88); + setWsState('resources', 1, 'cpu', 'current', 99); + setWsResourceChange({ version: 3, changedIds: new Set(['vm-1']) }); + setWsState('lastUpdate', 1738843203000); + }); + + await waitForValue( + () => result!.resources().find((resource) => resource.id === 'vm-1')?.cpu?.current, + 88, + ); + expect(historySpy).toHaveBeenCalledWith(1); + const vm2RowAfter = result!.resources().find((resource) => resource.id === 'vm-2'); + expect(vm2RowAfter).toBe(vm2RowBefore); + expect(apiFetchMock).not.toHaveBeenCalled(); + dispose(); + }); + it('hydrates realtime store entries that arrive without being flagged changed', async () => { // The incremental path only unwraps raw subtrees for entries it expects to // clone (changed ids, agents, and ids missing from the merge cache). An diff --git a/frontend-modern/src/hooks/useUnifiedResources.ts b/frontend-modern/src/hooks/useUnifiedResources.ts index 69d786aec..f07fb0eee 100644 --- a/frontend-modern/src/hooks/useUnifiedResources.ts +++ b/frontend-modern/src/hooks/useUnifiedResources.ts @@ -1633,10 +1633,21 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) { const allResourcesEntry = getUnifiedResourcesCacheEntry( buildScopedUnifiedResourcesCacheKey(ALL_RESOURCES_CACHE_KEY, currentOrgScope), ); - const canApplyIncrementally = - resourceChange.changedIds !== null && - allResourcesEntry.realtimeVersion > 0 && - resourceChange.version === allResourcesEntry.realtimeVersion + 1; + // An instance that mounts or resumes more than one tick behind the shared + // cache used to fall off the incremental path onto a full-estate remerge — + // the dominant Alerts-entry / tab re-entry cost in the 2026-08-25 profile. + // Catch up through the store's bounded changed-id history instead; a + // single-tick gap keeps using the latest delta without touching history. + const resolveCatchUpIds = (sinceVersion: number): ReadonlySet | null => { + if (sinceVersion <= 0 || resourceChange.version <= sinceVersion) return null; + if (resourceChange.version === sinceVersion + 1) return resourceChange.changedIds; + return wsStore.changedResourceIdsSince?.(sinceVersion) ?? null; + }; + const allEntryCatchUpIds = + allResourcesEntry.realtimeVersion > 0 + ? resolveCatchUpIds(allResourcesEntry.realtimeVersion) + : null; + const canApplyIncrementally = allEntryCatchUpIds !== null; const realtimeSnapshotAlreadyApplied = resourceChange.version > 0 && allResourcesEntry.hasSnapshot && @@ -1665,9 +1676,9 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) { ? allResourcesEntry.resources : canApplyIncrementally ? mergeCanonicalResourceDeltaSnapshot( - readWsResources(resourceChange.changedIds), + readWsResources(allEntryCatchUpIds), allResourcesEntry.resources, - resourceChange.changedIds!, + allEntryCatchUpIds!, ) : mergeCanonicalResourceSnapshot(readWsResources(null), allResourcesEntry.resources); const projectedResources = filterCanonicalUnifiedResources( @@ -1706,22 +1717,23 @@ export function useUnifiedResources(options?: UseUnifiedResourcesOptions) { allResourcesEntry.realtimeVersion = resourceChange.version; } - const canPatchProjectionIncrementally = - resourceChange.changedIds !== null && + const cacheEntryCatchUpIds = cacheEntry.hasSnapshot && cacheEntry.realtimeVersion > 0 && - resourceChange.version === cacheEntry.realtimeVersion + 1 && - resolvedProjectedResources === projectedResources; + resolvedProjectedResources === projectedResources + ? resolveCatchUpIds(cacheEntry.realtimeVersion) + : null; + const canPatchProjectionIncrementally = cacheEntryCatchUpIds !== null; const changedResourceTouchesAgent = canPatchProjectionIncrementally && mergedWsResources.some( - (resource) => resource.type === 'agent' && resourceChange.changedIds!.has(resource.id), + (resource) => resource.type === 'agent' && cacheEntryCatchUpIds!.has(resource.id), ); const incrementalPatchIndices = canPatchProjectionIncrementally ? resolveIncrementalResourcePatchIndices( resources as unknown as Resource[], resolvedProjectedResources, - resourceChange.changedIds!, + cacheEntryCatchUpIds!, changedResourceTouchesAgent, ) : null; diff --git a/frontend-modern/src/stores/__tests__/websocket-unified.test.ts b/frontend-modern/src/stores/__tests__/websocket-unified.test.ts index 85b37cb68..01b3d43f9 100644 --- a/frontend-modern/src/stores/__tests__/websocket-unified.test.ts +++ b/frontend-modern/src/stores/__tests__/websocket-unified.test.ts @@ -368,6 +368,167 @@ describe('websocket store unified resource contract', () => { } }); + it('expands capabilitiesRef through the catalog and synthesizes omitted default policies', async () => { + const { store, dispose } = await createStoreHarness(); + try { + await waitForOpenTick(); + + emitMessage({ + type: 'initialState', + data: { + connectedInfrastructure: [], + capabilityCatalog: { + cap1: [{ name: 'restart', type: 'common', description: 'Restart the guest' }], + }, + resources: [ + { + id: 'vm-1', + type: 'vm', + name: 'vm-1', + status: 'running', + capabilitiesRef: 'cap1', + }, + { + id: 'storage-1', + type: 'storage', + name: 'tank', + status: 'online', + policy: { sensitivity: 'sensitive', routing: { scope: 'local-first', redact: [] } }, + }, + ], + lastUpdate: 100, + activeAlerts: [], + recentlyResolved: [], + }, + }); + + const vm = store.state.resources.find((resource) => resource.id === 'vm-1'); + expect(vm?.capabilities?.map((capability) => capability.name)).toEqual(['restart']); + // Omitted policy means default posture; ingestion synthesizes it so every + // policy consumer keeps seeing an inline policy. + expect(vm?.policy).toEqual({ + sensitivity: 'internal', + routing: { scope: 'cloud-summary', redact: [] }, + }); + const storage = store.state.resources.find((resource) => resource.id === 'storage-1'); + expect(storage?.policy?.sensitivity).toBe('sensitive'); + + // A later delta can move a resource's ref; the catalog change rides the + // same frame and the patched row re-expands. + emitMessage({ + type: 'rawData', + data: { + lastUpdate: 200, + capabilityCatalog: { + cap1: [{ name: 'restart', type: 'common', description: 'Restart the guest' }], + cap2: [{ name: 'stop', type: 'common', description: 'Stop the guest' }], + }, + resourceDelta: { + upserts: [{ id: 'vm-1', capabilitiesRef: 'cap2' }], + removed: [], + order: ['vm-1', 'storage-1'], + }, + }, + }); + const vmAfter = store.state.resources.find((resource) => resource.id === 'vm-1'); + expect(vmAfter?.capabilities?.map((capability) => capability.name)).toEqual(['stop']); + + // A posture transition back to default nulls the policy in the patch; + // ingestion re-synthesizes the default instead of keeping the stale one. + emitMessage({ + type: 'rawData', + data: { + lastUpdate: 300, + resourceDelta: { + upserts: [{ id: 'storage-1', policy: null }], + removed: [], + order: ['vm-1', 'storage-1'], + }, + }, + }); + const storageAfter = store.state.resources.find((resource) => resource.id === 'storage-1'); + expect(storageAfter?.policy).toEqual({ + sensitivity: 'internal', + routing: { scope: 'cloud-summary', redact: [] }, + }); + } finally { + dispose(); + } + }); + + it('serves unions from the bounded changed-id history so late consumers catch up incrementally', async () => { + const { store, dispose } = await createStoreHarness(); + try { + await waitForOpenTick(); + + emitMessage({ + type: 'initialState', + data: { + connectedInfrastructure: [], + resources: [ + { id: 'agent-1', type: 'agent', name: 'agent-1', status: 'online' }, + { id: 'vm-1', type: 'vm', name: 'vm-1', status: 'running' }, + { id: 'vm-2', type: 'vm', name: 'vm-2', status: 'running' }, + ], + lastUpdate: 100, + activeAlerts: [], + recentlyResolved: [], + }, + }); + // version 1: full snapshot — not coverable by the history. + expect(store.changedResourceIdsSince(0)).toBeNull(); + + emitMessage({ + type: 'rawData', + data: { + lastUpdate: 200, + resourceDelta: { + upserts: [{ id: 'vm-1', lastSeen: 200 }], + removed: [], + order: ['agent-1', 'vm-1', 'vm-2'], + }, + }, + }); + emitMessage({ + type: 'rawData', + data: { + lastUpdate: 300, + resourceDelta: { + upserts: [{ id: 'vm-2', lastSeen: 300 }], + removed: [], + order: ['agent-1', 'vm-1', 'vm-2'], + }, + }, + }); + + // versions 2 and 3 are delta commits: a consumer still on version 1 gets + // the union, one on version 2 gets just the last delta, and a current + // consumer gets null (nothing to catch up). + expect(store.changedResourceIdsSince(1)).toEqual(new Set(['vm-1', 'vm-2'])); + expect(store.changedResourceIdsSince(2)).toEqual(new Set(['vm-2'])); + expect(store.changedResourceIdsSince(3)).toBeNull(); + // version 1 was a full snapshot, so a span reaching before it is not + // coverable and must force the caller onto the full-merge path. + expect(store.changedResourceIdsSince(0)).toBeNull(); + + emitMessage({ + type: 'initialState', + data: { + connectedInfrastructure: [], + resources: [{ id: 'agent-1', type: 'agent', name: 'agent-1', status: 'online' }], + lastUpdate: 400, + activeAlerts: [], + recentlyResolved: [], + }, + }); + // A later full snapshot invalidates the accumulated history. + expect(store.changedResourceIdsSince(3)).toBeNull(); + expect(store.changedResourceIdsSince(2)).toBeNull(); + } finally { + dispose(); + } + }); + it('keeps canonically merged hosts consistent with full snapshots across deltas (#1601)', async () => { const { store, dispose } = await createStoreHarness(); try { diff --git a/frontend-modern/src/stores/websocket-global.ts b/frontend-modern/src/stores/websocket-global.ts index 149353924..a363b0ab5 100644 --- a/frontend-modern/src/stores/websocket-global.ts +++ b/frontend-modern/src/stores/websocket-global.ts @@ -63,6 +63,7 @@ const createNoopWebSocketStore = (): ReturnType => initialDataReceived, updateProgress, resourceChange, + changedResourceIdsSince: () => null, shutdown: () => {}, reconnect: () => {}, switchUrl: () => {}, diff --git a/frontend-modern/src/stores/websocket.ts b/frontend-modern/src/stores/websocket.ts index 753965de6..abe11d153 100644 --- a/frontend-modern/src/stores/websocket.ts +++ b/frontend-modern/src/stores/websocket.ts @@ -7,7 +7,8 @@ import type { ResolvedAlert, ConnectedInfrastructureItem, } from '@/types/api'; -import type { Resource } from '@/types/resource'; +import type { Resource, ResourceCapability } from '@/types/resource'; +import { createDefaultResourcePolicy } from '@/types/resource'; import { logger } from '@/utils/logger'; import { POLLING_INTERVALS, WEBSOCKET } from '@/constants'; import { notificationStore } from './notifications'; @@ -252,6 +253,27 @@ export function createWebSocketStore(url: string) { version: number; changedIds: ReadonlySet | null; }>({ version: 0, changedIds: null }); + // Bounded per-tick changed-id history so a consumer that mounted or resumed + // a few ticks behind the live version can still catch up with a delta merge + // instead of a full-estate remerge. A full-snapshot commit (changedIds null) + // invalidates the whole span, so the history resets there. + const RESOURCE_CHANGE_HISTORY_LIMIT = 30; + let resourceChangeHistory: { version: number; changedIds: ReadonlySet }[] = []; + + const changedResourceIdsSince = (sinceVersion: number): ReadonlySet | null => { + if (sinceVersion >= resourceChangeVersion) return null; + if (resourceChangeHistory.length === 0) return null; + if (resourceChangeHistory[0].version > sinceVersion + 1) return null; + if (resourceChangeHistory[resourceChangeHistory.length - 1].version !== resourceChangeVersion) { + return null; + } + const union = new Set(); + for (const entry of resourceChangeHistory) { + if (entry.version <= sinceVersion) continue; + entry.changedIds.forEach((id) => union.add(id)); + } + return union; + }; // Track alerts with pending acknowledgment changes to prevent race conditions const pendingAckChanges = new Map(); @@ -340,6 +362,34 @@ export function createWebSocketStore(url: string) { // and Solid's reconcile mutates adopted objects in place. let rawServerResources: Resource[] | null = null; const deferredResourceIds = new Set(); + // Latest capabilityCatalog from the state payload. Broadcast resources carry + // capabilitiesRef instead of inline capability blobs; ingestion expands the + // ref back into `capabilities` so consumers keep the inline shape. + let capabilityCatalog: Record = {}; + // Records which catalog ref an expanded capabilities array was cloned from, + // so a patched row re-expands only when its ref actually changed. + const capabilitiesExpandedFromRef = new WeakMap(); + + const hydrateSlimResource = (resource: Resource): void => { + if (!resource || typeof resource !== 'object') return; + const ref = resource.capabilitiesRef; + if (ref) { + const current = resource.capabilities as unknown as object | undefined; + if (!current || capabilitiesExpandedFromRef.get(current) !== ref) { + const entry = capabilityCatalog[ref]; + if (entry) { + // Per-row clone: reconcile mutates adopted objects in place, so rows + // must not share one materialized capabilities array. + const expanded = structuredClone(entry); + capabilitiesExpandedFromRef.set(expanded as unknown as object, ref); + resource.capabilities = expanded; + } + } + } + if (!resource.policy) { + resource.policy = createDefaultResourcePolicy(); + } + }; let lastFullStateRecoveryAt = 0; // Set once the server has sent a full snapshot too large for the inbound // guard. Asking for another one over the socket would only reproduce the same @@ -456,10 +506,25 @@ export function createWebSocketStore(url: string) { changedCount: changedResourceIds?.size ?? nextResources.length, }); setState('resources', reconcile(nextResources, { key: 'id' })); + const committedChangedIds = changedResourceIds ? new Set(changedResourceIds) : null; setResourceChange({ version: ++resourceChangeVersion, - changedIds: changedResourceIds ? new Set(changedResourceIds) : null, + changedIds: committedChangedIds, }); + if (committedChangedIds) { + resourceChangeHistory.push({ + version: resourceChangeVersion, + changedIds: committedChangedIds, + }); + if (resourceChangeHistory.length > RESOURCE_CHANGE_HISTORY_LIMIT) { + resourceChangeHistory.splice( + 0, + resourceChangeHistory.length - RESOURCE_CHANGE_HISTORY_LIMIT, + ); + } + } else { + resourceChangeHistory = []; + } syncContainerCommands(nextResources, changedResourceIds); }; @@ -615,6 +680,13 @@ export function createWebSocketStore(url: string) { setInitialDataReceived(true); } + // Adopt the capability catalog before any resource handling below: + // the delta engine ships catalog changes in the same frame as the + // first resource that references a new entry. + if (message.data.capabilityCatalog !== undefined) { + capabilityCatalog = message.data.capabilityCatalog ?? {}; + } + // Canonical resource contract: // `state.resources` is the authoritative frontend model. // `state.connectedInfrastructure` is the authoritative reporting projection. @@ -645,6 +717,10 @@ export function createWebSocketStore(url: string) { if (message.data.resources !== undefined) { deferredResourceIds.clear(); if (Array.isArray(message.data.resources)) { + // Expand capabilitiesRef and synthesize omitted default + // policies before the rows feed the baseline clone and the + // canonical merge. + message.data.resources.forEach(hydrateSlimResource); // Only a full payload actually delivered on this socket can // establish its server-owned delta baseline. REST recovery is // an independently built display snapshot and stays baseline- @@ -678,6 +754,10 @@ export function createWebSocketStore(url: string) { rawServerResources, message.data.resourceDelta, ); + // Patched rows are fresh objects: re-synthesize a nulled-out + // default policy and re-expand capabilities when a patch moved + // the ref. Unpatched rows no-op. + appliedDelta.resources.forEach(hydrateSlimResource); rawServerResources = appliedDelta.resources; changedResourceIds = appliedDelta.changedIds; if (typeof document !== 'undefined' && document.visibilityState === 'hidden') { @@ -1167,6 +1247,7 @@ export function createWebSocketStore(url: string) { initialDataReceived, updateProgress, resourceChange, + changedResourceIdsSince, shutdown, reconnect: () => { if (isDisposed) return; @@ -1193,6 +1274,7 @@ export function createWebSocketStore(url: string) { setInitialDataReceived(false); setUpdateProgress(null); setResourceChange({ version: ++resourceChangeVersion, changedIds: null }); + resourceChangeHistory = []; setState(reconcile(createInitialState())); setActiveAlerts(reconcile({})); setRecentlyResolved(reconcile({})); diff --git a/frontend-modern/src/types/api.ts b/frontend-modern/src/types/api.ts index 69d1a0b16..67d0b9131 100644 --- a/frontend-modern/src/types/api.ts +++ b/frontend-modern/src/types/api.ts @@ -1,6 +1,6 @@ // Properly typed TypeScript interfaces for Pulse API -import type { Resource } from './resource'; +import type { Resource, ResourceCapability } from './resource'; import type { EvidenceEnvelope, LifecycleTransition, OperationalRecord } from './operationalTrust'; export interface APITokenRecord { @@ -31,6 +31,9 @@ export interface State { pveTagStyles?: Record; // Unified resources (canonical resource model) resources: Resource[]; + // Distinct capability blobs referenced by Resource.capabilitiesRef; the + // websocket store expands refs into inline capabilities at ingestion. + capabilityCatalog?: Record | null; } export interface ResourceStateDelta { diff --git a/frontend-modern/src/types/resource.ts b/frontend-modern/src/types/resource.ts index cf16c5bf6..01e0c7694 100644 --- a/frontend-modern/src/types/resource.ts +++ b/frontend-modern/src/types/resource.ts @@ -221,6 +221,17 @@ export const requiresGovernedResourceDisplay = (policy?: ResourcePolicy | null): if (!policy) return false; return policy.routing.scope === 'local-only' || (policy.routing.redact?.length ?? 0) > 0; }; + +// The posture the server omits from broadcast payloads: default-posture +// resources carry no inline policy on the wire (see the backend's +// isDefaultBroadcastResourcePolicy), and the websocket store synthesizes it at +// ingestion so every policy consumer keeps seeing an inline policy. Fresh +// objects per resource — a shared instance would be corrupted in place by +// store reconciliation when one resource's posture later changes. +export const createDefaultResourcePolicy = (): ResourcePolicy => ({ + sensitivity: 'internal', + routing: { scope: 'cloud-summary', redact: [] }, +}); export type ResourceChangeConfidence = 'high' | 'medium' | 'low'; export type ResourceChangeKind = | 'state_transition' @@ -1587,6 +1598,10 @@ export interface Resource { policy?: ResourcePolicy; aiSafeSummary?: string; capabilities?: ResourceCapability[]; + // Reference into the state payload's capabilityCatalog; the websocket store + // expands it into `capabilities` at ingestion so consumers keep reading the + // inline shape. + capabilitiesRef?: string; actionReadiness?: ResourceActionReadiness[]; sourceStatus?: Record; relationships?: ResourceRelationship[]; diff --git a/frontend-modern/src/utils/__tests__/resourceIdentity.test.ts b/frontend-modern/src/utils/__tests__/resourceIdentity.test.ts index 906e18ab2..b8c2d7a6e 100644 --- a/frontend-modern/src/utils/__tests__/resourceIdentity.test.ts +++ b/frontend-modern/src/utils/__tests__/resourceIdentity.test.ts @@ -89,6 +89,24 @@ describe('resourceIdentity', () => { ]); }); + it('resolves superseded canonical ids even though broadcast aliases no longer carry them', () => { + // Slimmed broadcast payloads stop duplicating supersededIds into the + // alias list, so the identity resolver has to consult them explicitly for + // alert overrides and lookups keyed by a retired spelling. + const aliases = getResourceIdentityAliases( + makeResource({ + canonicalIdentity: { + primaryId: 'agent:core2026', + aliases: ['agent:core2026', 'core2026'], + supersededIds: ['agent-core2026-a1b2c3', 'agent-core2026-d4e5f6'], + }, + }), + ); + expect(aliases).toContain('agent-core2026-a1b2c3'); + expect(aliases).toContain('agent-core2026-d4e5f6'); + expect(aliases).toContain('agent:core2026'); + }); + it('uses actionable linked identities before falling back to unified ids', () => { expect( getPrimaryResourceIdentity( diff --git a/frontend-modern/src/utils/resourceIdentity.ts b/frontend-modern/src/utils/resourceIdentity.ts index 3472d81a7..dcb1da81b 100644 --- a/frontend-modern/src/utils/resourceIdentity.ts +++ b/frontend-modern/src/utils/resourceIdentity.ts @@ -138,9 +138,17 @@ export const getResourceIdentityAliases = (resource: Resource): string[] => { .map((alias) => asTrimmedString(alias)) .filter((alias): alias is string => Boolean(alias)) : []; + // Broadcast payloads no longer duplicate superseded canonical ids into the + // alias list, so identity resolution has to consult them explicitly. + const canonicalSupersededIds = Array.isArray(canonical?.supersededIds) + ? canonical.supersededIds + .map((id) => asTrimmedString(id)) + .filter((id): id is string => Boolean(id)) + : []; const raw = [ ...canonicalAliases, + ...canonicalSupersededIds, resource.metricsTarget?.resourceId, resource.discoveryTarget?.agentId, resource.discoveryTarget?.resourceId, diff --git a/internal/models/converters.go b/internal/models/converters.go index d64dd3779..1e9e32cd1 100644 --- a/internal/models/converters.go +++ b/internal/models/converters.go @@ -1189,6 +1189,7 @@ type ResourceConvertInput struct { Policy json.RawMessage AISafeSummary string Capabilities json.RawMessage + CapabilitiesRef string Relationships json.RawMessage RecentChanges json.RawMessage FacetCounts json.RawMessage @@ -1272,6 +1273,7 @@ func ConvertResourceToFrontend(input ResourceConvertInput) ResourceFrontend { Policy: input.Policy, AISafeSummary: input.AISafeSummary, Capabilities: input.Capabilities, + CapabilitiesRef: input.CapabilitiesRef, Relationships: input.Relationships, RecentChanges: input.RecentChanges, FacetCounts: input.FacetCounts, diff --git a/internal/models/models_frontend.go b/internal/models/models_frontend.go index d65557098..41b762be4 100644 --- a/internal/models/models_frontend.go +++ b/internal/models/models_frontend.go @@ -964,6 +964,13 @@ type StateFrontend struct { // Unified resources - the new way to access all monitored entities Resources []ResourceFrontend `json:"resources"` ConnectedInfrastructure []ConnectedInfrastructureItemFrontend `json:"connectedInfrastructure"` + // CapabilityCatalog maps content-hash ids to capability blobs referenced by + // ResourceFrontend.CapabilitiesRef. Estates carry only a handful of + // distinct capability sets, so the catalog replaces per-resource inline + // duplication in broadcast payloads. The websocket delta engine diffs + // top-level state fields generically, so catalog changes ride the same + // frame as the resource refs that need them. + CapabilityCatalog map[string]json.RawMessage `json:"capabilityCatalog,omitempty"` } // EmptyStateFrontend returns a canonical empty frontend state with stable @@ -1064,12 +1071,18 @@ type ResourceFrontend struct { // Identity for deduplication Identity *ResourceIdentityFrontend `json:"identity,omitempty"` - DiscoveryTarget json.RawMessage `json:"discoveryTarget,omitempty"` - MetricsTarget json.RawMessage `json:"metricsTarget,omitempty"` - Canonical json.RawMessage `json:"canonicalIdentity,omitempty"` - Policy json.RawMessage `json:"policy,omitempty"` - AISafeSummary string `json:"aiSafeSummary,omitempty"` - Capabilities json.RawMessage `json:"capabilities,omitempty"` + DiscoveryTarget json.RawMessage `json:"discoveryTarget,omitempty"` + MetricsTarget json.RawMessage `json:"metricsTarget,omitempty"` + Canonical json.RawMessage `json:"canonicalIdentity,omitempty"` + Policy json.RawMessage `json:"policy,omitempty"` + AISafeSummary string `json:"aiSafeSummary,omitempty"` + Capabilities json.RawMessage `json:"capabilities,omitempty"` + // CapabilitiesRef references an entry in StateFrontend.CapabilityCatalog. + // Broadcast payloads dedupe the small set of distinct capability blobs + // through the catalog instead of inlining ~3KB per resource; consumers + // resolve the ref at ingestion. Inline Capabilities remains valid for + // producers that do not carry a catalog. + CapabilitiesRef string `json:"capabilitiesRef,omitempty"` Relationships json.RawMessage `json:"relationships,omitempty"` RecentChanges json.RawMessage `json:"recentChanges,omitempty"` FacetCounts json.RawMessage `json:"facetCounts,omitempty"` diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 64e57063a..2e3380b9a 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -604,7 +604,7 @@ func TestBroadcastResourceProjectionCoalescesSplitHostIdentities(t *testing.T) { for _, snippet := range []string{ "metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState)", "broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources)", - "frontendState.Resources = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver)", + "frontendState.Resources, frontendState.CapabilityCatalog = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver)", "frontendState.ConnectedInfrastructure = buildConnectedInfrastructure(broadcastResources, snapshot)", "func attachBroadcastMetricsTargets(", "allResources = unifiedresources.CoalescePresentationHostResources(allResources)", diff --git a/internal/monitoring/issue1597_metric_authority_test.go b/internal/monitoring/issue1597_metric_authority_test.go index 3b5e950e1..b62bdc1c7 100644 --- a/internal/monitoring/issue1597_metric_authority_test.go +++ b/internal/monitoring/issue1597_metric_authority_test.go @@ -87,7 +87,7 @@ func TestIssue1597UnifiedGuestCPUFeedsDashboardDetailsAndAlerts(t *testing.T) { }) adapter := unifiedresources.NewMonitorAdapter(registry) - frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter) + frontend, _ := convertResourcesForBroadcast(adapter.GetAll(), adapter) if len(frontend) != 1 { t.Fatalf("dashboard resource count = %d, want one unified guest", len(frontend)) } diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 5a629bb65..0bc9fe44a 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -2,6 +2,8 @@ package monitoring import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" stderrors "errors" "fmt" @@ -4165,7 +4167,7 @@ func (m *Monitor) buildBroadcastFrontendStateFromSnapshot(snapshot models.StateS metricsTargetResolver := broadcastMetricsTargetResolver(unifiedView.readState) broadcastResources := unifiedresources.CoalescePresentationHostResources(unifiedView.resources) broadcastResources = m.applyPersistedMetadataToUnifiedResources(broadcastResources) - frontendState.Resources = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver) + frontendState.Resources, frontendState.CapabilityCatalog = convertResourcesForBroadcast(broadcastResources, metricsTargetResolver) frontendState.ConnectedInfrastructure = buildConnectedInfrastructure(broadcastResources, snapshot) if !unifiedView.freshness.IsZero() { frontendState.LastUpdate = unifiedView.freshness.UnixMilli() @@ -5603,10 +5605,11 @@ func (m *Monitor) getResourcesForBroadcast() []models.ResourceFrontend { m.mu.RLock() store := m.resourceStore m.mu.RUnlock() - return convertResourcesForBroadcast( + resources, _ := convertResourcesForBroadcast( m.applyPersistedMetadataToUnifiedResources(m.getUnifiedResourcesForBroadcast()), broadcastMetricsTargetResolver(store), ) + return resources } func (m *Monitor) applyPersistedMetadataToUnifiedResources(resources []unifiedresources.Resource) []unifiedresources.Resource { @@ -5808,13 +5811,15 @@ func (m *Monitor) dockerAppContainerCustomURL( return "", false } -// convertResourcesForBroadcast converts unified resources into the frontend payload shape. +// convertResourcesForBroadcast converts unified resources into the frontend +// payload shape. Distinct capability blobs are deduped into the returned +// catalog and referenced per resource via capabilitiesRef. func convertResourcesForBroadcast( allResources []unifiedresources.Resource, metricsTargetResolvers ...MetricsTargetResourceStore, -) []models.ResourceFrontend { +) ([]models.ResourceFrontend, map[string]json.RawMessage) { if len(allResources) == 0 { - return []models.ResourceFrontend{} + return []models.ResourceFrontend{}, nil } allResources = attachBroadcastMetricsTargets( allResources, @@ -5828,8 +5833,15 @@ func convertResourcesForBroadcast( } converted := make([]broadcastResource, 0, len(allResources)) + capabilityCatalog := make(map[string]json.RawMessage) for _, r := range allResources { input := monitorResourceToConvertInput(r) + if len(input.Capabilities) > 0 { + id := capabilityCatalogID(input.Capabilities) + capabilityCatalog[id] = input.Capabilities + input.CapabilitiesRef = id + input.Capabilities = nil + } sortKey := strings.ToLower(input.DisplayName) if sortKey == "" { sortKey = strings.ToLower(input.Name) @@ -5852,7 +5864,10 @@ func convertResourcesForBroadcast( for i, resource := range converted { result[i] = models.ConvertResourceToFrontend(resource.input) } - return result + if len(capabilityCatalog) == 0 { + capabilityCatalog = nil + } + return result, capabilityCatalog } func broadcastMetricsTargetResolver(source interface{}) MetricsTargetResourceStore { @@ -5920,6 +5935,7 @@ func monitorResourceToConvertInput(resource unifiedresources.Resource) models.Re resource.MetricsTarget = monitorMetricsTarget(resource, resourceType) } unifiedresources.RefreshCanonicalMetadata(&resource) + slimResourceForBroadcast(&resource) name, displayName := monitorFrontendNames(resource, resourceType) platformID := monitorPlatformID(resource, resourceType) @@ -6007,6 +6023,59 @@ func monitorRawJSON(value interface{}) json.RawMessage { return encoded } +// slimResourceForBroadcast trims static metadata the client-facing stream was +// re-shipping on every resource (governed gap resource-payload-static-metadata). +// It only edits the freshly refreshed per-broadcast copy, never stored state: +// - Aliases duplicating superseded canonical ids are dropped; the ids remain +// available in canonicalIdentity.supersededIds, which identity consumers +// also resolve against. +// - Default-posture policy and its AI-safe summary are omitted; consumers +// treat a missing policy as the default posture, and the summary is only +// rendered for non-default postures. +func slimResourceForBroadcast(resource *unifiedresources.Resource) { + if resource == nil { + return + } + if canonical := resource.Canonical; canonical != nil && len(canonical.SupersededIDs) > 0 && + len(canonical.Aliases) > 0 { + superseded := make(map[string]struct{}, len(canonical.SupersededIDs)) + for _, id := range canonical.SupersededIDs { + superseded[strings.ToLower(strings.TrimSpace(id))] = struct{}{} + } + aliases := make([]string, 0, len(canonical.Aliases)) + for _, alias := range canonical.Aliases { + if _, dup := superseded[strings.ToLower(strings.TrimSpace(alias))]; dup { + continue + } + aliases = append(aliases, alias) + } + canonical.Aliases = aliases + } + if isDefaultBroadcastResourcePolicy(resource.Policy) { + resource.Policy = nil + resource.AISafeSummary = "" + } +} + +// isDefaultBroadcastResourcePolicy mirrors the frontend's +// hasDefaultResourcePolicyPosture: internal sensitivity, cloud-summary +// routing, and no redactions is the posture consumers assume when the +// broadcast omits the policy entirely. +func isDefaultBroadcastResourcePolicy(policy *unifiedresources.ResourcePolicy) bool { + return policy != nil && + policy.Sensitivity == unifiedresources.ResourceSensitivityInternal && + policy.Routing.Scope == unifiedresources.ResourceRoutingScopeCloudSummary && + len(policy.Routing.Redact) == 0 +} + +// capabilityCatalogID derives the stable catalog id for one marshaled +// capabilities blob. Content-addressed ids keep the catalog and refs stable +// across restarts and reconnects, so they never churn websocket deltas. +func capabilityCatalogID(encoded json.RawMessage) string { + digest := sha256.Sum256(encoded) + return hex.EncodeToString(digest[:6]) +} + func monitorDiscoveryTarget(resource unifiedresources.Resource, resourceType string) *unifiedresources.DiscoveryTarget { switch resourceType { case "agent", "docker-host": diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index e24740959..f12c1ce1c 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -104,6 +104,51 @@ func TestMonitoringBroadcastCarriesEveryAvailabilityProjection(t *testing.T) { } } +func TestBroadcastSlimmingKeepsHostAgentSupersededIdentityResolvable(t *testing.T) { + // A re-enrolled host carries retired canonical spellings. Broadcast + // slimming stops duplicating them into the alias list, but they must stay + // resolvable through supersededIds so alert overrides and identity + // consumers keep following the host across its retired ids. + input := monitorResourceToConvertInput(unifiedresources.Resource{ + ID: "agent-core2026", + Type: unifiedresources.ResourceTypeAgent, + Name: "core2026", + Status: unifiedresources.StatusOnline, + Sources: []unifiedresources.DataSource{unifiedresources.SourceAgent}, + SupersededCanonicalIDs: []string{"agent-core2026-a1b2c3", "agent-core2026-d4e5f6"}, + Agent: &unifiedresources.AgentData{ + AgentID: "core2026", + Hostname: "core2026", + }, + }) + + var canonical struct { + Aliases []string `json:"aliases"` + SupersededIDs []string `json:"supersededIds"` + } + if err := json.Unmarshal(input.Canonical, &canonical); err != nil { + t.Fatalf("unmarshal canonicalIdentity: %v (payload %s)", err, input.Canonical) + } + if len(canonical.SupersededIDs) != 2 { + t.Fatalf("supersededIds = %v, want both retired spellings", canonical.SupersededIDs) + } + for _, alias := range canonical.Aliases { + if alias == "agent-core2026-a1b2c3" || alias == "agent-core2026-d4e5f6" { + t.Fatalf("aliases must not duplicate superseded ids, got %v", canonical.Aliases) + } + } + // The live alias vocabulary the host is reachable by must survive the trim. + aliasSet := make(map[string]struct{}, len(canonical.Aliases)) + for _, alias := range canonical.Aliases { + aliasSet[alias] = struct{}{} + } + for _, required := range []string{"agent:core2026", "core2026"} { + if _, ok := aliasSet[required]; !ok { + t.Fatalf("aliases = %v, want %q retained", canonical.Aliases, required) + } + } +} + func TestHostAgentCorrelationKeepsSameNamedProxmoxProvidersDistinct(t *testing.T) { now := time.Now() state := &models.State{ diff --git a/internal/monitoring/monitor_unified_state_test.go b/internal/monitoring/monitor_unified_state_test.go index 67595876e..0be31b8cc 100644 --- a/internal/monitoring/monitor_unified_state_test.go +++ b/internal/monitoring/monitor_unified_state_test.go @@ -134,7 +134,7 @@ func TestConvertResourcesForBroadcastCoalescesSplitHostResources(t *testing.T) { }, } - frontend := convertResourcesForBroadcast(resources) + frontend, _ := convertResourcesForBroadcast(resources) if len(frontend) != 1 { t.Fatalf("expected split host resources to coalesce into 1 frontend resource, got %d: %#v", len(frontend), frontend) } @@ -218,7 +218,7 @@ func TestConvertResourcesForBroadcastAttachesResolvedStorageMetricsTarget(t *tes }) adapter := unifiedresources.NewMonitorAdapter(registry) - frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter) + frontend, _ := convertResourcesForBroadcast(adapter.GetAll(), adapter) var storagePayload struct { ID string `json:"id"` @@ -338,7 +338,7 @@ func TestConvertResourcesForBroadcastKeepsLinkedGuestCPUAndHistoryTargetAligned( }) adapter := unifiedresources.NewMonitorAdapter(registry) - frontend := convertResourcesForBroadcast(adapter.GetAll(), adapter) + frontend, _ := convertResourcesForBroadcast(adapter.GetAll(), adapter) if len(frontend) != 1 { t.Fatalf("broadcast resources = %d, want one merged guest: %+v", len(frontend), frontend) } @@ -686,7 +686,8 @@ func TestMonitorBuildBroadcastFrontendStateUsesCanonicalMockUnifiedResources(t * // pin an exact merge count. Require the broadcast count to fall // within a 5% tolerance of the convert-pipeline count so future // fixture bumps stay green as long as the merge contract is honored. - expectedBroadcastCount := len(convertResourcesForBroadcast(expectedResources)) + expectedBroadcastResources, _ := convertResourcesForBroadcast(expectedResources) + expectedBroadcastCount := len(expectedBroadcastResources) tolerance := expectedBroadcastCount / 20 if tolerance < 5 { tolerance = 5 @@ -783,7 +784,7 @@ func TestConvertResourcesForBroadcastIncludesCanonicalHealthContext(t *testing.T }, } - frontend := convertResourcesForBroadcast(resources) + frontend, _ := convertResourcesForBroadcast(resources) if len(frontend) != 2 { t.Fatalf("expected 2 frontend resources, got %d", len(frontend)) } @@ -1000,3 +1001,117 @@ func TestMonitorMonitoredSystemUsageDoesNotDoubleCountLinkedProxmoxHostContinuit t.Fatalf("MonitoredSystemUsage().Count = %d, want 1", usage.Count) } } + +func TestBroadcastPayloadSlimsStaticMetadata(t *testing.T) { + sharedCapabilities := []unifiedresources.ResourceCapability{{ + Name: "restart", + Type: unifiedresources.CapabilityTypeCommon, + Description: "Restart the guest", + MinimumApprovalLevel: unifiedresources.ApprovalAdmin, + }} + resources := []unifiedresources.Resource{ + { + ID: "vm-100", + Type: unifiedresources.ResourceTypeVM, + Name: "web-01", + Status: unifiedresources.StatusOnline, + LastSeen: time.Date(2026, 8, 25, 10, 0, 0, 0, time.UTC), + Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox}, + SupersededCanonicalIDs: []string{"vm-superseded-hash-1", "vm-superseded-hash-2"}, + Capabilities: sharedCapabilities, + }, + { + ID: "vm-101", + Type: unifiedresources.ResourceTypeVM, + Name: "web-02", + Status: unifiedresources.StatusOnline, + LastSeen: time.Date(2026, 8, 25, 10, 0, 0, 0, time.UTC), + Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox}, + Capabilities: sharedCapabilities, + }, + { + ID: "storage-1", + Type: unifiedresources.ResourceTypeStorage, + Name: "tank", + Status: unifiedresources.StatusOnline, + LastSeen: time.Date(2026, 8, 25, 10, 0, 0, 0, time.UTC), + Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox}, + }, + } + + frontend, catalog := convertResourcesForBroadcast(resources) + if len(frontend) != 3 { + t.Fatalf("expected 3 frontend resources, got %d", len(frontend)) + } + byID := make(map[string]map[string]json.RawMessage, len(frontend)) + refs := make(map[string]string, len(frontend)) + for _, resource := range frontend { + encoded, err := json.Marshal(resource) + if err != nil { + t.Fatalf("marshal %s: %v", resource.ID, err) + } + fields := map[string]json.RawMessage{} + if err := json.Unmarshal(encoded, &fields); err != nil { + t.Fatalf("unmarshal %s: %v", resource.ID, err) + } + byID[resource.ID] = fields + refs[resource.ID] = resource.CapabilitiesRef + } + + // Default-posture workloads must not re-ship policy or AI-safe prose. + vm := byID["vm-100"] + if _, ok := vm["policy"]; ok { + t.Fatalf("default-posture vm must omit policy, got %s", vm["policy"]) + } + if _, ok := vm["aiSafeSummary"]; ok { + t.Fatalf("default-posture vm must omit aiSafeSummary, got %s", vm["aiSafeSummary"]) + } + // Sensitive storage keeps its governed posture inline. + storage := byID["storage-1"] + var storagePolicy struct { + Sensitivity string `json:"sensitivity"` + } + if err := json.Unmarshal(storage["policy"], &storagePolicy); err != nil { + t.Fatalf("sensitive storage must keep policy inline: %v (payload %s)", err, storage["policy"]) + } + if storagePolicy.Sensitivity != "sensitive" { + t.Fatalf("storage sensitivity = %q", storagePolicy.Sensitivity) + } + + // Superseded ids stay resolvable via supersededIds but are no longer + // duplicated into the alias list. + var canonical struct { + Aliases []string `json:"aliases"` + SupersededIDs []string `json:"supersededIds"` + } + if err := json.Unmarshal(vm["canonicalIdentity"], &canonical); err != nil { + t.Fatalf("unmarshal canonicalIdentity: %v", err) + } + if len(canonical.SupersededIDs) != 2 { + t.Fatalf("supersededIds = %v", canonical.SupersededIDs) + } + for _, alias := range canonical.Aliases { + if alias == "vm-superseded-hash-1" || alias == "vm-superseded-hash-2" { + t.Fatalf("aliases must not duplicate superseded ids, got %v", canonical.Aliases) + } + } + + // Identical capability sets collapse onto one catalog entry. + if refs["vm-100"] == "" || refs["vm-100"] != refs["vm-101"] { + t.Fatalf("expected shared capabilitiesRef, got %q vs %q", refs["vm-100"], refs["vm-101"]) + } + if _, ok := vm["capabilities"]; ok { + t.Fatalf("catalog broadcast must not inline capabilities, got %s", vm["capabilities"]) + } + entry, ok := catalog[refs["vm-100"]] + if !ok { + t.Fatalf("catalog missing ref %q (catalog %v)", refs["vm-100"], catalog) + } + var decoded []unifiedresources.ResourceCapability + if err := json.Unmarshal(entry, &decoded); err != nil { + t.Fatalf("decode catalog entry: %v", err) + } + if len(decoded) != 1 || decoded[0].Name != "restart" { + t.Fatalf("catalog entry = %+v", decoded) + } +}