Stop a failing availability check keeping a node online (#2663)

Follows #2643, which made the stale pass read each source's own verdict
but left availability checks counting by delivery.

## Problem
Once every source with a verdict had gone quiet, any current
availability sighting kept its target online. A node that
`preserveOrExpireNodes` expired stayed online while its HTTP check
failed, and a quiet failing check lifted that node's offline verdict to
warning. One sighting stands for every check on a target and carries
whichever check was projected last, so a failing check that had just run
also vouched for another check's old pass.

## Change
- **Registry verdict** (`aggregateStatus`,
`availabilityChecksProveOnline`): each check on a monitored resource is
judged by its own evidence window (`Evidence.ValidUntil`). An enabled,
passing check with current evidence proves the resource answers and
reads online once one of the resource's own sources has gone quiet and
none that is current has a verdict. Every other check abstains, current
or quiet. A failing probe never makes a guest count as stopped
(workloads summary, Proxmox page counts). A check's own row keeps the
ordinary rules.
- **Ownership**: only a target's own quiet sources hand the stale pass
its status. Projecting, re-running or retargeting a check re-applies the
stale pass's rule to targets it owns, because the resources API replays
checks after its stale pass. A manual link judges the merged resource by
the checks it carries.
- **Poller evidence** (`internal/monitoring`):
- A local location beside other locations ages out like a silent remote
path ("no recent local check").
- An available check's evidence lasts only as long as its newest
reachable path.
- After an entitlement lapse, an adopted remote result keeps the earlier
of its agent check time and the server receipt time.
- A multi-location aggregate is never adopted as a local observation;
the target reads as never checked until the local poller runs.

Contracts: `unified-resources.md` ("Status follows each source's own
verdict") replaces the residual paragraph, and `monitoring.md` covers
the multi-location rules.

## Verification
- New tests in `registry_test.go`, `registry_merge_policy_test.go` and
`availability_probe_agent_test.go`. Each fails against the base runtime
via `go test -overlay`, including both projection orders.
- Full `internal/unifiedresources` and `internal/monitoring` suites
pass, plus the availability, connection, workload and resource-state
tests in `internal/api`, `resourceapi` and `chartapi`.

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -3175,7 +3175,20 @@ Monitoring retains one current status per target and location, using server
 receipt time for remote freshness while keeping the agent-authored observation
 time as evidence metadata. A stale or disconnected agent path derives as
 indeterminate at read time; slow or fast agent clocks cannot manufacture or
-conceal a disconnect. Aggregation is conservative: all reachable paths are
+conceal a disconnect. A local path beside other locations that has not run
+within the same window derives as indeterminate too ("no recent local check").
+The target takes its freshness from its newest location, so otherwise a
+failing remote report kept an old local pass counted as reachable and the
+check read available with current evidence. For the same reason an available
+check's evidence lasts only as long as its newest reachable path, not a
+fresher path that failed. When the external-probe entitlement lapses, a
+single remote result stands in for the local location until the local poller
+runs, with the earlier of its agent check time and the server's receipt time
+as its check time, so a fast agent clock cannot keep it current. An aggregate
+over several locations never stands in: a target without a retained local
+observation reads as never checked, and not connected, until the local poller
+runs.
+Aggregation is conservative: all reachable paths are
 healthy, mixed reachable and failed/unknown paths are degraded, all current
 paths unreachable are unavailable, and no reachable path with incomplete or
 indeterminate coverage is unknown. Only aggregate unavailability advances the

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

@@ -6184,9 +6184,20 @@ pass and manual links both apply that function:
   the best of equal ones. A quiet source drops out of that decision.
 - Once every source is quiet, an `offline` verdict survives and any other
   verdict reads `warning`. The best of those wins.
-- A current facet sighting carries no verdict (the PBS host-agent association,
-  an availability check projected onto its target). It counts as `online` only
-  when no current source has a verdict, and reads `warning` once quiet.
+- A current facet sighting carries no verdict (the PBS host-agent association).
+  It counts as `online` only when no current source has a verdict, and reads
+  `warning` once quiet.
+- Availability checks projected onto a monitored resource rank like that facet,
+  but their shared sighting is not their verdict. `availabilityChecksProveOnline`
+  judges each check at the pass's `now` by its own evidence window
+  (`Evidence.ValidUntil`: two poll intervals for a single local check, the
+  probe report window for a remote or multi-location one). An enabled,
+  passing check with current evidence proves the resource answers and reads
+  `online` once one of the resource's own sources has gone quiet and none that
+  is current has a verdict. Every other check abstains, current or quiet: a
+  failing check proves only that one port or service does not answer, and a
+  pass whose evidence lapsed proves nothing now. A check's own row keeps the ordinary
+  rules.
 
 The stale pass used to read delivery as the verdict. A Proxmox node the cluster
 reported offline on a live poll came back `online` once its linked agent fell
@@ -6210,24 +6221,52 @@ online stays `online` when its linked agent stops reporting. A live agent keeps
 a node `online` when the Proxmox poll reports it offline, which the Proxmox
 nodes table reads with `connectionHealth: error` as a stale provider. When a
 guest's or container's only source goes quiet, a running one stays `warning`
-and a stopped one stays `offline`.
+and a stopped one stays `offline`. In each quiet case a projected availability
+check with current evidence lifts the resource to `online`, as described
+below.
 
 The sighting still reads `stale`, so health keeps its `telemetry_stale` reason
 and monitored-system reasons are unchanged. In-memory clones keep the verdicts,
 including a facet's missing one. A serialized copy carries none, so
 `IngestResources` gives each sighting of a copy that lost them all the stored
 resource status; a merged row's separate source verdicts do not survive that
-round trip. A manual link that joins two resources reported by one source keeps
+round trip, and a copy whose status a passing check supplied gives that status
+to every source. A manual link that joins two resources reported by one source keeps
 the fresher sighting. `registry_merge_policy_test.go` pins the node,
 poller-expired, guest, round-trip, copy, agent, Docker, Kubernetes,
 mixed-source and manual-link cases, and the aggregation table.
 
-Availability checks still count by delivery. With every source that has a
-verdict quiet, a current check keeps its target `online` even when the check
-fails, as it did before. Several checks share one sighting whose freshness
-comes from whichever check was projected last, so an availability verdict
-needs each check's own freshness, including local checks that miss their
-cadence, and has to survive manual links. That is left to a follow-up.
+Availability checks used to count by delivery too. With every source that
+has a verdict quiet, a current failing check kept a node `preserveOrExpireNodes`
+expired `online`, and a quiet one lifted it to `warning`. Several checks share
+one sighting whose freshness comes from whichever check was projected last, so
+a failing check that just ran could also vouch for another check's old pass.
+Each check is now judged by its own evidence. A failing probe no longer holds
+an expired node `online`, and it never makes a guest read stopped to the
+consumers that count `offline` as stopped (the workloads summary, the Proxmox
+page counts). The failure stays on the check's own row, which raises an
+outage incident once the check confirms one (past its failure threshold, from
+every location). A check observed from several locations relies on the poller
+ageing out a location that stopped running and lasting an available check's
+evidence only as long as its newest reachable path (monitoring.md), so one
+location's fresh failure cannot lend its freshness to another location's old
+pass.
+
+The verdict is read from the checks on the resource when status is
+aggregated. Projecting, re-running or retargeting a check also re-applies the
+stale pass's rule to every target it touches whose status that pass owns,
+because a check is not a delivery from the target's own sources and the
+resources API replays checks after its stale pass. The pass owns a status only
+through the target's own sightings: the checks' sighting going quiet neither
+hands it the target nor, once removed, takes the target away, so a target
+whose own sources have not gone quiet keeps the status they gave it. A check
+retargeted away therefore takes its verdict with it at once. A
+manual link keeps only the primary's own checks, so the merged resource is
+judged by the checks it shows and never by the linked resource's fresher
+sighting. `registry_test.go` pins the expired-node and quiet-guest cases, both
+projection orders, re-judging on check changes, ownership through the
+target's own sightings and the manual link; the aggregation table pins the
+check rules.
 
 ### Canonical object drawer hierarchy
 

internal/monitoring/availability_poller.go

@@ -321,8 +321,7 @@ func (m *Monitor) availabilityStatusSnapshotForTargets(targets []config.Availabi
 				}
 				if legacyLocationID == expected[0] || expected[0] == config.AvailabilityObservationLocationLocal {
 					if expected[0] == config.AvailabilityObservationLocationLocal {
-						legacy.ProbeAgentID = ""
-						legacy.ProbeReportReceivedAt = time.Time{}
+						legacy = adoptAvailabilityStatusAsLocal(legacy)
 					}
 					locations[expected[0]] = legacy
 				}
@@ -330,13 +329,22 @@ func (m *Monitor) availabilityStatusSnapshotForTargets(targets []config.Availabi
 		}
 		if len(expected) == 1 && expected[0] == config.AvailabilityObservationLocationLocal {
 			if _, ok := locations[expected[0]]; !ok {
-				if legacy, exists := out[target.ID]; exists {
-					legacy.ProbeAgentID = ""
-					legacy.ProbeReportReceivedAt = time.Time{}
-					locations[expected[0]] = legacy
-				} else {
+				legacy, exists := out[target.ID]
+				switch {
+				case !exists:
 					delete(out, target.ID)
 					continue
+				case legacy.ExpectedLocations <= 1:
+					locations[expected[0]] = adoptAvailabilityStatusAsLocal(legacy)
+				default:
+					// An aggregate over several locations is not one observation
+					// the local location can stand in for: its freshness may be
+					// one path's and its availability another's. The target
+					// reads as never checked until the local poller runs, with
+					// an entry of its own so connection health cannot fall back
+					// to the cached state from before the lapse.
+					out[target.ID] = availabilityStatusFromTarget(target)
+					continue
 				}
 			}
 		}
@@ -348,6 +356,8 @@ func (m *Monitor) availabilityStatusSnapshotForTargets(targets []config.Availabi
 			}
 			if status.ProbeAgentID != "" {
 				status = m.deriveAvailabilityProbeStaleness(target, status, now)
+			} else if len(expected) > 1 {
+				status = deriveLocalAvailabilityObservationStaleness(target, status, now)
 			}
 			locations[locationID] = status
 		}
@@ -356,6 +366,21 @@ func (m *Monitor) availabilityStatusSnapshotForTargets(targets []config.Availabi
 	return out
 }
 
+// adoptAvailabilityStatusAsLocal presents a retained observation as the local
+// location's once the check falls back to local observation, as it does when
+// the external-probe entitlement lapses. A local location's check time is its
+// freshness, but a remote observation's check time is the agent's clock, so it
+// is held to the server's receipt time: a fast agent clock must not keep the
+// adopted result current after its receipt is cleared.
+func adoptAvailabilityStatusAsLocal(status AvailabilityProbeStatus) AvailabilityProbeStatus {
+	if !status.ProbeReportReceivedAt.IsZero() && status.LastChecked.After(status.ProbeReportReceivedAt) {
+		status.LastChecked = status.ProbeReportReceivedAt
+	}
+	status.ProbeAgentID = ""
+	status.ProbeReportReceivedAt = time.Time{}
+	return status
+}
+
 func (m *Monitor) RefreshAvailabilityTargets() {
 	if m == nil {
 		return
@@ -986,6 +1011,12 @@ func availabilityEvidenceEnvelope(
 	}
 
 	freshnessAt := status.FreshnessTime()
+	if reachableAt := availabilityReachablePathFreshness(status); !reachableAt.IsZero() {
+		// An available check's claim rests on its reachable paths alone, so
+		// its evidence lasts as long as the newest of them, not as long as a
+		// fresher path that failed.
+		freshnessAt = reachableAt
+	}
 	if freshnessAt.IsZero() {
 		freshnessAt = observedAt
 	}
@@ -1028,6 +1059,22 @@ func availabilityEvidenceEnvelope(
 	return &envelope
 }
 
+// availabilityReachablePathFreshness returns the newest server-authored
+// freshness among the current paths that reached an available check observed
+// from several locations, and zero otherwise.
+func availabilityReachablePathFreshness(status AvailabilityProbeStatus) time.Time {
+	if !status.Available || len(status.Locations) < 2 {
+		return time.Time{}
+	}
+	var latest time.Time
+	for _, location := range status.Locations {
+		if location.Available && !location.Stale && location.FreshnessAt.After(latest) {
+			latest = location.FreshnessAt
+		}
+	}
+	return latest
+}
+
 func availabilityResourceTags(target config.AvailabilityTarget) []string {
 	tags := []string{"agentless"}
 	if target.TargetKind != "" {

internal/monitoring/availability_probe_agent.go

@@ -23,6 +23,10 @@ const availabilityProbeStaleFloor = 5 * time.Minute
 // assigned agent stops reporting.
 const availabilityProbeStaleError = "no recent report from probe agent"
 
+// availabilityLocalCheckStaleError explains a local observation that stopped
+// running while the check's other locations kept reporting.
+const availabilityLocalCheckStaleError = "no recent local check"
+
 // ProbeAvailabilityResult is one availability observation reported by a remote
 // host agent that owns the target's execution.
 type ProbeAvailabilityResult struct {
@@ -316,9 +320,34 @@ func (m *Monitor) availabilityProbeAssignmentReference(targetID, locationID stri
 	return now
 }
 
+// deriveLocalAvailabilityObservationStaleness ages out a local observation
+// that stopped running when the check also has other locations. Alone, a
+// local observation's check time is the check's freshness, so its evidence
+// lapses with it. Beside other locations the check takes the newest
+// location's freshness, so a local pass that stopped would otherwise keep the
+// check available on the strength of a failing remote report.
+func deriveLocalAvailabilityObservationStaleness(
+	target config.AvailabilityTarget,
+	status AvailabilityProbeStatus,
+	now time.Time,
+) AvailabilityProbeStatus {
+	if status.LastChecked.IsZero() || !availabilityProbeReportIsStale(target, status.LastChecked, now) {
+		return status
+	}
+	status.Outcome = string(AvailabilityProbeIndeterminate)
+	status.Available = false
+	status.LastError = availabilityLocalCheckStaleError
+	status.LatencyMillis = 0
+	return status
+}
+
 func availabilityProbeStatusIsStale(status AvailabilityProbeStatus) bool {
-	return status.Outcome == string(AvailabilityProbeIndeterminate) &&
-		strings.EqualFold(strings.TrimSpace(status.LastError), availabilityProbeStaleError)
+	if status.Outcome != string(AvailabilityProbeIndeterminate) {
+		return false
+	}
+	lastError := strings.TrimSpace(status.LastError)
+	return strings.EqualFold(lastError, availabilityProbeStaleError) ||
+		strings.EqualFold(lastError, availabilityLocalCheckStaleError)
 }
 
 func availabilityProbeReportIsStale(target config.AvailabilityTarget, lastChecked time.Time, now time.Time) bool {

internal/monitoring/availability_probe_agent_test.go

@@ -165,6 +165,91 @@ func TestMultiLocationAvailabilityTreatsLapsedPathAsUnknownCoverage(t *testing.T
 	}
 }
 
+// The local twin of the lapsed-path case. A local observation that stopped
+// running keeps its last outcome, and a check with several locations takes the
+// newest location's freshness, so a failing edge report used to keep vouching
+// for a local pass from an hour ago: the check read available with current
+// evidence, which the registry takes as proof that its target answers.
+func TestMultiLocationAvailabilityAgesOutALocalPathThatStopped(t *testing.T) {
+	target := config.NormalizeAvailabilityTarget(config.AvailabilityTarget{
+		ID:                     "service",
+		Name:                   "Customer API",
+		Address:                "api.service.local",
+		Protocol:               config.AvailabilityProbeHTTPS,
+		Enabled:                true,
+		PollIntervalSecs:       60,
+		FailureThreshold:       2,
+		ObservationLocationIDs: []string{config.AvailabilityObservationLocationLocal, config.AvailabilityAgentObservationLocationID("edge-1")},
+	})
+	monitor := newProbeAgentTestMonitor(t, target)
+	monitor.SetLicenseChecker(licenseWithExternalProbe(true))
+	now := time.Now().UTC()
+	old := now.Add(-time.Hour)
+	monitor.applyAvailabilityObservation(target, "local-ok", old, 8*time.Millisecond, AvailabilityProbeReachable, nil, nil, "", time.Time{})
+	for attempt := 0; attempt < 2; attempt++ {
+		at := now.Add(time.Duration(attempt) * time.Second)
+		monitor.applyAvailabilityObservation(target, "edge-fail-"+string(rune('a'+attempt)), at, 30*time.Millisecond, AvailabilityProbeUnreachable, context.DeadlineExceeded, nil, "edge-1", at)
+	}
+
+	readAt := now.Add(2 * time.Second)
+	status := monitor.availabilityStatusSnapshotForTargets([]config.AvailabilityTarget{target}, readAt)[target.ID]
+	if status.Available || status.AggregateState != AvailabilityAggregateUnknown {
+		t.Fatalf("aggregate status = %+v, want unknown coverage rather than an available check", status)
+	}
+	if status.ReportingLocations != 1 || status.ExpectedLocations != 2 {
+		t.Fatalf("coverage = %d/%d, want the stopped local path excluded", status.ReportingLocations, status.ExpectedLocations)
+	}
+	local := status.Locations[0]
+	if local.LocationID != config.AvailabilityObservationLocationLocal || !local.Stale ||
+		local.Outcome != string(AvailabilityProbeIndeterminate) || local.LastError != availabilityLocalCheckStaleError {
+		t.Fatalf("local location = %+v, want stale unknown evidence", local)
+	}
+	resource, _ := availabilityResourceFromTarget(target, status, "", readAt)
+	if resource.Availability.Available || resource.Status != unifiedresources.StatusWarning || len(resource.Incidents) != 0 {
+		t.Fatalf("resource = available %v status %q incidents %+v, want an unproven check and no outage", resource.Availability.Available, resource.Status, resource.Incidents)
+	}
+
+	monitor.applyAvailabilityObservation(target, "local-ok-again", readAt, 9*time.Millisecond, AvailabilityProbeReachable, nil, nil, "", time.Time{})
+	status = monitor.availabilityStatusSnapshotForTargets([]config.AvailabilityTarget{target}, readAt)[target.ID]
+	if !status.Available || status.AggregateState != AvailabilityAggregateDegraded || status.Locations[0].Stale {
+		t.Fatalf("aggregate status = %+v, want a current local pass to make the check available again", status)
+	}
+}
+
+// An available check's evidence lasts as long as its newest reachable path. A
+// registry that re-reads this generation later judges the check at its own
+// time, so evidence anchored on a fresher failing path would keep proving the
+// target answers after the only passing path had aged out.
+func TestMultiLocationAvailableEvidenceLastsAsLongAsItsReachablePath(t *testing.T) {
+	target := config.NormalizeAvailabilityTarget(config.AvailabilityTarget{
+		ID:                     "service",
+		Name:                   "Customer API",
+		Address:                "api.service.local",
+		Protocol:               config.AvailabilityProbeHTTPS,
+		Enabled:                true,
+		PollIntervalSecs:       60,
+		FailureThreshold:       2,
+		ObservationLocationIDs: []string{config.AvailabilityObservationLocationLocal, config.AvailabilityAgentObservationLocationID("edge-1")},
+	})
+	monitor := newProbeAgentTestMonitor(t, target)
+	monitor.SetLicenseChecker(licenseWithExternalProbe(true))
+	now := time.Now().UTC()
+	passedAt := now.Add(-4 * time.Minute)
+	monitor.applyAvailabilityObservation(target, "local-ok", passedAt, 8*time.Millisecond, AvailabilityProbeReachable, nil, nil, "", time.Time{})
+	monitor.applyAvailabilityObservation(target, "edge-fail", now, 30*time.Millisecond, AvailabilityProbeUnreachable, context.DeadlineExceeded, nil, "edge-1", now)
+
+	status := monitor.availabilityStatusSnapshotForTargets([]config.AvailabilityTarget{target}, now)[target.ID]
+	if !status.Available || status.AggregateState != AvailabilityAggregateDegraded {
+		t.Fatalf("aggregate status = %+v, want an available check with one failing path", status)
+	}
+	resource, _ := availabilityResourceFromTarget(target, status, "", now)
+	evidence := resource.Availability.Evidence
+	want := passedAt.Add(availabilityProbeStaleWindow(target))
+	if evidence == nil || evidence.ValidUntil == nil || !evidence.ValidUntil.Equal(want) {
+		t.Fatalf("evidence = %+v, want it valid until %s, one window after the reachable path", evidence, want)
+	}
+}
+
 func licenseWithExternalProbe(enabled bool) func(string) bool {
 	return func(feature string) bool {
 		return enabled && feature == pkglicensing.FeatureExternalProbe
@@ -237,6 +322,74 @@ func TestAvailabilityPollProviderSkipsProbeAssignedTargetsAndResumesOnLapse(t *t
 	}
 }
 
+// When the entitlement lapses, a remote check's last result stands in as the
+// local location's until the local poller runs. A local location's check time
+// is its freshness, so the agent's clock must not carry over: a result checked
+// a day ahead would otherwise read current, and keep its target answering, for
+// a day without a local execution.
+func TestAvailabilityLapseAdoptsARemoteResultAtItsReceiptTime(t *testing.T) {
+	target := config.NormalizeAvailabilityTarget(probeAgentTarget("remote", "agent-1"))
+	monitor := newProbeAgentTestMonitor(t, target)
+	monitor.SetLicenseChecker(licenseWithExternalProbe(true))
+	now := time.Now().UTC()
+	receivedAt := now.Add(-time.Minute)
+	agentClock := now.Add(24 * time.Hour)
+	monitor.applyAvailabilityObservation(target, "remote-ok", agentClock, 8*time.Millisecond, AvailabilityProbeReachable, nil, nil, "agent-1", receivedAt)
+
+	monitor.SetLicenseChecker(licenseWithExternalProbe(false))
+	status := monitor.availabilityStatusSnapshotForTargets([]config.AvailabilityTarget{target}, now)[target.ID]
+	if status.ProbeAgentID != "" || !status.LastChecked.Equal(receivedAt) {
+		t.Fatalf("adopted status = %+v, want a local result checked at the server receipt time", status)
+	}
+	resource, _ := availabilityResourceFromTarget(target, status, "", now)
+	evidence := resource.Availability.Evidence
+	want := receivedAt.Add(2 * time.Duration(target.EffectivePollIntervalSecs()) * time.Second)
+	if evidence == nil || evidence.ValidUntil == nil || !evidence.ValidUntil.Equal(want) {
+		t.Fatalf("evidence = %+v, want it valid until %s", evidence, want)
+	}
+}
+
+// The stored result of a check with several locations is an aggregate, whose
+// freshness may be one path's and whose availability another's. After a lapse
+// it must not stand in for the local location: one path passing an hour ago
+// and another failing now would read as a current local pass.
+func TestAvailabilityLapseNeverAdoptsAMultiLocationAggregate(t *testing.T) {
+	target := config.NormalizeAvailabilityTarget(config.AvailabilityTarget{
+		ID:               "service",
+		Name:             "Customer API",
+		Address:          "api.service.local",
+		Protocol:         config.AvailabilityProbeHTTPS,
+		Enabled:          true,
+		PollIntervalSecs: 60,
+		FailureThreshold: 2,
+		ObservationLocationIDs: []string{
+			config.AvailabilityAgentObservationLocationID("edge-1"),
+			config.AvailabilityAgentObservationLocationID("edge-2"),
+		},
+	})
+	monitor := newProbeAgentTestMonitor(t, target)
+	monitor.SetLicenseChecker(licenseWithExternalProbe(true))
+	now := time.Now().UTC()
+	passedAt := now.Add(-time.Hour)
+	monitor.applyAvailabilityObservation(target, "edge-1-ok", passedAt, 8*time.Millisecond, AvailabilityProbeReachable, nil, nil, "edge-1", passedAt)
+	monitor.applyAvailabilityObservation(target, "edge-2-fail", now, 30*time.Millisecond, AvailabilityProbeUnreachable, context.DeadlineExceeded, nil, "edge-2", now)
+
+	monitor.SetLicenseChecker(licenseWithExternalProbe(false))
+	status, ok := monitor.availabilityStatusSnapshotForTargets([]config.AvailabilityTarget{target}, now)[target.ID]
+	if !ok || status.Available || !status.LastChecked.IsZero() || status.AggregateState != "" {
+		t.Fatalf("status after the lapse = %+v (present %v), want a never-checked target", status, ok)
+	}
+	if resource, _ := availabilityResourceFromTarget(target, status, "", now); resource.Status != unifiedresources.StatusUnknown {
+		t.Fatalf("check row status = %q, want unknown like any never-checked target", resource.Status)
+	}
+	// An absent entry would let connection health fall back to the last
+	// cached state, which still says connected.
+	connected, reported := availabilityPollProvider{}.ConnectionStatuses(monitor)[availabilityConnectionKey(target.ID)]
+	if !reported || connected {
+		t.Fatalf("connection status = %v (reported %v), want not connected", connected, reported)
+	}
+}
+
 func TestRefreshAvailabilityTargetsUnschedulesProbeAssignedTargets(t *testing.T) {
 	monitor := newProbeAgentTestMonitor(t,
 		probeAgentTarget("remote", "agent-1"),

internal/unifiedresources/availability.go

@@ -4,6 +4,9 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust"
 )
 
 // AvailabilityChecksForResource returns the complete canonical availability
@@ -28,6 +31,25 @@ func AvailabilityCheckByTargetID(resource Resource, targetID string) *Availabili
 	return nil
 }
 
+// availabilityChecksProveOnline reports whether any availability check
+// projected onto a monitored resource proves, at now, that the resource
+// answers. Only a passing check with current evidence does. Each check is
+// judged by its own evidence window (two poll intervals for a local check,
+// the report window for a remote probe), so one check's recent run cannot
+// vouch for another check's old pass, and a local check that missed its
+// cadence keeps Available but proves nothing. A failing check proves only
+// that one port or service does not answer, so it never decides the
+// resource's status at all: not online, offline or warning, current or quiet.
+func availabilityChecksProveOnline(checks []AvailabilityData, now time.Time) bool {
+	for _, check := range checks {
+		if check.Enabled && check.Available && check.Evidence != nil &&
+			check.Evidence.FreshnessAt(now) == operationaltrust.EvidenceFresh {
+			return true
+		}
+	}
+	return false
+}
+
 func normalizeResourceAvailability(resource *Resource) {
 	if resource == nil {
 		return

internal/unifiedresources/registry.go

@@ -1805,16 +1805,16 @@ func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSo
 				changed = changed || status.Status != "stale"
 				status.Status = "stale"
 				resource.SourceStatus[source] = status
-				staleFound = true
+				// The sighting of checks projected onto a resource is not one
+				// of its own sources, so its going quiet alone hands the stale
+				// pass nothing to decide; see reapplyStaleStatusForChecks.
+				if source != SourceAvailability || isAvailabilityOwnedResource(*resource) {
+					staleFound = true
+				}
 			}
 		}
 		if staleFound {
-			recomputed := aggregateStatus(resource)
-			if recomputed != StatusUnknown {
-				resource.Status = recomputed
-			} else if resource.Status == StatusOnline {
-				resource.Status = StatusWarning
-			}
+			applyStaleStatus(resource, now)
 		}
 		changed = changed || resource.Status != previousStatus
 	}
@@ -1823,6 +1823,34 @@ func (rr *ResourceRegistry) markStaleLocked(now time.Time, thresholds map[DataSo
 	}
 }
 
+// applyStaleStatus is the stale pass's status rule for a resource with a
+// quiet sighting.
+func applyStaleStatus(resource *Resource, now time.Time) {
+	recomputed := aggregateStatus(resource, now)
+	if recomputed != StatusUnknown {
+		resource.Status = recomputed
+	} else if resource.Status == StatusOnline {
+		resource.Status = StatusWarning
+	}
+}
+
+// reapplyStaleStatusForChecks re-applies the stale pass's rule to a resource
+// whose availability checks just changed, when that pass owns its status
+// because one of the resource's own sources went quiet. Projecting or
+// retargeting a check is not a delivery from those sources, and the resources
+// API replays checks after its stale pass, so the verdict the old checks gave
+// would otherwise stand. The checks' own sighting never confers ownership:
+// a resource whose own sources have not gone quiet keeps the status they gave
+// it, whatever its checks say.
+func reapplyStaleStatusForChecks(resource *Resource, now time.Time) {
+	for source, sighting := range resource.SourceStatus {
+		if source != SourceAvailability && sighting.Status == "stale" {
+			applyStaleStatus(resource, now)
+			return
+		}
+	}
+}
+
 func (rr *ResourceRegistry) ingestProxmoxNode(node models.Node, linkedHost *models.Host) {
 	resource, identity := resourceFromProxmoxNode(node, linkedHost)
 	rr.ingest(SourceProxmox, node.ID, resource, identity)
@@ -3046,6 +3074,7 @@ func (rr *ResourceRegistry) projectAvailabilityCheckLocked(
 		Status:   sourceSightingStatus(checkResource.LastSeen),
 		LastSeen: checkResource.LastSeen,
 	}
+	reapplyStaleStatusForChecks(target, time.Now().UTC())
 }
 
 func (rr *ResourceRegistry) removeAvailabilityProjectionLocked(
@@ -3080,6 +3109,7 @@ func (rr *ResourceRegistry) removeAvailabilityProjectionLocked(
 				resource.Sources = removeDataSource(resource.Sources, SourceAvailability)
 				delete(resource.SourceStatus, SourceAvailability)
 			}
+			reapplyStaleStatusForChecks(resource, time.Now().UTC())
 		}
 
 		// Build a fresh slice rather than compacting in place: callers may hold
@@ -4784,7 +4814,7 @@ func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource
 	// Manual links combine already-normalized resources. Preserve each metric's
 	// recorded source instead of flattening the linked resource to SourceAgent.
 	primary.Metrics = mergeMetrics(primary, primary.Metrics, other.Metrics, "", time.Now().UTC(), primary.SourceStatus, thresholds)
-	primary.Status = aggregateStatus(primary)
+	primary.Status = aggregateStatus(primary, time.Now().UTC())
 }
 
 func (rr *ResourceRegistry) updateSourceMappings(fromID, toID string) {
@@ -6372,15 +6402,25 @@ func chooseStatus(existing ResourceStatus, incoming ResourceStatus, source DataS
 // order: the highest-priority verdict, the best of equal ones. A source that
 // went quiet drops out of that decision, so a node the cluster reports
 // offline stays offline when its linked agent falls silent. A current facet
-// sighting without a verdict (the PBS association, an availability check)
-// counts as online only when no current source has one. Once every source is
-// quiet, an offline verdict survives (a node the poller expired, an agent past
-// its lease) and any other reads as warning; the best of those wins.
-func aggregateStatus(resource *Resource) ResourceStatus {
+// sighting without a verdict (the PBS association) counts as online only when
+// no current source has one. Availability checks projected onto a monitored
+// resource rank the same way, but their sighting is not their verdict: one
+// sighting stands for every check, so each check is judged at now by its own
+// evidence (see availabilityChecksProveOnline), and a check that does not
+// prove the resource answers abstains. Checks only decide once one of the
+// resource's own sources has gone quiet, so the stale pass owns the status and
+// reverts it when the evidence lapses. Once every source is quiet, an
+// offline verdict survives (a node the poller expired, an agent past its
+// lease) and any other reads as warning; the best of those wins.
+func aggregateStatus(resource *Resource, now time.Time) ResourceStatus {
 	var current, quiet ResourceStatus
 	currentPriority := -1
 	deliveredWithoutVerdict := false
+	checkedTarget := !isAvailabilityOwnedResource(*resource)
 	for source, sighting := range resource.SourceStatus {
+		if source == SourceAvailability && checkedTarget {
+			continue
+		}
 		switch strings.ToLower(strings.TrimSpace(sighting.Status)) {
 		case "online":
 			switch sighting.reported {
@@ -6408,6 +6448,8 @@ func aggregateStatus(resource *Resource) ResourceStatus {
 		return current
 	case deliveredWithoutVerdict:
 		return StatusOnline
+	case checkedTarget && quiet != "" && availabilityChecksProveOnline(AvailabilityChecksForResource(*resource), now):
+		return StatusOnline
 	case quiet != "":
 		return quiet
 	default:

internal/unifiedresources/registry_merge_policy_test.go

@@ -712,18 +712,30 @@ func TestIngestResourcesKeepsAFacetSightingWithoutAVerdict(t *testing.T) {
 
 // aggregateStatus is the one rule the stale pass and manual links apply.
 // Current sources decide by their own verdicts in chooseStatus's priority
-// order; quiet sources only decide when no source is current.
+// order; quiet sources only decide when no source is current. Availability
+// checks on a monitored resource are judged by their own evidence, not their
+// shared sighting.
 func TestAggregateStatusReadsSourceVerdictsNotDelivery(t *testing.T) {
+	now := time.Now().UTC()
+	lapsed := now.Add(-10 * time.Minute)
 	current := func(reported ResourceStatus) SourceStatus {
 		return SourceStatus{Status: "online", reported: reported}
 	}
 	quiet := func(reported ResourceStatus) SourceStatus {
 		return SourceStatus{Status: "stale", reported: reported}
 	}
+	check := func(passing bool, checkedAt time.Time) []AvailabilityData {
+		return []AvailabilityData{{
+			TargetID: "probe-1", Enabled: true, Available: passing, LastChecked: &checkedAt,
+			Evidence: availabilityProbeEvidence(t, "probe-1", checkedAt),
+		}}
+	}
 	for _, tc := range []struct {
-		name      string
-		sightings map[DataSource]SourceStatus
-		want      ResourceStatus
+		name         string
+		resourceType ResourceType
+		sightings    map[DataSource]SourceStatus
+		checks       []AvailabilityData
+		want         ResourceStatus
 	}{
 		{
 			name:      "live poll reports offline, agent quiet",
@@ -767,9 +779,57 @@ func TestAggregateStatusReadsSourceVerdictsNotDelivery(t *testing.T) {
 		},
 		{
 			name:      "a facet without a verdict never outranks a current verdict",
-			sightings: map[DataSource]SourceStatus{SourceProxmox: current(StatusOffline), SourceAvailability: current("")},
+			sightings: map[DataSource]SourceStatus{SourceProxmox: current(StatusOffline), SourcePBS: current("")},
 			want:      StatusOffline,
 		},
+		{
+			name:      "a current failing check abstains",
+			sightings: map[DataSource]SourceStatus{SourceProxmox: quiet(StatusOffline), SourceAvailability: current("")},
+			checks:    check(false, now),
+			want:      StatusOffline,
+		},
+		{
+			name:      "a quiet failing check abstains rather than lifting an offline verdict to warning",
+			sightings: map[DataSource]SourceStatus{SourceProxmox: quiet(StatusOffline), SourceAvailability: quiet("")},
+			checks:    check(false, lapsed),
+			want:      StatusOffline,
+		},
+		{
+			name:      "a passing check whose evidence lapsed abstains under a current sighting",
+			sightings: map[DataSource]SourceStatus{SourceProxmox: quiet(StatusOnline), SourceAvailability: current("")},
+			checks:    check(true, lapsed),
+			want:      StatusWarning,
+		},
+		{
+			name:      "a passing check with current evidence proves the target answers under a quiet sighting",
+			sightings: map[DataSource]SourceStatus{SourceProxmox: quiet(StatusOffline), SourceAvailability: quiet("")},
+			checks:    check(true, now),
+			want:      StatusOnline,
+		},
+		{
+			name:      "a passing check never lifts a resource whose own sources have not gone quiet",
+			sightings: map[DataSource]SourceStatus{SourceProxmox: {Status: "unknown"}, SourceAvailability: current("")},
+			checks:    check(true, now),
+			want:      StatusUnknown,
+		},
+		{
+			name:      "a passing check never outranks a current verdict",
+			sightings: map[DataSource]SourceStatus{SourceProxmox: current(StatusOffline), SourceAvailability: current("")},
+			checks:    check(true, now),
+			want:      StatusOffline,
+		},
+		{
+			name:         "a check's own row keeps its quiet verdict",
+			resourceType: ResourceTypeNetworkEndpoint,
+			sightings:    map[DataSource]SourceStatus{SourceAvailability: quiet(StatusOffline)},
+			want:         StatusOffline,
+		},
+		{
+			name:         "a check's own row keeps its current verdict",
+			resourceType: ResourceTypeNetworkEndpoint,
+			sightings:    map[DataSource]SourceStatus{SourceAvailability: current(StatusWarning)},
+			want:         StatusWarning,
+		},
 		{
 			name:      "current unknown verdict leaves the quiet sources to decide",
 			sightings: map[DataSource]SourceStatus{SourceProxmox: current(StatusUnknown), SourceAgent: quiet(StatusOnline)},
@@ -782,7 +842,8 @@ func TestAggregateStatusReadsSourceVerdictsNotDelivery(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			if got := aggregateStatus(&Resource{SourceStatus: tc.sightings}); got != tc.want {
+			resource := &Resource{Type: tc.resourceType, SourceStatus: tc.sightings, AvailabilityChecks: tc.checks}
+			if got := aggregateStatus(resource, now); got != tc.want {
 				t.Fatalf("aggregateStatus = %q, want %q", got, tc.want)
 			}
 		})

internal/unifiedresources/registry_test.go

@@ -5686,7 +5686,8 @@ func TestMarkStaleRecomputesFromRemainingFreshSources(t *testing.T) {
 			Availability: &AvailabilityData{
 				TargetID: "probe-delly", LinkedResourceID: host.ID,
 				Address: "192.0.2.5", Protocol: "tcp", Port: 8006,
-				Enabled: true, Available: true,
+				Enabled: true, Available: true, LastChecked: &recentNow,
+				Evidence: availabilityProbeEvidence(t, "probe-delly", recentNow),
 			},
 		},
 		Identity: ResourceIdentity{IPAddresses: []string{"192.0.2.5"}},
@@ -5759,6 +5760,7 @@ func TestMarkStaleKeepsProxmoxRuntimeStateIndependentOfAvailabilityFacet(t *test
 			Availability: &AvailabilityData{
 				TargetID: "probe-102", LinkedResourceID: checkedID,
 				Address: "192.0.2.102", Protocol: "icmp", Enabled: true, Available: true,
+				LastChecked: &freshSeen, Evidence: availabilityProbeEvidence(t, "probe-102", freshSeen),
 			},
 		},
 	}})
@@ -5781,6 +5783,305 @@ func TestMarkStaleKeepsProxmoxRuntimeStateIndependentOfAvailabilityFacet(t *test
 	}
 }
 
+// availabilityCheckRecord is a check the poller ran at checkedAt against the
+// resource targetID names. Its evidence stays current for two minutes, the
+// two poll intervals of a local check; Available stays as the last run left
+// it, as a local check that misses its cadence does.
+func availabilityCheckRecord(t *testing.T, checkID, targetID string, passing bool, checkedAt time.Time) IngestRecord {
+	t.Helper()
+	status, failures := StatusOnline, 0
+	if !passing {
+		status, failures = StatusOffline, 3
+	}
+	return IngestRecord{
+		SourceID: checkID,
+		Resource: Resource{
+			Type: ResourceTypeNetworkEndpoint, Name: checkID, Status: status, LastSeen: checkedAt,
+			Sources: []DataSource{SourceAvailability},
+			Availability: &AvailabilityData{
+				TargetID: checkID, LinkedResourceID: targetID,
+				Address: "192.0.2.80", Protocol: "http", Port: 443,
+				Enabled: true, Available: passing, LastChecked: &checkedAt,
+				ConsecutiveFailures: failures, FailureThreshold: 2, PollIntervalSeconds: 60,
+				Evidence: availabilityProbeEvidence(t, checkID, checkedAt),
+			},
+		},
+	}
+}
+
+func ingestProxmoxGuestSeenAt(t *testing.T, rr *ResourceRegistry, sourceID string, seen time.Time) string {
+	t.Helper()
+	rr.IngestRecords(SourceProxmox, []IngestRecord{{
+		SourceID: sourceID,
+		Resource: Resource{
+			Type: ResourceTypeSystemContainer, Name: sourceID, Status: StatusOnline, LastSeen: seen,
+			Proxmox: &ProxmoxData{RuntimeStatus: "running", NodeName: "node-a"},
+		},
+	}})
+	for _, resource := range rr.ListByType(ResourceTypeSystemContainer) {
+		if resource.Name == sourceID {
+			return resource.ID
+		}
+	}
+	t.Fatalf("guest %s not ingested", sourceID)
+	return ""
+}
+
+// A failing availability check proves only that one port or service does not
+// answer, so it never decides its target's status: a node the poller expired
+// stays offline, a guest whose poll went quiet stays a warning instead of
+// counting as stopped, and neither reads online. Only a passing check with
+// current evidence proves the target answers. The stale pass used to count
+// any current check as online, so a failing probe kept an expired node up.
+func TestStalePassReadsAvailabilityCheckVerdicts(t *testing.T) {
+	now := time.Now().UTC()
+	quietSeen := now.Add(-5 * time.Minute)
+	lapsed := now.Add(-10 * time.Minute)
+	expiredNode := models.StateSnapshot{Nodes: []models.Node{{
+		ID: "homelab-pve9", Name: "pve9", Instance: "homelab", ClusterName: "homelab",
+		Status: "offline", LastSeen: quietSeen,
+	}}}
+	quietGuest := models.StateSnapshot{VMs: []models.VM{{
+		ID: "homelab:pve9:101", Name: "web", Node: "pve9", Instance: "homelab",
+		VMID: 101, Status: "running", Type: "qemu", LastSeen: quietSeen,
+	}}}
+	for _, tc := range []struct {
+		name       string
+		snapshot   models.StateSnapshot
+		targetType ResourceType
+		passing    bool
+		checkedAt  time.Time
+		want       ResourceStatus
+	}{
+		{"expired node, current failing check", expiredNode, ResourceTypeAgent, false, now, StatusOffline},
+		{"expired node, lapsed failing check", expiredNode, ResourceTypeAgent, false, lapsed, StatusOffline},
+		{"expired node, lapsed passing check", expiredNode, ResourceTypeAgent, true, lapsed, StatusOffline},
+		{"expired node, current passing check", expiredNode, ResourceTypeAgent, true, now, StatusOnline},
+		{"quiet guest, current failing check", quietGuest, ResourceTypeVM, false, now, StatusWarning},
+		{"quiet guest, lapsed failing check", quietGuest, ResourceTypeVM, false, lapsed, StatusWarning},
+		{"quiet guest, current passing check", quietGuest, ResourceTypeVM, true, now, StatusOnline},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			rr := NewRegistry(nil)
+			rr.IngestSnapshot(tc.snapshot)
+			target := onlyResourceOfType(t, rr, tc.targetType)
+			rr.IngestRecords(SourceAvailability, []IngestRecord{
+				availabilityCheckRecord(t, "probe-1", target.ID, tc.passing, tc.checkedAt),
+			})
+			rr.MarkStale(now, nil)
+
+			got, ok := rr.Get(target.ID)
+			if !ok || len(AvailabilityChecksForResource(*got)) != 1 {
+				t.Fatalf("check not projected onto %s: %+v", target.ID, got.AvailabilityChecks)
+			}
+			if got.Status != tc.want {
+				t.Fatalf("status = %q, want %q", got.Status, tc.want)
+			}
+		})
+	}
+}
+
+// One availability sighting stands for every check on a target, and it
+// carries whichever check was projected last. Each check is judged by its own
+// evidence instead, so a failing check that just ran cannot lend its
+// freshness to another check's old pass, in either projection order.
+func TestAvailabilityChecksAreJudgedByTheirOwnEvidence(t *testing.T) {
+	now := time.Now().UTC()
+	lapsed := now.Add(-10 * time.Minute)
+	for _, tc := range []struct {
+		name               string
+		passedAt, failedAt time.Time
+		want               ResourceStatus
+	}{
+		{name: "lapsed pass, current failure", passedAt: lapsed, failedAt: now, want: StatusWarning},
+		{name: "current pass, lapsed failure", passedAt: now, failedAt: lapsed, want: StatusOnline},
+	} {
+		for _, order := range [][]string{{"pass", "fail"}, {"fail", "pass"}} {
+			t.Run(tc.name+", "+strings.Join(order, " then "), func(t *testing.T) {
+				rr := NewRegistry(nil)
+				target := ingestProxmoxGuestSeenAt(t, rr, "lab:node-a:101", now.Add(-5*time.Minute))
+				records := map[string]IngestRecord{
+					"pass": availabilityCheckRecord(t, "probe-pass", target, true, tc.passedAt),
+					"fail": availabilityCheckRecord(t, "probe-fail", target, false, tc.failedAt),
+				}
+				for _, key := range order {
+					rr.IngestRecords(SourceAvailability, []IngestRecord{records[key]})
+				}
+				rr.MarkStale(now, nil)
+
+				got, ok := rr.Get(target)
+				if !ok {
+					t.Fatal("target missing")
+				}
+				if checks := AvailabilityChecksForResource(*got); len(checks) != 2 {
+					t.Fatalf("expected both checks on the target, got %+v", checks)
+				}
+				if got.Status != tc.want {
+					t.Fatalf("status = %q, want %q", got.Status, tc.want)
+				}
+			})
+		}
+	}
+}
+
+// A check's verdict moves with the check. Projecting, re-running or
+// retargeting a check is not a delivery from the resource's own sources, so a
+// target whose status the stale pass owns is judged again at once rather than
+// keeping the old checks' verdict until the next pass; the resources API
+// replays checks after its stale pass and never runs another.
+func TestChangedAvailabilityChecksJudgeTheirTargetsAgain(t *testing.T) {
+	now := time.Now().UTC()
+	quietSeen := now.Add(-5 * time.Minute)
+	rr := NewRegistry(nil)
+	first := ingestProxmoxGuestSeenAt(t, rr, "lab:node-a:101", quietSeen)
+	second := ingestProxmoxGuestSeenAt(t, rr, "lab:node-a:102", quietSeen)
+	rr.MarkStale(now, nil)
+
+	status := func(id string) ResourceStatus {
+		t.Helper()
+		resource, ok := rr.Get(id)
+		if !ok {
+			t.Fatalf("resource %s missing", id)
+		}
+		return resource.Status
+	}
+
+	rr.IngestRecords(SourceAvailability, []IngestRecord{
+		availabilityCheckRecord(t, "probe-moving", first, true, now),
+		availabilityCheckRecord(t, "probe-failing", first, false, now),
+	})
+	if got := status(first); got != StatusOnline {
+		t.Fatalf("first status = %q once a passing check lands, want online", got)
+	}
+
+	rr.IngestRecords(SourceAvailability, []IngestRecord{
+		availabilityCheckRecord(t, "probe-moving", second, true, now),
+	})
+	left, _ := rr.Get(first)
+	if checks := AvailabilityChecksForResource(*left); len(checks) != 1 || checks[0].TargetID != "probe-failing" {
+		t.Fatalf("first checks = %+v, want only probe-failing", checks)
+	}
+	if got := status(first); got != StatusWarning {
+		t.Fatalf("first status = %q after its passing check moved away, want warning", got)
+	}
+	if got := status(second); got != StatusOnline {
+		t.Fatalf("second status = %q after the passing check moved onto it, want online", got)
+	}
+
+	rr.IngestRecords(SourceAvailability, []IngestRecord{
+		availabilityCheckRecord(t, "probe-moving", second, false, now),
+	})
+	if got := status(second); got != StatusWarning {
+		t.Fatalf("second status = %q once its only check fails, want warning", got)
+	}
+
+	rr.MarkStale(now, nil)
+	if got := status(first); got != StatusWarning {
+		t.Fatalf("first status = %q after the next pass, want warning", got)
+	}
+	if got := status(second); got != StatusWarning {
+		t.Fatalf("second status = %q after the next pass, want warning", got)
+	}
+}
+
+// A projected check's sighting is not one of its target's own sources. Its
+// going quiet hands the stale pass nothing to decide, and replacing the check
+// takes no ownership away, so a target whose own source has never delivered
+// keeps the status that source gave it whatever its checks say.
+func TestAvailabilitySightingNeverHandsTheStalePassItsTarget(t *testing.T) {
+	now := time.Now().UTC()
+	quickChecks := map[DataSource]time.Duration{SourceAvailability: 30 * time.Second}
+	for _, tc := range []struct {
+		name     string
+		ingested ResourceStatus
+		passing  bool
+	}{
+		{name: "online guest, quiet failing check", ingested: StatusOnline, passing: false},
+		{name: "offline placeholder, quiet passing check", ingested: StatusOffline, passing: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			rr := NewRegistry(nil)
+			rr.IngestRecords(SourceProxmox, []IngestRecord{{
+				SourceID: "lab:node-a:101",
+				Resource: Resource{
+					Type: ResourceTypeSystemContainer, Name: "lab:node-a:101", Status: tc.ingested,
+					Proxmox: &ProxmoxData{RuntimeStatus: "running", NodeName: "node-a"},
+				},
+			}})
+			target := onlyResourceOfType(t, rr, ResourceTypeSystemContainer).ID
+
+			rr.IngestRecords(SourceAvailability, []IngestRecord{
+				availabilityCheckRecord(t, "probe-1", target, tc.passing, now.Add(-time.Minute)),
+			})
+			rr.MarkStale(now, quickChecks)
+			got, _ := rr.Get(target)
+			if sighting := got.SourceStatus[SourceAvailability]; sighting.Status != "stale" {
+				t.Fatalf("availability sighting = %+v, want stale", sighting)
+			}
+			if got.Status != tc.ingested {
+				t.Fatalf("status = %q with a quiet check, want %q from the target's own source", got.Status, tc.ingested)
+			}
+
+			rr.IngestRecords(SourceAvailability, []IngestRecord{
+				availabilityCheckRecord(t, "probe-1", target, !tc.passing, now),
+			})
+			rr.MarkStale(now, quickChecks)
+			if got, _ := rr.Get(target); got.Status != tc.ingested {
+				t.Fatalf("status = %q after the check flipped, want %q", got.Status, tc.ingested)
+			}
+		})
+	}
+}
+
+// A manual link keeps the primary's own availability checks and not the
+// linked resource's, so the merged resource is judged by the checks it
+// carries. The linked resource's fresher availability sighting survives the
+// merge as delivery bookkeeping; it must not become a verdict for checks the
+// merged resource does not show.
+func TestManualLinkJudgesOnlyTheChecksTheMergedResourceCarries(t *testing.T) {
+	now := time.Now().UTC()
+	quietSeen := now.Add(-5 * time.Minute)
+	store := NewMemoryStore()
+	if err := store.AddLink(ResourceLink{ResourceA: "vm-web", ResourceB: "agent-web", PrimaryID: "vm-web"}); err != nil {
+		t.Fatalf("add link: %v", err)
+	}
+	rr := NewRegistry(store)
+	rr.IngestResources([]Resource{
+		{
+			ID: "vm-web", Type: ResourceTypeVM, Name: "web", Status: StatusWarning, LastSeen: quietSeen,
+			Sources: []DataSource{SourceProxmox},
+			SourceStatus: map[DataSource]SourceStatus{
+				SourceProxmox: {Status: "stale", LastSeen: quietSeen, reported: StatusOnline},
+			},
+		},
+		{
+			ID: "agent-web", Type: ResourceTypeAgent, Name: "web", Status: StatusWarning, LastSeen: now,
+			Sources: []DataSource{SourceAgent, SourceAvailability},
+			SourceStatus: map[DataSource]SourceStatus{
+				SourceAgent:        {Status: "stale", LastSeen: quietSeen, reported: StatusOnline},
+				SourceAvailability: {Status: "online", LastSeen: now},
+			},
+			AvailabilityChecks: []AvailabilityData{{
+				TargetID: "probe-web", Enabled: true, Available: false, LastChecked: &now,
+				ConsecutiveFailures: 3, FailureThreshold: 2,
+				Evidence: availabilityProbeEvidence(t, "probe-web", now),
+			}},
+			Agent: &AgentData{AgentID: "host-web", Hostname: "web"},
+		},
+	})
+
+	merged, ok := rr.Get("vm-web")
+	if !ok {
+		t.Fatal("linked resource missing")
+	}
+	if checks := AvailabilityChecksForResource(*merged); len(checks) != 0 {
+		t.Fatalf("merged resource exposes the linked resource's checks: %+v", checks)
+	}
+	if merged.Status != StatusWarning {
+		t.Fatalf("status = %q, want warning from its own quiet sources", merged.Status)
+	}
+}
+
 func TestResourceRegistryUsesConfiguredProxmoxStaleThresholds(t *testing.T) {
 	seen := time.Now().UTC().Add(-90 * time.Second).Truncate(time.Millisecond)
 	snapshot := models.StateSnapshot{