diff --git a/Makefile b/Makefile
index 73dfd1b..40f0da1 100644
--- a/Makefile
+++ b/Makefile
@@ -13,7 +13,7 @@ SAMPLE_IMAGE ?= rebuildstack-e2e-goworkload:latest
 # read the controller's spool (the agent image is distroless — no shell).
 SPOOL_READER_IMAGE ?= busybox:1.37
 
-.PHONY: build test lint tidy clean cluster-up cluster-down e2e smoke image sample-image kind-load node-e2e inventory-e2e restarts-e2e lifecycle-e2e policy-e2e profile-gate-e2e profile-capture-e2e
+.PHONY: build test lint tidy clean cluster-up cluster-down e2e smoke image sample-image kind-load node-e2e inventory-e2e restarts-e2e lifecycle-e2e policy-e2e watch-e2e profile-gate-e2e profile-capture-e2e
 
 build: ## Build the agent binary into bin/
 	go build -ldflags '$(LDFLAGS)' -o bin/agent ./cmd/agent
@@ -125,6 +125,18 @@ restarts-e2e: kind-load ## Deploy the controller in kind next to a crash-looping
 		-run 'TestContainerRestartsEndToEnd|TestRestartCountersCarryTheHistoryTheAgentDidNotWatch' 2>&1 \
 		| tee test/e2e/logs/restarts-e2e-$$(date +%Y%m%d-%H%M%S).log
 
+watch-e2e: kind-load ## Deny the pods grant, then revoke a policy grant from a running agent, and assert the first stops the agent while the second only degrades its payload (ADR 0035); slow by nature — an established watch is re-authorized only when client-go re-establishes it, every 5–10 minutes; log goes to test/e2e/logs/
+	@mkdir -p test/e2e/logs
+	docker pull $(SPOOL_READER_IMAGE)
+	go tool kind load docker-image $(SPOOL_READER_IMAGE) --name $(E2E_CLUSTER)
+	set -o pipefail; \
+	E2E_KUBE_CONTEXT=kind-$(E2E_CLUSTER) \
+	E2E_AGENT_IMAGE=$(IMAGE):$(IMAGE_TAG) \
+	E2E_SPOOL_READER_IMAGE=$(SPOOL_READER_IMAGE) \
+	go test -tags e2e -count=1 -timeout 45m -v ./test/e2e/ \
+		-run 'TestAnAgentDeniedAGatingPermissionStopsInsteadOfWaitingForever|TestAPermissionRevokedFromTheRunningAgentReachesThePayload' 2>&1 \
+		| tee test/e2e/logs/watch-e2e-$$(date +%Y%m%d-%H%M%S).log
+
 lifecycle-e2e: kind-load ## Deploy the controller in kind, stage an unschedulable pod and a scheduler preemption, and assert both reach the payloads (ADR 0021); log goes to test/e2e/logs/
 	@mkdir -p test/e2e/logs
 	docker pull $(SPOOL_READER_IMAGE)
diff --git a/docs/adr/0033-policy-sources-degrade.md b/docs/adr/0033-policy-sources-degrade.md
index 6e52f88..7226996 100644
--- a/docs/adr/0033-policy-sources-degrade.md
+++ b/docs/adr/0033-policy-sources-degrade.md
@@ -4,6 +4,7 @@ Date: 2026-08-24
 
 Status: Accepted
 Amends: 0032 §1
+Amended by: 0035
 
 Changes how the agent behaves when it cannot read a policy source, and adds an
 `unavailable_sources` line to `workload_policy` and `cluster_policy`. No new
diff --git a/docs/adr/0035-watch-failures-are-noticed.md b/docs/adr/0035-watch-failures-are-noticed.md
new file mode 100644
index 0000000..96a8d25
--- /dev/null
+++ b/docs/adr/0035-watch-failures-are-noticed.md
@@ -0,0 +1,151 @@
+# 0035. A cache that stopped being fed is noticed: gating caches stop the agent, the rest degrade their payload
+
+Date: 2026-08-25
+
+Status: Accepted
+Amends: 0033 §5
+
+Closes the limitation ADR 0033 recorded as known and unfixed. No new kind, no new
+field, no new RBAC, no change to any payload's shape: `unavailable_sources`
+already exists and now answers a second question with the same word.
+
+## Context
+
+ADR 0033 chose `HasSynced` as the availability signal and got the start of an
+agent's life right: an empty list and a refused read became different states,
+and a customer who declines one of the seven policy grants keeps the usage
+collection they installed the agent for.
+
+`HasSynced` is a one-way latch. It turns true when the first LIST succeeds and
+never turns back. A cache whose watch is refused afterwards therefore keeps
+reporting availability while the store answers every query from what it last
+held, and the reflector retries in the background forever. ADR 0033 §5 recorded
+this, judged the damage to be staleness, and declined to build the detector.
+
+Two things about that judgement were wrong.
+
+**Stale is not the right word for a superseding snapshot.** `workload_policy` and
+`cluster_policy` replace their predecessor and state what is true at
+`captured_at`. A budget deleted after the refusal keeps shipping; one created
+after it never ships; and the payload declares every source read, because the
+omission of `unavailable_sources` is itself the claim that nothing was missed.
+The report is not behind — it is wrong, and it is wrong identically at every
+capture until someone restarts the agent. §5's own mitigation, "any restart
+reports correctly", rests on an event this project deliberately made rare: the
+controller is a single-replica Deployment with no volume and can run for weeks.
+
+**The latch is not a property of the seven policy caches.** It is a property of
+every informer. The pod index, the owner chain and the namespace cache are what
+every other payload is assembled from, and a refusal there is not a degraded
+payload but a blind agent producing payloads of entirely ordinary shape. The
+comment above `PodWatcher.Run` had already stated the intended behaviour —
+*a failing watcher must bring the agent down rather than leave it half-blind* —
+and the agent could not honour it, because nothing told it the watch had died.
+
+There is a third case the same blindness produces, and it is the worst of them.
+`WaitForCacheSync` has no timeout. An agent installed without the `pods` grant
+never syncs, never errors and never exits: a Running pod, a healthy Deployment,
+no data, and nothing anywhere that says why.
+
+## Decision
+
+**1. Watch failures are recorded, and the recording is per cache.**
+`SetWatchErrorHandler` fires on every failed `ListAndWatch`, including a failed
+LIST, and is the only signal client-go offers — there is no matching callback for
+success. Each informer gets a health record holding when its current run of
+failures began and when it was last seen failing.
+
+**2. The cause is not examined.** ADR 0033 §2 decided that the agent does not
+classify why a source is unreadable, and the same holds here. A revoked grant, a
+deleted ServiceAccount whose token stops being accepted, an API version dropped
+by a cluster upgrade and an authorization webhook that changed its mind are one
+state — the cache is not being fed — and the payload has no field in which the
+difference could be expressed.
+
+What classification would have bought, persistence buys instead: a transient
+error happens once, a refusal repeats until someone fixes it.
+
+**3. Recovery is inferred from silence, with the reflector's own number.** A run
+of failures ends after two minutes without one. That is not a guess: while a
+watch is failing the reflector retries at most every [30,60) seconds, and
+client-go states the identical judgement in `defaultBackoffReset` — "if we don't
+backoff for 2min, assume API server is healthy". Adopting its constant keeps the
+two from disagreeing.
+
+A streak's length is measured between its first and last failure, never up to
+now. One failure spans nothing and is therefore never an outage, which is the
+intended reading of a single dropped connection.
+
+**4. A gating cache that fails for five minutes stops the agent.** Gating is the
+split ADR 0033 §1 already drew — a cache that gates a signal against one that
+adds one — applied to the running agent rather than only to its start. Pods,
+replicasets, jobs, deployments, statefulsets, daemonsets, cronjobs, namespaces
+and nodes gate; the seven policy caches do not.
+
+Not the first error, because a restart is not free: the in-memory windows, the
+restart baselines and the spool are lost with the process (ADR 0026), so dying
+on a five-second API server hiccup would cost more than the blindness it
+prevents. Five minutes is several retries, and it bounds how long the agent can
+be blind before it says so.
+
+The watchdog runs from before the first cache sync, so the never-synced case and
+the went-dark case end the same way: an exit whose message names the resource,
+and a pod in CrashLoopBackOff where a customer can see it. The informers run on
+the watchdog's context rather than the caller's, or the factory's shutdown would
+wait for watches that stop only when the process is signalled.
+
+**5. A policy cache that failed within the last three minutes declares itself
+unavailable.** The existing field, unchanged, now covers both halves of one
+statement: a source is unavailable if it never filled *or* if it is failing now.
+The two are not distinguished, for the same reason the causes are not — a
+consumer does the same thing with either.
+
+Three minutes is longer than the [30,60)-second retry interval, so a live
+refusal is declared at every capture rather than at some of them. A permission
+restored is therefore declared unavailable for a capture or two more, and a
+transient error can cost one capture's completeness claim. Both are the safe
+direction: the field overstating what was missed is a smaller error than the
+payload overstating what was read.
+
+**6. What the payload carries is the source name.** The error text is logged and
+goes no further. A refusal names the agent's own ServiceAccount and a resource
+class rather than any customer object, but the rule that identities never leave
+the cluster (CLAUDE.md invariant 6) is not one to defend case by case.
+
+**7. Known limitation: revocation is noticed when the watch is next
+re-established.** An established watch is not re-authorized by the API server, so
+a grant removed from a running agent has no effect until client-go re-opens the
+connection — which it does on its own timeout, between five and ten minutes.
+Detection therefore lags a revocation by up to ten minutes, and the gating exit
+by up to fifteen.
+
+This is not fixable from inside the agent without polling permissions, which
+would mean creating SelfSubjectAccessReviews: a write verb in the audit log of a
+product whose promise is that it only reads, answering a narrower question than
+the handler already answers. The lag is recorded here so it stays a decision.
+
+## Consequences
+
+**Easier.** The three ways an agent could go quietly blind — never synced,
+stopped being fed, and the wait with no timeout — now end in either a named exit
+or a payload that says what it could not see. A customer who narrows the
+ClusterRole of a running agent gets the same honest answer they already got when
+narrowing it before install.
+
+**Harder / given up.** The agent can now stop for a reason that is not its own
+fault, and a sustained API server outage will restart it after five minutes,
+losing an interval of accumulated windows. That is the trade ADR 0026 already
+implies: state that cannot survive a restart is state whose loss is preferable to
+its silent corruption.
+
+`unavailable_sources` also became slightly less precise: it can name a source
+that is readable again but failed within the window. The alternative — a field
+that is precise and occasionally false — is worse for a consumer that must decide
+what a report rests on.
+
+**Not changed.** No payload shape, no RBAC, no kind, no collection behaviour when
+the read succeeds. The seven policy caches still degrade rather than stop, which
+is ADR 0033's decision and the bug it fixed.
+
+This ADR records decisions implemented in the same pull request, per the process
+in [`README.md`](README.md).
diff --git a/docs/adr/README.md b/docs/adr/README.md
index 61cab50..b7e1365 100644
--- a/docs/adr/README.md
+++ b/docs/adr/README.md
@@ -92,3 +92,4 @@ line to paste.
 - [0032. What bounds a workload from outside its own spec, and the first widening of read access](0032-cluster-policy.md)
 - [0033. A permission the agent was not given degrades one payload, and says so](0033-policy-sources-degrade.md)
 - [0034. The restart counter is shipped as a reading, so the day-one history is not lost](0034-restart-counter-readings.md)
+- [0035. A cache that stopped being fed is noticed: gating caches stop the agent, the rest degrade their payload](0035-watch-failures-are-noticed.md)
diff --git a/docs/backend-requirements.md b/docs/backend-requirements.md
index dfebf6e..e892559 100644
--- a/docs/backend-requirements.md
+++ b/docs/backend-requirements.md
@@ -222,6 +222,32 @@ A container whose counter is zero has no record: the payload is a snapshot, so
 absence means zero rather than unknown. An empty record list means no collected
 container has ever restarted, and MUST NOT be read as a failed collection.
 
+**`unavailable_sources` disqualifies the payload's absences, and says nothing
+about why** ([ADR 0033](adr/0033-policy-sources-degrade.md),
+[ADR 0035](adr/0035-watch-failures-are-noticed.md)). `workload_policy` and
+`cluster_policy` each declare the sources they could not read at that capture.
+The field names resource classes, is omitted when every source was read, and
+each payload declares only its own sources.
+
+- The rule it exists for: in these payloads an absent record is a positive
+  statement — "nothing constrains this workload". When a source is listed, that
+  statement is void for anything that source would have supplied, including
+  workloads that vanish from the payload entirely rather than appearing with a
+  field missing. The backend MUST NOT treat the payload as a complete picture of
+  what it declares unreadable.
+- One list means one thing: not available at this capture. It does **not** say
+  whether the permission was never granted, was taken away from the running
+  agent, or was lost to a webhook or an unreachable API server, and the backend
+  MUST NOT infer a cause from it.
+- It is about that capture and no other. A source listed once and not again was
+  readable again; a source that is readable again may still be listed for a
+  capture or two, because the agent errs toward declaring a gap. Neither is a
+  claim that the source is permanently gone, and the backend MUST NOT latch it.
+- A missing grant the agent cannot work without produces no payload at all — the
+  agent exits (ADR 0035). Absence of deliveries and a declared source are
+  therefore different signals, and the backend SHOULD alert on the first through
+  whatever it uses for a silent cluster.
+
 **`pod_disruptions` is what the cluster took away, not what failed**
 ([ADR 0021](adr/0021-pod-lifecycle-journal.md)). One record per pod per
 hour-aligned window: preempted, evicted under node pressure, drained by a taint,
diff --git a/docs/security.md b/docs/security.md
index 673ca81..6b0a320 100644
--- a/docs/security.md
+++ b/docs/security.md
@@ -136,6 +136,23 @@ indistinguishable from "your cluster has none of those"
 ([ADR 0033](adr/0033-policy-sources-degrade.md)). Granting a permission later
 takes effect at the next flush, without restarting the agent.
 
+**This holds for a permission you narrow later, not only for one you never
+granted.** A grant removed from a running agent is noticed and declared the same
+way, which it was not before
+([ADR 0035](adr/0035-watch-failures-are-noticed.md)); detection lags the change
+by up to ten minutes, because an established watch is not re-authorized by the
+API server until the client re-opens it. What the payload names is the resource
+class, never an object.
+
+**Withdrawing a grant the agent cannot work without stops the agent, visibly.**
+`pods`, `namespaces`, `nodes` and the owner-chain resources are what every
+payload is assembled from, so there is no single payload left to degrade. If one
+of those becomes unreadable — before the first sync or five minutes into a
+refusal that started later — the agent exits with a message naming the resource
+and its pod enters `CrashLoopBackOff`. It does not keep answering from a frozen
+cache, and it does not sit silently waiting, which is what it used to do when a
+grant was missing at install (ADR 0035).
+
 **Authenticating node reports adds no RBAC.** When the node DaemonSet is
 installed (ADR 0010), the controller authenticates each node's projected
 ServiceAccount token by verifying its signature against the cluster's published
@@ -403,8 +420,8 @@ appearing here (ADR 0022).
 | `job_runs` | Per hour: each Job that finished — when it started and finished, whether it succeeded, the failure reason, its pod success/failure counts, and its declared `parallelism`/`completions`/`backoffLimit` | **yes** |
 | `deployment_revisions` | Per flush: each ReplicaSet of a collected Deployment — its revision number, when it was created, its desired/current/ready replica counts, and each container's image reference | **yes** |
 | `workload_metadata` | Declared shape per (namespace, workload, container, image digest): image, requests, limits, ports, QoS, replica counts by phase, by node, and by unscheduled reason, and the pod's reduced placement constraints (ADR 0031) | no |
-| `workload_policy` | Per collected workload, what bounds it from outside its own spec: disruption budgets covering it, autoscalers driving it, and the volume claims its pods mount (ADR 0032); plus `unavailable_sources`, naming any of those the agent could not read (ADR 0033) | no |
-| `cluster_policy` | Per collected namespace, its LimitRanges and ResourceQuotas; plus the cluster's PriorityClasses and StorageClasses, which workloads reference by name (ADR 0032), and `unavailable_sources` for any of those the agent could not read (ADR 0033). StorageClass `parameters` are never read | no |
+| `workload_policy` | Per collected workload, what bounds it from outside its own spec: disruption budgets covering it, autoscalers driving it, and the volume claims its pods mount (ADR 0032); plus `unavailable_sources`, naming any of those the agent could not read — never granted, or no longer being fed (ADR 0033, ADR 0035) | no |
+| `cluster_policy` | Per collected namespace, its LimitRanges and ResourceQuotas; plus the cluster's PriorityClasses and StorageClasses, which workloads reference by name (ADR 0032), and `unavailable_sources` for any of those the agent could not read — never granted, or no longer being fed (ADR 0033, ADR 0035). StorageClass `parameters` are never read | no |
 | `node_metadata` | Per node: name, size, instance and capacity type, zone, region, kernel version, CPU architecture | no |
 | `go_inventory` | Per (namespace, workload, container): Go version, main module, image digest, PGO flag, plus a fleet-coverage block | no |
 | `go_build` | Per image digest, written once: Go version, main module, dependency module **paths**, allow-listed build settings | no |
@@ -585,6 +602,14 @@ records, and the distinction is unrecoverable once the moment has passed
 ([ADR 0013](adr/0013-observation-completeness.md)). No pod, node, or workload is
 named in this data; the failure counts are cluster-wide totals.
 
+The same honesty applies to the API server rather than the kubelet. A cache the
+agent cannot read is declared in `unavailable_sources` on the payload that reads
+it — whether the permission was never granted or was taken away from the running
+agent — and a cache the agent cannot work without stops the agent instead
+([ADR 0035](adr/0035-watch-failures-are-noticed.md)). Both name a resource class
+and nothing else; the underlying error, which names the agent's own
+ServiceAccount, stays in the agent's log and is never part of a payload.
+
 ### Profile filtering (applies on the node, before egress — ADR 0011)
 
 - Stack frames are filtered by Go module path on the node that captured them.
diff --git a/internal/collector/nodes.go b/internal/collector/nodes.go
index 64c5b82..4990f44 100644
--- a/internal/collector/nodes.go
+++ b/internal/collector/nodes.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sort"
 	"sync"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/informers"
@@ -51,13 +52,35 @@ type NodeWatcher struct {
 
 	mu    sync.RWMutex
 	nodes map[string]NodeInfo
+
+	// The node list gates a signal rather than adding one: the usage poller
+	// takes its target list from it, so a frozen cache stops new nodes from ever
+	// being polled while the old ones are polled forever. A sustained failure
+	// stops the agent (ADR 0035).
+	limits watchLimits
+	now    func() time.Time
 }
 
 // NewNodeWatcher returns a watcher that calls onNode for every node present
 // at start and for every node added afterwards. onNode is called from the
 // informer goroutine and must not block.
 func NewNodeWatcher(clientset kubernetes.Interface, onNode func(NodeInfo)) *NodeWatcher {
-	return &NodeWatcher{clientset: clientset, onNode: onNode, nodes: make(map[string]NodeInfo)}
+	return &NodeWatcher{
+		clientset: clientset,
+		onNode:    onNode,
+		nodes:     make(map[string]NodeInfo),
+		limits:    defaultWatchLimits(),
+		now:       time.Now,
+	}
+}
+
+// clock is the watchdog's time source, defaulting to the wall clock for a
+// watcher assembled by hand in a test.
+func (w *NodeWatcher) clock() time.Time {
+	if w.now == nil {
+		return time.Now()
+	}
+	return w.now()
 }
 
 // Names returns the currently known node names, sorted. The usage poller
@@ -96,11 +119,31 @@ func (w *NodeWatcher) Run(ctx context.Context) error {
 	// Informers must be instantiated before Start, or the factory won't
 	// run them.
 	nodesInformer := factory.Core().V1().Nodes().Informer()
+	gating := map[string]*watchHealth{
+		"nodes": trackWatch(nodesInformer, w.limits.streakGap, w.clock),
+	}
+
+	// Started before the wait, and the informers run on runCtx, for the two
+	// reasons spelled out in PodWatcher.Run: a cache refused from the first LIST
+	// never syncs and the wait has no timeout, and Shutdown waits for every
+	// informer it started.
+	runCtx, cancelRun := context.WithCancel(ctx)
+	defer cancelRun()
+	fatal := make(chan error, 1)
+	go func() {
+		if err := watchdog(runCtx, w.clock, w.limits, gating); err != nil {
+			fatal <- err
+			cancelRun()
+		}
+	}()
 
-	factory.Start(ctx.Done())
+	factory.Start(runCtx.Done())
 	defer factory.Shutdown()
 
-	if !cache.WaitForCacheSync(ctx.Done(), nodesInformer.HasSynced) {
+	if !cache.WaitForCacheSync(runCtx.Done(), nodesInformer.HasSynced) {
+		if err := takeWatchFailure(fatal); err != nil {
+			return err
+		}
 		if ctx.Err() != nil {
 			return nil // canceled during sync — a normal shutdown, not a failure
 		}
@@ -139,8 +182,8 @@ func (w *NodeWatcher) Run(ctx context.Context) error {
 	}
 	defer func() { _ = nodesInformer.RemoveEventHandler(reg) }()
 
-	<-ctx.Done()
-	return nil
+	<-runCtx.Done()
+	return takeWatchFailure(fatal)
 }
 
 // upsert stores the node's current collected view and reports it through
diff --git a/internal/collector/pods.go b/internal/collector/pods.go
index 18ec52d..30f8afa 100644
--- a/internal/collector/pods.go
+++ b/internal/collector/pods.go
@@ -9,6 +9,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -172,6 +173,19 @@ type PodWatcher struct {
 	// still starting.
 	policySources []policySource
 
+	// gating holds the health record of every cache the collected view is
+	// assembled from: the pod index, the owner chain, the namespaces the
+	// opt-out is read from. A sustained failure in any of them stops the agent
+	// rather than degrading a payload, because there is no payload left to
+	// degrade — the same split ADR 0033 §1 drew between caches that gate a
+	// signal and caches that add one, applied to the running agent and not only
+	// to its start (ADR 0035).
+	gating map[string]*watchHealth
+	// limits and now are the watchdog's constants and clock, fields so tests can
+	// compress five minutes into milliseconds.
+	limits watchLimits
+	now    func() time.Time
+
 	// Placement terms the reduction refused to carry, counted per pod
 	// description. Aggregate only: what was dropped is counted, never named
 	// (CLAUDE.md invariant 6). Both should sit at zero on an ordinary cluster;
@@ -233,6 +247,8 @@ func NewPodWatcher(clientset kubernetes.Interface, onPod func(PodInfo)) *PodWatc
 		clientset:           clientset,
 		onPod:               onPod,
 		factory:             factory,
+		limits:              defaultWatchLimits(),
+		now:                 time.Now,
 		filter:              NewFilter(nil, nil),
 		reportedOOMs:        make(map[string]struct{}),
 		restartCounts:       make(map[string]restartBaseline),
@@ -261,6 +277,19 @@ func NewPodWatcher(clientset kubernetes.Interface, onPod func(PodInfo)) *PodWatc
 	// Calling Lister() above already registered each informer with the
 	// factory, so Start runs them; what differs is that nothing blocks on
 	// them (ADR 0033).
+	//
+	// Each also gets a health record, because HasSynced answers only half the
+	// question: it says whether the cache ever filled, never whether it is
+	// still being fed (ADR 0035).
+	policyInformers := map[string]watchInformer{
+		"pod_disruption_budgets":     factory.Policy().V1().PodDisruptionBudgets().Informer(),
+		"horizontal_pod_autoscalers": factory.Autoscaling().V2().HorizontalPodAutoscalers().Informer(),
+		"persistent_volume_claims":   factory.Core().V1().PersistentVolumeClaims().Informer(),
+		"limit_ranges":               factory.Core().V1().LimitRanges().Informer(),
+		"resource_quotas":            factory.Core().V1().ResourceQuotas().Informer(),
+		"priority_classes":           factory.Scheduling().V1().PriorityClasses().Informer(),
+		"storage_classes":            factory.Storage().V1().StorageClasses().Informer(),
+	}
 	w.policySources = []policySource{
 		{name: "pod_disruption_budgets", synced: factory.Policy().V1().PodDisruptionBudgets().Informer().HasSynced},
 		{name: "horizontal_pod_autoscalers", synced: factory.Autoscaling().V2().HorizontalPodAutoscalers().Informer().HasSynced},
@@ -270,40 +299,85 @@ func NewPodWatcher(clientset kubernetes.Interface, onPod func(PodInfo)) *PodWatc
 		{name: "priority_classes", synced: factory.Scheduling().V1().PriorityClasses().Informer().HasSynced},
 		{name: "storage_classes", synced: factory.Storage().V1().StorageClasses().Informer().HasSynced},
 	}
+	for i, source := range w.policySources {
+		w.policySources[i].health = trackWatch(policyInformers[source.name], w.limits.streakGap, w.clock)
+	}
+
+	// The gating caches. Every one of them is instantiated here — the listers
+	// above did it — so the handlers are all registered before the factory
+	// starts, which is the only time SetWatchErrorHandler is allowed to be
+	// called. The names are the resource classes as the ClusterRole spells
+	// them, so a failure reads as the grant it concerns.
+	w.gating = map[string]*watchHealth{
+		"pods":         trackWatch(factory.Core().V1().Pods().Informer(), w.limits.streakGap, w.clock),
+		"replicasets":  trackWatch(factory.Apps().V1().ReplicaSets().Informer(), w.limits.streakGap, w.clock),
+		"jobs":         trackWatch(factory.Batch().V1().Jobs().Informer(), w.limits.streakGap, w.clock),
+		"deployments":  trackWatch(factory.Apps().V1().Deployments().Informer(), w.limits.streakGap, w.clock),
+		"statefulsets": trackWatch(factory.Apps().V1().StatefulSets().Informer(), w.limits.streakGap, w.clock),
+		"daemonsets":   trackWatch(factory.Apps().V1().DaemonSets().Informer(), w.limits.streakGap, w.clock),
+		"cronjobs":     trackWatch(factory.Batch().V1().CronJobs().Informer(), w.limits.streakGap, w.clock),
+		"namespaces":   trackWatch(factory.Core().V1().Namespaces().Informer(), w.limits.streakGap, w.clock),
+	}
 	return w
 }
 
-// policySource is one cache a policy payload reads, paired with a way to ask
-// whether it ever filled.
+// policySource is one cache a policy payload reads, paired with two ways to ask
+// about it: whether it ever filled, and whether it is still being fed.
 //
-// HasSynced is the whole signal, and choosing it over "the list came back
-// empty" is the point: a cluster with no PodDisruptionBudgets syncs fine and
-// lists zero objects, while a cluster that denied the permission never syncs at
-// all. An empty result and a refused read are therefore different states rather
-// than the same silence.
+// HasSynced answers the first, and choosing it over "the list came back empty"
+// is the point: a cluster with no PodDisruptionBudgets syncs fine and lists zero
+// objects, while a cluster that denied the permission never syncs at all. An
+// empty result and a refused read are therefore different states rather than the
+// same silence.
+//
+// It cannot answer the second, because it is a one-way latch — a cache that
+// filled and then lost its watch keeps saying yes while serving what it last
+// held. The health record covers that half (ADR 0035).
 type policySource struct {
 	name   string
 	synced cache.InformerSynced
+	health *watchHealth
+}
+
+// clock is the watchdog's time source, defaulting to the wall clock for a
+// watcher assembled by hand in a test.
+func (w *PodWatcher) clock() time.Time {
+	if w.now == nil {
+		return time.Now()
+	}
+	return w.now()
 }
 
-// unavailablePolicySources returns the names of the caches that have not filled,
-// sorted, for the payload to declare.
+// unavailablePolicySources returns the names of the caches this payload cannot
+// be assembled from, sorted, for the payload to declare.
+//
+// A source is unavailable if it never filled or if it is failing now. The two
+// are one statement to a consumer — "not available at this capture" — and
+// deliberately not distinguished in the payload, for the same reason ADR 0033 §2
+// refused to distinguish the causes: neither the shape nor the reader has a use
+// for the difference.
 //
 // The names are resource classes, never customer objects, so nothing here can
 // identify a workload (CLAUDE.md invariant 6). The answer is about this capture
 // and no other: a cache that fills a minute later is simply present at the next
-// flush.
+// flush, and a permission restored is declared unavailable for a capture or two
+// more, which is the direction to be wrong in.
 func (w *PodWatcher) unavailablePolicySources(want ...string) []string {
 	wanted := make(map[string]bool, len(want))
 	for _, name := range want {
 		wanted[name] = true
 	}
+	now := w.clock()
+	window := w.limits.unavailableFor
+	if window == 0 {
+		window = watchUnavailableFor
+	}
 	var out []string
 	for _, source := range w.policySources {
 		if !wanted[source.name] {
 			continue
 		}
-		if source.synced == nil || !source.synced() {
+		if source.synced == nil || !source.synced() || source.health.failedWithin(now, window) {
 			out = append(out, source.name)
 		}
 	}
@@ -352,10 +426,36 @@ func (w *PodWatcher) Run(ctx context.Context) error {
 	// registered with the factory — every Lister() call instantiates one — so
 	// Start runs them like the rest.
 
-	factory.Start(ctx.Done())
+	// The watchdog starts before the wait, not after it, and that ordering is
+	// the point. A gating cache that is refused from the first LIST never syncs,
+	// and WaitForCacheSync has no timeout — the agent would sit there
+	// indefinitely, holding no data and reporting no error. Watching the
+	// failures makes the two cases one: a cache that never filled and a cache
+	// that stopped being fed both end the same way, with an error naming the
+	// resource (ADR 0035).
+	//
+	// The informers run on runCtx rather than on ctx, because Shutdown below
+	// waits for every one of them: started on the outer context they would
+	// outlive the watchdog's verdict and hold the agent up until the process
+	// was signalled — which is the shutdown this decision exists to avoid
+	// waiting for.
+	runCtx, cancelRun := context.WithCancel(ctx)
+	defer cancelRun()
+	fatal := make(chan error, 1)
+	go func() {
+		if err := watchdog(runCtx, w.clock, w.limits, w.gating); err != nil {
+			fatal <- err
+			cancelRun()
+		}
+	}()
+
+	factory.Start(runCtx.Done())
 	defer factory.Shutdown()
 
-	if !cache.WaitForCacheSync(ctx.Done(), append(ownerSynced, podsInformer.HasSynced)...) {
+	if !cache.WaitForCacheSync(runCtx.Done(), append(ownerSynced, podsInformer.HasSynced)...) {
+		if err := takeWatchFailure(fatal); err != nil {
+			return err
+		}
 		if ctx.Err() != nil {
 			return nil // canceled during sync — a normal shutdown, not a failure
 		}
@@ -447,8 +547,8 @@ func (w *PodWatcher) Run(ctx context.Context) error {
 	}
 	defer func() { _ = jobsInformer.RemoveEventHandler(jobReg) }()
 
-	<-ctx.Done()
-	return nil
+	<-runCtx.Done()
+	return takeWatchFailure(fatal)
 }
 
 // indexPod records an admitted pod for sample attribution. info is the pod's
diff --git a/internal/collector/policy_test.go b/internal/collector/policy_test.go
index c3b5f1f..ceb1c8e 100644
--- a/internal/collector/policy_test.go
+++ b/internal/collector/policy_test.go
@@ -67,15 +67,17 @@ func replicaSetOwnedBy(name, deployment string) runtime.Object {
 	return replicaSet(name, deployment, "1", "example.com/app:1", 1, 1)
 }
 
-func budget(name, app, minAvailable string, allowed int32) *policyv1.PodDisruptionBudget {
+// budget is the fixture every policy test wants: a budget covering the `web`
+// workload that forbids eviction outright, which is the case worth reporting.
+func budget() *policyv1.PodDisruptionBudget {
 	return &policyv1.PodDisruptionBudget{
-		ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: name},
+		ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "web-pdb"},
 		Spec: policyv1.PodDisruptionBudgetSpec{
-			Selector:     &metav1.LabelSelector{MatchLabels: map[string]string{"app": app}},
-			MinAvailable: ptr.To(intstr.Parse(minAvailable)),
+			Selector:     &metav1.LabelSelector{MatchLabels: map[string]string{"app": "web"}},
+			MinAvailable: ptr.To(intstr.Parse("100%")),
 		},
 		Status: policyv1.PodDisruptionBudgetStatus{
-			DisruptionsAllowed: allowed, CurrentHealthy: 2, DesiredHealthy: 2, ExpectedPods: 2,
+			DisruptionsAllowed: 0, CurrentHealthy: 2, DesiredHealthy: 2, ExpectedPods: 2,
 		},
 	}
 }
@@ -154,7 +156,7 @@ func TestBudgetAttachesToTheWorkloadOfTheSelectedPods(t *testing.T) {
 	got, _ := collectPolicy(t, NewFilter(nil, nil), hasRecords,
 		policyPod("web-1", "web"),
 		replicaSetOwnedBy("web-abc", "web"),
-		budget("web-pdb", "web", "100%", 0),
+		budget(),
 	)
 	if len(got) != 1 {
 		t.Fatalf("records = %d, want 1: %+v", len(got), got)
@@ -184,7 +186,7 @@ func TestBudgetOverExcludedPodsNamesNothing(t *testing.T) {
 		func([]WorkloadPolicy, ClusterPolicy) bool { return filter.Snapshot().PodsObserved > 0 },
 		policyPod("web-1", "web"),
 		replicaSetOwnedBy("web-abc", "web"),
-		budget("web-pdb", "web", "100%", 0),
+		budget(),
 		&corev1.LimitRange{ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "defaults"}},
 	)
 	if len(got) != 0 {
@@ -388,7 +390,7 @@ func TestDeniedPolicySourceDegradesInsteadOfStoppingTheAgent(t *testing.T) {
 	clientset := fake.NewClientset(
 		policyPod("web-1", "web"),
 		replicaSetOwnedBy("web-abc", "web"),
-		budget("web-pdb", "web", "100%", 0),
+		budget(),
 	)
 	forbid := func(k8stesting.Action) (bool, runtime.Object, error) {
 		return true, nil, apierrors.NewForbidden(
diff --git a/internal/collector/watchhealth.go b/internal/collector/watchhealth.go
new file mode 100644
index 0000000..066932a
--- /dev/null
+++ b/internal/collector/watchhealth.go
@@ -0,0 +1,244 @@
+package collector
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"sync"
+	"time"
+
+	"k8s.io/client-go/tools/cache"
+)
+
+// The watch-health machinery answers a question `HasSynced` cannot: is this
+// cache still being fed?
+//
+// `HasSynced` is a one-way latch. It turns true when the first LIST succeeds
+// and never turns back, so a cache whose watch is refused afterwards keeps
+// reporting availability while serving whatever it last held. The reflector
+// retries forever and the store answers every query, which means the agent goes
+// blind without noticing and every payload it assembles still claims to be
+// current (ADR 0035).
+//
+// The only signal client-go offers is the other direction: SetWatchErrorHandler
+// fires on every failed ListAndWatch, including a failed LIST. There is no
+// matching "it succeeded" callback, so recovery has to be inferred from
+// silence — which is exactly what the reflector itself does, and with the same
+// number (see watchStreakGap).
+
+const (
+	// watchStreakGap is how long without a failure means the failures stopped.
+	//
+	// It is the reflector's own constant, for the reflector's own reason: while
+	// a watch is failing it retries at most every [30,60) seconds, so a longer
+	// quiet stretch cannot be part of a live failure. client-go states the same
+	// judgement as `defaultBackoffReset` — "if we don't backoff for 2min, assume
+	// API server is healthy" — and there is no case for a second opinion here.
+	watchStreakGap = 2 * time.Minute
+	// watchFatalStreak is how long a gating cache must fail continuously before
+	// the agent stops.
+	//
+	// Not the first error: a restart costs the in-memory windows, the restart
+	// baselines and the spool (emptyDir, ADR 0026), so dying on a five-second
+	// API server hiccup would cost more than the blindness it prevents. Five
+	// minutes is several retries, and it bounds how long the agent can be blind
+	// before it says so.
+	watchFatalStreak = 5 * time.Minute
+	// watchUnavailableFor is how recent a failure must be for a policy source to
+	// declare itself unavailable in the payload it feeds.
+	//
+	// Longer than the [30,60)-second retry interval, so a live refusal is
+	// declared at every capture rather than at some of them; long enough that
+	// the minute-by-minute flush cannot fall in a gap. A permission restored is
+	// therefore still declared unavailable for a capture or two, which is the
+	// direction to be wrong in.
+	watchUnavailableFor = 3 * time.Minute
+	// watchCheckInterval is how often the watchdog re-reads the streaks.
+	watchCheckInterval = 30 * time.Second
+)
+
+// watchLimits are the durations above, held in a field so tests can compress
+// them. Production never builds one by hand.
+type watchLimits struct {
+	streakGap      time.Duration
+	fatalStreak    time.Duration
+	unavailableFor time.Duration
+	checkInterval  time.Duration
+}
+
+func defaultWatchLimits() watchLimits {
+	return watchLimits{
+		streakGap:      watchStreakGap,
+		fatalStreak:    watchFatalStreak,
+		unavailableFor: watchUnavailableFor,
+		checkInterval:  watchCheckInterval,
+	}
+}
+
+// watchHealth is one cache's record of failing: when the current run of
+// failures began, when it was last seen failing, and what it last said.
+//
+// The error text is kept for the log and goes nowhere near a payload. A refusal
+// names the agent's own ServiceAccount and the resource class it was denied,
+// never a customer object — but the rule that identities of filtered-out
+// objects never leave the cluster (CLAUDE.md invariant 6) is not one to defend
+// case by case, so the payload carries the source name and nothing else.
+type watchHealth struct {
+	mu      sync.Mutex
+	gap     time.Duration
+	first   time.Time
+	last    time.Time
+	lastErr string
+}
+
+func newWatchHealth(gap time.Duration) *watchHealth {
+	return &watchHealth{gap: gap}
+}
+
+// setGap re-times an already-registered record. Only tests call it, to compress
+// two minutes into milliseconds without rebuilding the informers.
+func (h *watchHealth) setGap(gap time.Duration) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.gap = gap
+}
+
+// record notes one failed ListAndWatch. A failure further than gap from the
+// previous one starts a new run rather than extending the old one.
+func (h *watchHealth) record(now time.Time, err error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.last.IsZero() || now.Sub(h.last) > h.gap {
+		h.first = now
+	}
+	h.last = now
+	if err != nil {
+		h.lastErr = err.Error()
+	}
+}
+
+// failingFor is how long this cache has been observed failing without pause,
+// and zero when it is not currently failing.
+//
+// The span measured is between the first and last failure, not up to now: it
+// counts only time the agent watched the cache fail, never the interval since.
+// One failure is therefore never a streak, which is the intent — a single
+// dropped connection is not an outage.
+func (h *watchHealth) failingFor(now time.Time) (time.Duration, string) {
+	if h == nil {
+		return 0, ""
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.last.IsZero() || now.Sub(h.last) > h.gap {
+		return 0, ""
+	}
+	return h.last.Sub(h.first), h.lastErr
+}
+
+// failedWithin reports whether this cache failed at any point in the last d.
+func (h *watchHealth) failedWithin(now time.Time, d time.Duration) bool {
+	if h == nil {
+		return false
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return !h.last.IsZero() && now.Sub(h.last) <= d
+}
+
+// watchdog blocks until ctx is canceled, or until one of the gating caches has
+// been failing for longer than limits.fatalStreak — in which case it returns
+// the error that must stop the agent.
+//
+// Stopping is the honest response for a cache that gates a signal: the pod
+// index, the owner chain and the node list are what every other payload is
+// assembled from, so serving them from a frozen store produces payloads of
+// entirely ordinary shape that describe a cluster as it was. A visibly crashing
+// pod is a worse day for the customer and a better one for the data.
+func watchdog(ctx context.Context, now func() time.Time, limits watchLimits, gating map[string]*watchHealth) error {
+	if len(gating) == 0 {
+		<-ctx.Done()
+		return nil
+	}
+	names := make([]string, 0, len(gating))
+	for name := range gating {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	ticker := time.NewTicker(limits.checkInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			at := now()
+			for _, name := range names {
+				failing, lastErr := gating[name].failingFor(at)
+				if failing >= limits.fatalStreak {
+					return fmt.Errorf(
+						"the %s watch has been failing continuously for %s and its cache can no longer be called current: %s",
+						name, failing.Round(time.Second), lastErr)
+				}
+			}
+		}
+	}
+}
+
+// useWatchLimits re-times an assembled watcher, for tests that cannot wait five
+// minutes to watch a five-minute rule fire. It must be called before Run: the
+// health records are already registered with their informers, so only their
+// timings are replaced, never the wiring under test.
+func (w *PodWatcher) useWatchLimits(limits watchLimits, now func() time.Time) {
+	w.limits = limits
+	w.now = now
+	for _, h := range w.gating {
+		h.setGap(limits.streakGap)
+	}
+	for _, source := range w.policySources {
+		source.health.setGap(limits.streakGap)
+	}
+}
+
+// takeWatchFailure reads the watchdog's verdict without blocking. A canceled
+// run has either an error waiting or nothing to say.
+func takeWatchFailure(fatal <-chan error) error {
+	select {
+	case err := <-fatal:
+		return err
+	default:
+		return nil
+	}
+}
+
+// recordWatchFailures returns the handler an informer calls when ListAndWatch
+// drops, bound to one cache's health record.
+//
+// The cause is deliberately not examined. ADR 0033 §2 already decided that the
+// agent does not classify why a source is unreadable — RBAC, a webhook, a
+// removed API version and an unreachable API server are one state, "not
+// available", and telling them apart buys nothing the payload can express.
+// Persistence does the work that classification would: a transient error
+// happens once, a refusal repeats until it is fixed.
+func recordWatchFailures(h *watchHealth, now func() time.Time) cache.WatchErrorHandler {
+	return func(_ *cache.Reflector, err error) {
+		h.record(now(), err)
+	}
+}
+
+// watchInformer is the part of cache.SharedIndexInformer this file needs.
+type watchInformer interface {
+	SetWatchErrorHandler(cache.WatchErrorHandler) error
+}
+
+// trackWatch registers the failure handler and returns the health record.
+//
+// SetWatchErrorHandler refuses only on an informer that has already started,
+// and every call site here runs before the factory does, so the error carries
+// no information a caller could act on.
+func trackWatch(informer watchInformer, gap time.Duration, now func() time.Time) *watchHealth {
+	h := newWatchHealth(gap)
+	_ = informer.SetWatchErrorHandler(recordWatchFailures(h, now))
+	return h
+}
diff --git a/internal/collector/watchhealth_test.go b/internal/collector/watchhealth_test.go
new file mode 100644
index 0000000..6dc3005
--- /dev/null
+++ b/internal/collector/watchhealth_test.go
@@ -0,0 +1,353 @@
+package collector
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes/fake"
+	k8stesting "k8s.io/client-go/testing"
+)
+
+// The arithmetic first: what counts as "failing", given that client-go offers a
+// failure callback and no matching success one.
+
+// A single dropped connection is not an outage. The streak is measured between
+// failures, so one of them spans nothing.
+func TestOneFailureIsNotAnOutage(t *testing.T) {
+	t0 := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+	h := newWatchHealth(2 * time.Minute)
+	h.record(t0, errors.New("connection reset"))
+
+	if d, _ := h.failingFor(t0.Add(time.Second)); d != 0 {
+		t.Errorf("failing for %s after a single error, want 0", d)
+	}
+	if !h.failedWithin(t0.Add(time.Second), time.Minute) {
+		t.Error("the failure was not recorded at all")
+	}
+}
+
+// The span counts only time the agent watched the cache fail — never the
+// interval since the last failure. Measuring up to now would let one error
+// grow into an outage just by waiting.
+func TestTheStreakSpansTheFailuresAndNotTheSilenceAfterThem(t *testing.T) {
+	t0 := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+	h := newWatchHealth(2 * time.Minute)
+	h.record(t0, errors.New("forbidden"))
+	h.record(t0.Add(time.Minute), errors.New("forbidden"))
+	h.record(t0.Add(3*time.Minute), errors.New("forbidden"))
+
+	if d, _ := h.failingFor(t0.Add(3 * time.Minute)); d != 3*time.Minute {
+		t.Errorf("streak = %s, want the 3m between the first failure and the last", d)
+	}
+	// Nothing more arrives. The streak is over, not still growing.
+	if d, _ := h.failingFor(t0.Add(10 * time.Minute)); d != 0 {
+		t.Errorf("streak = %s seven minutes after the last failure, want 0", d)
+	}
+}
+
+// Two minutes of quiet is the reflector's own definition of a healthy API
+// server, and it is what separates one outage from the next.
+func TestAQuietStretchStartsANewStreakInsteadOfExtendingTheOld(t *testing.T) {
+	t0 := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+	h := newWatchHealth(2 * time.Minute)
+	h.record(t0, errors.New("forbidden"))
+	h.record(t0.Add(4*time.Minute), errors.New("forbidden"))
+
+	if d, _ := h.failingFor(t0.Add(4 * time.Minute)); d != 0 {
+		t.Errorf("streak = %s, want 0: the four-minute gap ended the first run", d)
+	}
+	h.record(t0.Add(5*time.Minute), errors.New("forbidden"))
+	if d, _ := h.failingFor(t0.Add(5 * time.Minute)); d != time.Minute {
+		t.Errorf("streak = %s, want the 1m of the second run only", d)
+	}
+}
+
+// The watchdog waits out the limit and then names what failed. A customer
+// reading the log has to learn which grant to look at.
+func TestTheWatchdogStopsOnlyAfterTheStreakOutlivesTheLimitAndNamesTheResource(t *testing.T) {
+	t0 := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+	var mu sync.Mutex
+	now := t0
+	clock := func() time.Time {
+		mu.Lock()
+		defer mu.Unlock()
+		return now
+	}
+	advance := func(d time.Duration) {
+		mu.Lock()
+		now = now.Add(d)
+		mu.Unlock()
+	}
+
+	h := newWatchHealth(2 * time.Minute)
+	gating := map[string]*watchHealth{"pods": h}
+	limits := watchLimits{
+		streakGap:     2 * time.Minute,
+		fatalStreak:   5 * time.Minute,
+		checkInterval: 5 * time.Millisecond,
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- watchdog(ctx, clock, limits, gating) }()
+
+	// Five failures a minute apart span four minutes — the streak runs from the
+	// first to the last, so it is not yet enough.
+	for i := 1; i <= 5; i++ {
+		advance(time.Minute)
+		h.record(clock(), apierrors.NewForbidden(
+			schema.GroupResource{Resource: "pods"}, "", errors.New("no permission")))
+	}
+	if d, _ := h.failingFor(clock()); d != 4*time.Minute {
+		t.Fatalf("streak = %s, want 4m", d)
+	}
+	select {
+	case err := <-done:
+		t.Fatalf("stopped after four minutes of failure: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	advance(time.Minute)
+	h.record(clock(), apierrors.NewForbidden(
+		schema.GroupResource{Resource: "pods"}, "", errors.New("no permission")))
+
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatal("the watchdog returned no error after five minutes of failure")
+		}
+		if !strings.Contains(err.Error(), "pods") {
+			t.Errorf("error does not name the resource: %v", err)
+		}
+		if !strings.Contains(err.Error(), "5m0s") {
+			t.Errorf("error does not state how long it failed: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("the watchdog did not stop after the streak outlived the limit")
+	}
+}
+
+// A recovered cache is not stopped for having failed earlier: the streak has to
+// be live, not merely to have happened.
+func TestTheWatchdogIgnoresAStreakThatEnded(t *testing.T) {
+	t0 := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+	h := newWatchHealth(2 * time.Minute)
+	for i := range 10 {
+		h.record(t0.Add(time.Duration(i)*time.Minute), errors.New("forbidden"))
+	}
+	// Ten minutes of failure, then quiet: the watch is being served again.
+	now := t0.Add(30 * time.Minute)
+	limits := watchLimits{streakGap: 2 * time.Minute, fatalStreak: 5 * time.Minute, checkInterval: 5 * time.Millisecond}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	if err := watchdog(ctx, func() time.Time { return now }, limits, map[string]*watchHealth{"pods": h}); err != nil {
+		t.Fatalf("stopped over a streak that already ended: %v", err)
+	}
+}
+
+// Now the wiring. Every cache the collected view is assembled from must be
+// tracked — a handler registered on six of eight informers is a hole shaped
+// exactly like the one this change closes.
+func TestEveryGatingCacheIsTracked(t *testing.T) {
+	w := NewPodWatcher(fake.NewClientset(), func(PodInfo) {})
+	want := []string{
+		"cronjobs", "daemonsets", "deployments", "jobs",
+		"namespaces", "pods", "replicasets", "statefulsets",
+	}
+	if len(w.gating) != len(want) {
+		t.Errorf("tracked %d gating caches, want %d: %v", len(w.gating), len(want), w.gating)
+	}
+	for _, name := range want {
+		if w.gating[name] == nil {
+			t.Errorf("the %s cache is not tracked", name)
+		}
+	}
+	for _, source := range w.policySources {
+		if source.health == nil {
+			t.Errorf("the %s policy source is not tracked", source.name)
+		}
+	}
+}
+
+// The headline for the gating half: a refusal the agent cannot collect around
+// stops it. Before this change the pod cache never synced, WaitForCacheSync had
+// no timeout, and the agent sat there holding nothing and reporting nothing.
+func TestASustainedRefusalStopsTheAgentInsteadOfWaitingForever(t *testing.T) {
+	clientset := fake.NewClientset()
+	forbidden := func(k8stesting.Action) (bool, watch.Interface, error) {
+		return true, nil, apierrors.NewForbidden(
+			schema.GroupResource{Resource: "pods"}, "", errors.New("no permission"))
+	}
+	clientset.PrependReactor("list", "pods", func(k8stesting.Action) (bool, runtime.Object, error) {
+		return true, nil, apierrors.NewForbidden(
+			schema.GroupResource{Resource: "pods"}, "", errors.New("no permission"))
+	})
+	clientset.PrependWatchReactor("pods", forbidden)
+
+	w := NewPodWatcher(clientset, func(PodInfo) {})
+	w.SetFilter(NewFilter(nil, nil))
+	// Real clock, compressed limits: the reflector's own retries supply the
+	// streak, so what is under test is the wiring and not a stubbed record.
+	w.useWatchLimits(watchLimits{
+		streakGap:      10 * time.Second,
+		fatalStreak:    2 * time.Second,
+		unavailableFor: 3 * time.Minute,
+		checkInterval:  20 * time.Millisecond,
+	}, time.Now)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	err := w.Run(ctx)
+	if err == nil {
+		t.Fatal("the watcher returned no error while the pod cache was refused")
+	}
+	if !strings.Contains(err.Error(), "pods") {
+		t.Errorf("error does not name the refused resource: %v", err)
+	}
+}
+
+// The other half, and the one ADR 0033 §5 recorded as unfixed: a source that
+// filled and then lost its watch. HasSynced still says yes and the store still
+// answers, so without the health record the payload would declare completeness
+// over a frozen cache.
+func TestAPolicySourceThatFilledAndThenWentDarkIsDeclaredUnavailable(t *testing.T) {
+	clientset := fake.NewClientset(
+		policyPod("web-1", "web"),
+		replicaSetOwnedBy("web-abc", "web"),
+		budget(),
+	)
+	feed := newStoppableWatch()
+	var mu sync.Mutex
+	var revoked bool
+	clientset.PrependWatchReactor("poddisruptionbudgets", func(k8stesting.Action) (bool, watch.Interface, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		if revoked {
+			return true, nil, apierrors.NewForbidden(
+				schema.GroupResource{Group: "policy", Resource: "poddisruptionbudgets"},
+				"", errors.New("no permission"))
+		}
+		return true, feed, nil
+	})
+
+	w := NewPodWatcher(clientset, func(PodInfo) {})
+	w.SetFilter(NewFilter(nil, nil))
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- w.Run(ctx) }()
+
+	// The permission is intact: the cache fills and the payload declares nothing.
+	waitFor(t, 5*time.Second, "the budget cache to fill", func() bool {
+		_, unavailable := w.WorkloadPolicies()
+		return len(w.Pods()) > 0 && len(unavailable) == 0
+	})
+
+	// Now it is taken away from the running agent. The store keeps its contents;
+	// only the feed stops.
+	mu.Lock()
+	revoked = true
+	mu.Unlock()
+	feed.Stop()
+
+	waitFor(t, 10*time.Second, "the revoked source to be declared", func() bool {
+		_, unavailable := w.WorkloadPolicies()
+		return len(unavailable) == 1 && unavailable[0] == "pod_disruption_budgets"
+	})
+
+	// The revocation touched one payload and nothing else: the pods are still
+	// collected, and the agent is still running (ADR 0033 §1).
+	if len(w.Pods()) == 0 {
+		t.Error("the revoked policy source stopped unrelated collection")
+	}
+	cancel()
+	if err := <-done; err != nil {
+		t.Fatalf("the watcher stopped over a policy source: %v", err)
+	}
+}
+
+// A cluster policy payload asks about its own sources, so the revocation of a
+// workload-policy source must not appear in it.
+func TestAFailingSourceIsDeclaredOnlyByThePayloadThatReadsIt(t *testing.T) {
+	now := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+	synced := func() bool { return true }
+	failing := newWatchHealth(2 * time.Minute)
+	failing.record(now.Add(-time.Minute), errors.New("forbidden"))
+
+	w := &PodWatcher{
+		now:    func() time.Time { return now },
+		limits: defaultWatchLimits(),
+		policySources: []policySource{
+			{name: "pod_disruption_budgets", synced: synced, health: failing},
+			{name: "horizontal_pod_autoscalers", synced: synced, health: newWatchHealth(time.Minute)},
+			{name: "persistent_volume_claims", synced: synced, health: newWatchHealth(time.Minute)},
+			{name: "limit_ranges", synced: synced, health: newWatchHealth(time.Minute)},
+			{name: "resource_quotas", synced: synced, health: newWatchHealth(time.Minute)},
+			{name: "priority_classes", synced: synced, health: newWatchHealth(time.Minute)},
+			{name: "storage_classes", synced: synced, health: newWatchHealth(time.Minute)},
+		},
+	}
+	workload := w.unavailablePolicySources(workloadPolicySources...)
+	if len(workload) != 1 || workload[0] != "pod_disruption_budgets" {
+		t.Errorf("workload payload sources = %v, want the failing one", workload)
+	}
+	if cluster := w.unavailablePolicySources(clusterPolicySources...); len(cluster) != 0 {
+		t.Errorf("cluster payload sources = %v, want none: it does not read budgets", cluster)
+	}
+}
+
+// A failure old enough to have healed is not carried into later captures: the
+// field is about this capture and no other (ADR 0033 §4).
+func TestAnOldFailureIsNotCarriedIntoLaterCaptures(t *testing.T) {
+	now := time.Date(2026, 8, 25, 12, 0, 0, 0, time.UTC)
+	h := newWatchHealth(2 * time.Minute)
+	h.record(now.Add(-time.Hour), errors.New("forbidden"))
+	w := &PodWatcher{
+		now:    func() time.Time { return now },
+		limits: defaultWatchLimits(),
+		policySources: []policySource{
+			{name: "pod_disruption_budgets", synced: func() bool { return true }, health: h},
+		},
+	}
+	if got := w.unavailablePolicySources("pod_disruption_budgets"); len(got) != 0 {
+		t.Errorf("unavailable = %v, want none: the failure is an hour old", got)
+	}
+}
+
+// stoppableWatch is a watch whose Stop is idempotent, so a test may end the
+// feed without racing the reflector's own shutdown.
+type stoppableWatch struct {
+	ch   chan watch.Event
+	once sync.Once
+}
+
+func newStoppableWatch() *stoppableWatch {
+	return &stoppableWatch{ch: make(chan watch.Event)}
+}
+
+func (s *stoppableWatch) Stop() { s.once.Do(func() { close(s.ch) }) }
+
+func (s *stoppableWatch) ResultChan() <-chan watch.Event { return s.ch }
+
+// waitFor polls until cond holds or the deadline passes.
+func waitFor(t *testing.T, within time.Duration, what string, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(within)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for %s", what)
+}
diff --git a/test/e2e/inventory_test.go b/test/e2e/inventory_test.go
index 4123296..bd96cff 100644
--- a/test/e2e/inventory_test.go
+++ b/test/e2e/inventory_test.go
@@ -435,12 +435,26 @@ func deploySampleWorkload(ctx context.Context, t *testing.T, cs kubernetes.Inter
 // this namespace, and adds the shell sidecar that shares the spool volume so the
 // test can read the payload the controller writes.
 func deployController(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns, image, configYAML string) {
+	t.Helper()
+	deployControllerWithRole(ctx, t, cs, ns, image, configYAML, nil)
+}
+
+// controllerClusterRoleName is the per-run name deployController gives the
+// cluster-scoped role, so a test can narrow the grant it was installed with.
+func controllerClusterRoleName(ns string) string {
+	return "runtime-agent-controller-" + ns
+}
+
+// deployControllerWithRole is deployController with a hook that rewrites the
+// ClusterRole before it is created — the only way to start an agent that was
+// never given a grant, since patching afterwards races the pod's first LIST.
+func deployControllerWithRole(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns, image, configYAML string, narrow func(*rbacv1.ClusterRole)) {
 	t.Helper()
 	data, err := os.ReadFile(controllerManifestPath)
 	if err != nil {
 		t.Fatalf("reading manifest %s: %v", controllerManifestPath, err)
 	}
-	clusterRoleName := "runtime-agent-controller-" + ns
+	clusterRoleName := controllerClusterRoleName(ns)
 	reader := utilyaml.NewYAMLReader(bufio.NewReader(bytes.NewReader(data)))
 	for {
 		doc, err := reader.Read()
@@ -472,6 +486,9 @@ func deployController(ctx context.Context, t *testing.T, cs kubernetes.Interface
 			var cr rbacv1.ClusterRole
 			mustUnmarshal(t, doc, &cr)
 			cr.Name = clusterRoleName
+			if narrow != nil {
+				narrow(&cr)
+			}
 			create(ctx, t, "ClusterRole", func() error {
 				_, err := cs.RbacV1().ClusterRoles().Create(ctx, &cr, metav1.CreateOptions{})
 				return err
diff --git a/test/e2e/watch_test.go b/test/e2e/watch_test.go
new file mode 100644
index 0000000..1766219
--- /dev/null
+++ b/test/e2e/watch_test.go
@@ -0,0 +1,311 @@
+//go:build e2e
+
+package e2e
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+// The two halves of ADR 0035, staged against a real API server because the
+// thing under test is what the API server does with a permission that changes
+// while a client is connected — which no fake reproduces.
+//
+// Both tests take minutes rather than seconds, and the reason is the mechanism
+// itself: an established watch is not re-authorized, so a revoked grant is
+// noticed when client-go next re-establishes the connection, which it does on a
+// timeout of its own choosing between five and ten minutes.
+
+// TestAnAgentDeniedAGatingPermissionStopsInsteadOfWaitingForever installs the
+// agent without the `pods` grant.
+//
+// Before ADR 0035 this produced the worst outcome available: WaitForCacheSync
+// has no timeout, so the process sat in it forever — a Running pod, a healthy
+// Deployment, no data, and no error anywhere. Now the failing watch is watched,
+// and the agent exits naming the resource, which is what puts the pod in
+// CrashLoopBackOff where a customer can see it.
+//
+// Gated on E2E_AGENT_IMAGE; use `make watch-e2e`.
+func TestAnAgentDeniedAGatingPermissionStopsInsteadOfWaitingForever(t *testing.T) {
+	agentImage := os.Getenv("E2E_AGENT_IMAGE")
+	if agentImage == "" {
+		t.Skip("E2E_AGENT_IMAGE not set; run `make watch-e2e`")
+	}
+	clientset := clusterClient(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 14*time.Minute)
+	defer cancel()
+
+	ns := fmt.Sprintf("runtime-agent-watch-deny-e2e-%d", os.Getpid())
+	createNamespace(ctx, t, clientset, ns)
+
+	deployControllerWithRole(ctx, t, clientset, ns, agentImage, renderControllerConfig(ns),
+		func(cr *rbacv1.ClusterRole) {
+			removeResource(cr, "", "pods")
+		})
+	pod := waitDeploymentPod(ctx, t, clientset, ns, "controller")
+	t.Logf("controller pod: %s (installed without the pods grant)", pod)
+
+	// The watchdog waits out a five-minute streak before it acts, deliberately:
+	// a restart costs the in-memory windows and the spool, so a brief API server
+	// hiccup must not buy one. Allow for that plus a scheduling round.
+	restarts := waitForControllerRestart(ctx, t, clientset, ns, pod, 11*time.Minute)
+	t.Logf("the controller restarted %d time(s)", restarts)
+
+	logs := previousControllerLogs(ctx, t, clientset, ns, pod)
+	for _, want := range []string{"pods", "failing continuously"} {
+		if !strings.Contains(logs, want) {
+			t.Errorf("the exit did not say %q; a crash that does not name the grant is not an improvement:\n%s",
+				want, tail(logs, 2000))
+		}
+	}
+}
+
+// TestAPermissionRevokedFromTheRunningAgentReachesThePayload is the case
+// ADR 0033 §5 recorded as undetected: a grant taken away from an agent that
+// already synced.
+//
+// HasSynced is a one-way latch, so before ADR 0035 the cache went on answering
+// from what it last held and `workload_policy` went on declaring every source
+// read — indefinitely, and identically at every capture. The claim under test
+// is that the payload now says it cannot see budgets, while everything the
+// revocation did not touch keeps being collected (ADR 0033 §1).
+//
+// Gated on E2E_AGENT_IMAGE and E2E_SPOOL_READER_IMAGE; use `make watch-e2e`.
+func TestAPermissionRevokedFromTheRunningAgentReachesThePayload(t *testing.T) {
+	agentImage := os.Getenv("E2E_AGENT_IMAGE")
+	if agentImage == "" {
+		t.Skip("E2E_AGENT_IMAGE not set; run `make watch-e2e`")
+	}
+	config := clusterConfig(t)
+	clientset := clusterClient(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
+	defer cancel()
+
+	ns := fmt.Sprintf("runtime-agent-watch-revoke-e2e-%d", os.Getpid())
+	fixtureNS := ns + "-fixtures"
+	createNamespace(ctx, t, clientset, ns)
+	createNamespace(ctx, t, clientset, fixtureNS)
+	createBudgetFixture(ctx, t, clientset, fixtureNS)
+
+	deployController(ctx, t, clientset, ns, agentImage, renderControllerConfig(ns))
+	pod := waitDeploymentPod(ctx, t, clientset, ns, "controller")
+	t.Logf("controller pod: %s", pod)
+
+	// First the agent must actually read budgets, or the second half proves
+	// nothing: a source that was never available cannot become unavailable.
+	waitForPolicySources(ctx, t, config, clientset, ns, pod, 5*time.Minute,
+		"the budget cache to fill", func(sources []string) bool { return len(sources) == 0 })
+	t.Log("budgets are readable and the payload declares nothing")
+
+	revokeBudgetGrant(ctx, t, clientset, ns)
+	t.Log("the poddisruptionbudgets grant is revoked from the running agent")
+
+	// The established watch is not re-authorized; client-go re-establishes it on
+	// its own five-to-ten-minute timeout, and the refusal lands then.
+	waitForPolicySources(ctx, t, config, clientset, ns, pod, 13*time.Minute,
+		"the revoked source to be declared", func(sources []string) bool {
+			for _, s := range sources {
+				if s == "pod_disruption_budgets" {
+					return true
+				}
+			}
+			return false
+		})
+
+	// One payload degraded, the agent untouched: the split ADR 0033 §1 drew
+	// between a cache that gates a signal and one that adds it.
+	if restarts := controllerRestarts(ctx, t, clientset, ns, pod); restarts != 0 {
+		t.Errorf("the controller restarted %d time(s); a policy source must degrade a payload, not the agent", restarts)
+	}
+	raw, ok := readSpoolFile(ctx, t, config, clientset, ns, pod, workloadMetadataSpoolPath)
+	if !ok {
+		t.Fatal("workload metadata stopped being written when one policy grant was revoked")
+	}
+	var metadata struct {
+		Records []json.RawMessage `json:"records"`
+	}
+	if err := json.Unmarshal([]byte(raw), &metadata); err != nil {
+		t.Fatalf("decoding workload metadata: %v", err)
+	}
+	if len(metadata.Records) == 0 {
+		t.Error("workload metadata is empty: the revocation cost more than the payload that reads budgets")
+	}
+}
+
+// removeResource drops one resource from every rule of the role that names it,
+// leaving the rest of the grant intact — the shape of a customer editing a
+// ClusterRole they do not fully agree with.
+func removeResource(cr *rbacv1.ClusterRole, group, resource string) {
+	rules := cr.Rules[:0]
+	for _, rule := range cr.Rules {
+		if containsString(rule.APIGroups, group) {
+			kept := make([]string, 0, len(rule.Resources))
+			for _, r := range rule.Resources {
+				if r != resource {
+					kept = append(kept, r)
+				}
+			}
+			rule.Resources = kept
+		}
+		// A rule granting nothing is not a narrower grant, it is invalid: the
+		// API server requires at least one resource. Dropping the rule is what a
+		// customer removing the last resource of one would have to do anyway.
+		if len(rule.Resources) == 0 && len(rule.NonResourceURLs) == 0 {
+			continue
+		}
+		rules = append(rules, rule)
+	}
+	cr.Rules = rules
+}
+
+func containsString(list []string, want string) bool {
+	for _, s := range list {
+		if s == want {
+			return true
+		}
+	}
+	return false
+}
+
+// revokeBudgetGrant removes poddisruptionbudgets from the deployed ClusterRole
+// while the agent holds an open watch on them.
+func revokeBudgetGrant(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns string) {
+	t.Helper()
+	name := controllerClusterRoleName(ns)
+	role, err := cs.RbacV1().ClusterRoles().Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("reading ClusterRole %s: %v", name, err)
+	}
+	removeResource(role, "policy", "poddisruptionbudgets")
+	if _, err := cs.RbacV1().ClusterRoles().Update(ctx, role, metav1.UpdateOptions{}); err != nil {
+		t.Fatalf("revoking the budget grant: %v", err)
+	}
+}
+
+// waitForPolicySources polls the workload-policy payload until its declared
+// sources satisfy want.
+func waitForPolicySources(ctx context.Context, t *testing.T, config *rest.Config, cs kubernetes.Interface, ns, pod string, within time.Duration, what string, want func([]string) bool) {
+	t.Helper()
+	deadline := time.Now().Add(within)
+	var last []string
+	for time.Now().Before(deadline) {
+		raw, ok := readSpoolFile(ctx, t, config, cs, ns, pod, workloadPolicySpoolPath)
+		if ok {
+			var payload struct {
+				UnavailableSources []string `json:"unavailable_sources"`
+			}
+			if err := json.Unmarshal([]byte(raw), &payload); err == nil {
+				last = payload.UnavailableSources
+				if want(last) {
+					return
+				}
+			}
+		}
+		time.Sleep(10 * time.Second)
+	}
+	t.Fatalf("timed out waiting for %s; last declared sources = %v", what, last)
+}
+
+// waitForControllerRestart waits for the agent container to have exited at
+// least once and returns the restart count.
+func waitForControllerRestart(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns, pod string, within time.Duration) int32 {
+	t.Helper()
+	deadline := time.Now().Add(within)
+	for time.Now().Before(deadline) {
+		if n := controllerRestarts(ctx, t, cs, ns, pod); n > 0 {
+			return n
+		}
+		time.Sleep(10 * time.Second)
+	}
+	t.Fatalf("the agent was still running %s after being denied the pods grant; it is waiting on a sync that will never come", within)
+	return 0
+}
+
+// controllerRestarts is the agent container's restart count, 0 when the pod is
+// gone (a pod that no longer exists cannot be said to have restarted).
+func controllerRestarts(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns, pod string) int32 {
+	t.Helper()
+	p, err := cs.CoreV1().Pods(ns).Get(ctx, pod, metav1.GetOptions{})
+	if err != nil {
+		return 0
+	}
+	for _, status := range p.Status.ContainerStatuses {
+		if status.Name == "controller" {
+			return status.RestartCount
+		}
+	}
+	return 0
+}
+
+// previousControllerLogs returns the log of the agent container's previous
+// incarnation — the one that exited.
+func previousControllerLogs(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns, pod string) string {
+	t.Helper()
+	stream, err := cs.CoreV1().Pods(ns).GetLogs(pod, &corev1.PodLogOptions{
+		Container: "controller",
+		Previous:  true,
+	}).Stream(ctx)
+	if err != nil {
+		t.Fatalf("reading the previous controller log: %v", err)
+	}
+	defer func() { _ = stream.Close() }()
+	out, err := io.ReadAll(stream)
+	if err != nil {
+		t.Fatalf("reading the previous controller log: %v", err)
+	}
+	return string(out)
+}
+
+// createNamespace creates ns and schedules its removal.
+func createNamespace(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns string) {
+	t.Helper()
+	if _, err := cs.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{Name: ns},
+	}, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("creating namespace %s: %v", ns, err)
+	}
+	t.Cleanup(func() {
+		cleanupCtx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+		defer cancel()
+		_ = cs.CoreV1().Namespaces().Delete(cleanupCtx, ns, metav1.DeleteOptions{})
+	})
+}
+
+// createBudgetFixture gives the workload-policy payload something to read, so
+// "sources all read" is a claim about a cache that actually holds an object.
+func createBudgetFixture(ctx context.Context, t *testing.T, cs kubernetes.Interface, ns string) {
+	t.Helper()
+	minAvailable := intstr.FromString("50%")
+	pdb := &policyv1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: "web-pdb"},
+		Spec: policyv1.PodDisruptionBudgetSpec{
+			MinAvailable: &minAvailable,
+			Selector:     &metav1.LabelSelector{MatchLabels: map[string]string{"app": "web"}},
+		},
+	}
+	if _, err := cs.PolicyV1().PodDisruptionBudgets(ns).Create(ctx, pdb, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("creating the budget fixture: %v", err)
+	}
+}
+
+// tail returns the last n bytes of s, for log excerpts in failure messages.
+func tail(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return "…" + s[len(s)-n:]
+}