diff --git a/charts/runtime-agent/templates/NOTES.txt b/charts/runtime-agent/templates/NOTES.txt
index 8adb12a..c61ea9a 100644
--- a/charts/runtime-agent/templates/NOTES.txt
+++ b/charts/runtime-agent/templates/NOTES.txt
@@ -9,6 +9,17 @@ profile (ADR 0069). Readiness means the caches are synced, so the rollout below
 returns when the agent is actually collecting rather than when it has started:
 
   kubectl -n {{ .Release.Namespace }} rollout status deployment/{{ include "runtime-agent.controllerName" . }}
+{{- if .Values.collection.enabled }}
+
+To see which objects your filters admitted and which they refused — by name,
+with the reason — port-forward the controller's loopback listener (ADR 0084):
+
+  kubectl -n {{ .Release.Namespace }} port-forward deployment/{{ include "runtime-agent.controllerName" . }} {{ include "runtime-agent.collectionPort" . }}:{{ include "runtime-agent.collectionPort" . }} &
+  curl -s localhost:{{ include "runtime-agent.collectionPort" . }}/collection | jq 'select(.collected | not)'
+
+Nothing it answers is ever transmitted. It binds loopback and nothing else, so
+a port-forward is the only way in.
+{{- end }}
 {{- if include "runtime-agent.nodeEnabled" . }}
 
 CHECK ONE THING: the NetworkPolicy needs a CNI that enforces it.
diff --git a/charts/runtime-agent/templates/_helpers.tpl b/charts/runtime-agent/templates/_helpers.tpl
index a50ba04..9934825 100644
--- a/charts/runtime-agent/templates/_helpers.tpl
+++ b/charts/runtime-agent/templates/_helpers.tpl
@@ -38,6 +38,14 @@ the hole in the network policy stays one.
 */}}
 {{- define "runtime-agent.healthPort" -}}9090{{- end -}}
 
+{{/*
+Where the controller answers which objects the filters decided about. Loopback
+and not a value: the address is the whole of what keeps those names inside the
+cluster, and the agent refuses any other host anyway (ADR 0084).
+*/}}
+{{- define "runtime-agent.collectionPort" -}}9091{{- end -}}
+{{- define "runtime-agent.collectionAddress" -}}127.0.0.1:{{ include "runtime-agent.collectionPort" . }}{{- end -}}
+
 {{/* The in-cluster base URL of the controller's receiver. */}}
 {{- define "runtime-agent.controllerEndpoint" -}}
 http://{{ include "runtime-agent.controllerName" . }}.{{ .Release.Namespace }}.svc:8080
diff --git a/charts/runtime-agent/templates/controller-configmap.yaml b/charts/runtime-agent/templates/controller-configmap.yaml
index aae8bf4..a7a74d0 100644
--- a/charts/runtime-agent/templates/controller-configmap.yaml
+++ b/charts/runtime-agent/templates/controller-configmap.yaml
@@ -46,6 +46,11 @@ data:
       # The port the probes ask on. Rendered in every profile: the two questions
       # a kubelet asks are not a feature of one (ADR 0069).
       listenAddress: ":{{ include "runtime-agent.healthPort" . }}"
+    {{- if .Values.collection.enabled }}
+    collection:
+      # Loopback, so a port-forward is the only way in.
+      listenAddress: {{ include "runtime-agent.collectionAddress" . | quote }}
+    {{- end }}
     nodeIntake:
       {{- if include "runtime-agent.nodeEnabled" . }}
       enabled: true
diff --git a/charts/runtime-agent/values.schema.json b/charts/runtime-agent/values.schema.json
index 21af2bb..91863c6 100644
--- a/charts/runtime-agent/values.schema.json
+++ b/charts/runtime-agent/values.schema.json
@@ -56,6 +56,14 @@
         "baseURL": {"type": "string"}
       }
     },
+    "collection": {
+      "description": "The loopback listener that names which objects the filters admitted and which they refused (ADR 0084). No address here: loopback is the whole of the control, and the agent refuses any other host.",
+      "type": "object",
+      "additionalProperties": false,
+      "properties": {
+        "enabled": {"type": "boolean"}
+      }
+    },
     "nodeIntake": {
       "type": "object",
       "additionalProperties": false,
diff --git a/charts/runtime-agent/values.yaml b/charts/runtime-agent/values.yaml
index e8351f4..dd1516a 100644
--- a/charts/runtime-agent/values.yaml
+++ b/charts/runtime-agent/values.yaml
@@ -72,6 +72,14 @@ spool:
 backend:
   baseURL: ""
 
+# The in-cluster listener that answers, by name, which objects your filters
+# admitted and which they refused. It binds loopback and nothing else — no
+# Service, no container port, no hole in the network policy — so the only way
+# to read it is `kubectl port-forward`, whose permission your RBAC already
+# defines. What it answers never leaves the cluster (docs/security.md §11).
+collection:
+  enabled: true
+
 nodeIntake:
   # The audience the node's projected token requests and the controller
   # requires. One value: the chart writes it into both the DaemonSet's token
diff --git a/cmd/agent/main.go b/cmd/agent/main.go
index 707dbfe..f967032 100644
--- a/cmd/agent/main.go
+++ b/cmd/agent/main.go
@@ -30,6 +30,7 @@ import (
 
 	"github.com/RebuildStackCo/runtime-agent/internal/collector"
 	"github.com/RebuildStackCo/runtime-agent/internal/config"
+	"github.com/RebuildStackCo/runtime-agent/internal/decisions"
 	"github.com/RebuildStackCo/runtime-agent/internal/health"
 	"github.com/RebuildStackCo/runtime-agent/internal/inventory"
 	"github.com/RebuildStackCo/runtime-agent/internal/journal"
@@ -1029,6 +1030,16 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 			metrics.Handler(sources.gather, logger), logger).Run
 	}
 
+	// The listener that names what the filters decided about, on a loopback
+	// address of its own. Not a path on the health listener: that port is open
+	// to the whole cluster by a rule that cannot name the kubelet it exists
+	// for, and every path on it answers without naming a cluster object
+	// (ADR 0084).
+	if addr := cfg.Collection.ListenAddress; addr != "" {
+		tasks["collection"] = decisions.New(addr,
+			decisions.Handler(podWatcher.Decisions, podWatcher.Synced, logger), logger).Run
+	}
+
 	// The node-intake receiver is optional (only the ebpf/node profile ships a
 	// DaemonSet). When enabled it becomes one more lifecycle task, validating
 	// node tokens locally against the cluster JWKS (ADR 0010) and joining each
diff --git a/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md b/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md
index 80bc13b..32e1e63 100644
--- a/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md
+++ b/docs/adr/0069-liveness-is-the-process-readiness-is-the-caches.md
@@ -2,7 +2,7 @@
 
 Date: 2026-09-04
 Status: Accepted
-Amended by: 0070
+Amended by: 0070, 0084
 
 ## Context
 
diff --git a/docs/adr/0084-what-was-excluded-is-named-inside-the-cluster.md b/docs/adr/0084-what-was-excluded-is-named-inside-the-cluster.md
new file mode 100644
index 0000000..1b30ec1
--- /dev/null
+++ b/docs/adr/0084-what-was-excluded-is-named-inside-the-cluster.md
@@ -0,0 +1,137 @@
+# 0084. What was excluded is named, inside the cluster and only there
+
+Date: 2026-09-17
+
+Status: Accepted
+
+Amends: 0069 §1
+
+Adds a second listener to the controller, bound to loopback, answering which
+namespaces, pods and Jobs the filters admitted and which they refused, by name
+and with the reason. No payload byte changes, no RBAC, no Service, no container
+port, and no rule in the shipped NetworkPolicy.
+
+## Context
+
+The agent counts what it excludes and never names it. That is the promise
+(CLAUDE.md invariant 6), and ADR 0054 built the coverage payload to keep it: how
+many objects each of the customer's five controls removed, never which.
+
+The promise is about what leaves the cluster. It has been read, everywhere it
+has been implemented, as though it were about what the agent may *know* — and so
+the customer, who owns every one of those annotations and can read them with
+`kubectl`, learns from this agent only that nine of something were excluded.
+
+Two questions follow from that and neither has an answer today.
+
+**Someone turned a sensor off and nobody can say which.** An annotation is a
+deliberate act by a person with access to an object. Later, a different person
+reads a coverage number that moved. Finding the annotation means walking every
+namespace, workload and pod by hand, because the object that carries it is
+precisely the object nothing reports.
+
+**"Why is my workload not in the report?"** has five possible answers — the
+allow list, the deny list, three annotations at three levels — and the agent,
+which has already computed the answer for that exact object, states only a
+total.
+
+`security.md` §11 promises that "filtering behavior is verifiable from output
+rather than only from configuration". What that rests on today is a structured
+log line and a set of counters, neither of which can be resolved to an object.
+
+## Decision
+
+**1. The filters answer by name, to a reader inside the cluster.** One line per
+object the filter gates: the namespace, the pod, the Job, whether it is
+collected, whether it may be profiled, and which control refused it. The pod's
+blind spot is there too — a controller of a kind the agent cannot read
+(ADR 0028 §3) — which is where a customer finds out that their workload
+annotation never had a chance to work.
+
+Nothing is retained to answer this. The informer caches already hold every
+namespace, pod and Job whether admitted or not, so the listing is a projection
+of live state through the same `Admit*` calls the event path makes, and the
+answer cannot drift from the decision. Reading it counts nothing: the counters
+describe decisions the agent acted on, and a read is not one (ADR 0054).
+
+**2. Loopback, and any other host is a startup failure.**
+
+Not a path on the health listener, which is where it would otherwise belong.
+That port is opened by the shipped NetworkPolicy to **any** source, because the
+caller is the kubelet arriving from the node's address and no pod selector in a
+template can name it (ADR 0069 §5). A NetworkPolicy selects ports, not paths, so
+a path there is reachable by every pod in the cluster — and the rule's own
+justification is that every path on that listener "names nothing in the
+cluster", which this one does not satisfy.
+
+Two more reasons the policy cannot be the control: it is not rendered at all in
+an installation with no node DaemonSet, and where it is rendered it is enforced
+by the CNI, so on a cluster whose CNI ignores NetworkPolicy it is inert. The
+bind address is the only control that holds in every case.
+
+Loopback also answers authorization without inventing any. The only way in is
+`kubectl port-forward`, which Kubernetes already gates on `create` of
+`pods/portforward` in the agent's namespace — a permission the customer has
+already granted to exactly the people who should be asking. Requiring a
+projected token on a shared port would have added a credential, a failure mode
+and a request to parse, to arrive at a narrower version of a permission that
+already exists.
+
+So `collection.listenAddress` accepts a loopback host or nothing, and refuses
+anything else at startup rather than normalizing it. An address that exposed
+this listing to every pod in the cluster must not be reachable by a typo.
+
+**3. No query is parsed; the whole list is always answered.** Newline-delimited
+JSON, one object per line, so a reader works through it as it arrives. A parameter would be a request this
+endpoint acts on, which is the shape invariant 1 exists to refuse, and an
+endpoint that filters can return an empty answer that a reader mistakes for
+"nothing matched". Selecting is `jq`'s job. While the gating caches are still
+filling the answer is a 503 and no names, because a short list here is
+indistinguishable from a small cluster.
+
+**4. It never becomes a payload.** This is the one surface that names what the
+filters excluded, and what keeps those names inside the cluster is not the
+handler — it is that nothing which can transmit knows the package exists. That
+is a property of the import graph and is asserted as one.
+
+## Consequences
+
+**Easier.** The customer answers both questions themselves, in one command,
+without sending anything to anyone: which object carries the annotation, and
+under which control a given workload is missing from the report. The §11 promise
+gains a mechanism that resolves to an object. And the support question that
+needs a cluster we cannot see is now answerable by the person who has it.
+
+**Harder / given up.** A second listener in the controller, and with it the end
+of one-listener-per-role (ADR 0069 §1). The boundary is the property, not the
+convenience: the health listener's paths answer without naming anything in the
+cluster, and that property is what a rule opening it to the whole cluster rests
+on. A path that names objects does not share that port.
+
+Anyone who can port-forward into the agent's namespace reads the deny list,
+which ADR 0054 §2 refused to *hash* into a payload. Inside the cluster that is
+not the same disclosure — the reader can already list namespaces, and the list
+is the customer's own configuration — but it is a real widening of who learns it
+easily, and it is stated rather than left to be discovered (ADR 0039).
+
+This covers the objects the controller's filters gate. The node's process scan
+has exclusions of its own (ADR 0015) and is not answered here; it would be the
+same shape on the node's listener, and is not built until someone asks.
+
+Nothing here changes the inference ADR 0054 named: a workload that disappears
+from the collected set as a counter moves can still be identified by
+subtraction, by a backend that chooses to. This listing neither closes that nor
+widens it — it is not shipped.
+
+**Rejected.** Naming excluded objects in the coverage payload, behind a switch
+the operator turns on. An opt-out annotation is an answer, not a question: a
+customer who writes one has said what they want, and a tumbler that offers to
+send the name treats their decision as ours to bargain with.
+
+A `kubectl` plugin computing the same answer from the same filter code, with the
+reader's own credentials. It answers a different question — what the filter
+*would* decide, seen through the caller's RBAC — and the question worth
+answering is what this agent decided through its own.
+
+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 81de2fd..ccde1c0 100644
--- a/docs/adr/README.md
+++ b/docs/adr/README.md
@@ -141,3 +141,4 @@ line to paste.
 - [0081. A weakened security default is a build fact like any other](0081-a-weakened-default-is-a-build-fact.md)
 - [0082. A confirmed endpoint is named on the port it answered on](0082-a-confirmed-endpoint-is-named-on-its-port.md)
 - [0083. The agent names itself in the request it makes](0083-the-agent-names-itself-in-the-request.md)
+- [0084. What was excluded is named, inside the cluster and only there](0084-what-was-excluded-is-named-inside-the-cluster.md)
diff --git a/docs/prose_test.go b/docs/prose_test.go
index ffc3cdb..37a1cdf 100644
--- a/docs/prose_test.go
+++ b/docs/prose_test.go
@@ -39,9 +39,10 @@ var ceilings = []struct {
 	// (ADR 0078); a connection that confirmed an endpoint once per image now
 	// repeats once a minute per process (ADR 0080); twelve more GODEBUG names
 	// are read out of the build, on a second axis from the two the promise
-	// named before (ADR 0081); and every request now names the agent that made
-	// it (ADR 0083).
-	{"security.md", 1121, 366},
+	// named before (ADR 0081); every request now names the agent that made it
+	// (ADR 0083); and a listener answers inside the cluster what the filters
+	// excluded, by name (ADR 0084).
+	{"security.md", 1124, 366},
 	// Each raise names its obligation here too. The body is encoded, so a
 	// payload's size is no longer the request's (ADR 0076); a window states
 	// whether it is finished (ADR 0078); gauge samples arrive, where adding
diff --git a/docs/security.md b/docs/security.md
index 481ff8a..5548869 100644
--- a/docs/security.md
+++ b/docs/security.md
@@ -212,6 +212,7 @@ collection stage rather than at the API server, and the counts in
 | Backend egress | controller → one address you configure | Ship the payloads listed in [§8](#8-data-collected-and-data-leaving-the-cluster) | **Off unless you set an address, and there is no default** — without one the controller writes to its local spool and sends nothing, which is what every installation does today. Where it is set it is the only cross-boundary connection in the system: one `POST` per payload, the body the spool file verbatim. **Plain HTTP at this stage** — no TLS, no credential, no identity ([§6](#6-agent-identity-and-credential-lifecycle-planned) is unbuilt) — so it is for a backend you control on a network you trust. Nothing in a reply changes what the agent does: only the status code is read ([ADR 0075](adr/0075-a-payload-leaves-the-spool-only-when-the-backend-has-it.md)). **The body is gzip-encoded**, always and without negotiating — the agent cannot ask what the backend accepts — and what it encodes is the spool file unchanged, so what leaves the cluster is what [§8](#8-data-collected-and-data-leaving-the-cluster) lists and nothing else ([ADR 0076](adr/0076-the-payload-travels-encoded.md)). The NetworkPolicy that ships with the chart is **ingress only, by decision and not pending**: an egress rule would have to permit the API server, DNS, and arbitrary pod addresses for endpoint confirmation, which is everything, and a rule permitting everything is a boundary in name only. What bounds this egress is the code — one package opens outbound connections, to one address you set |
 | Health probes | kubelet → both roles, in-cluster only | Answer whether the agent is working and whether it may be depended on yet ([ADR 0069](adr/0069-liveness-is-the-process-readiness-is-the-caches.md)) | **Port 9090 on the controller pod and on every node pod, in every profile** — separate from the report receiver above. Two paths, `/livez` and `/readyz`; GET and HEAD only, and nothing about a request changes what the agent does (§9). The reply is a status code and one line about the agent itself — whether its own collection pass is still running, whether its caches have filled — and names **no** object in your cluster. Not exposed by any Service. Where a NetworkPolicy ships it carries a second rule admitting this port from **any** source, because the caller is your kubelet arriving from the node's address, which no pod selector can name |
 | Agent metrics | your monitoring → both roles, in-cluster only | Alert on the agent itself: whether it is still collecting, which of its reads are failing, what its filters excluded, what its spool evicted ([ADR 0070](adr/0070-the-agent-says-what-it-did-in-the-format-the-cluster-already-reads.md)) | **`/metrics` on the same port 9090 as the probes above, in every profile** — GET and HEAD only, Prometheus text format, and no `ServiceMonitor` or other Prometheus Operator resource is installed. Every number is one the agent already keeps and already reports in `collection_coverage` ([§11](#11-your-controls-and-what-the-agent-says-about-itself)); the two cannot disagree, because they read the same counters. Labels come from a **closed set of nine names**, and no value is ever a namespace, pod, node, workload or image name — the counts are the same aggregates the coverage payload carries. Reached by the same NetworkPolicy rule as the probes, so where a policy is enforced any pod in the cluster may scrape it; what it learns is how many objects this agent observed and how many its filters excluded, never which |
+| Collection listing | you → the controller pod, **loopback only** | Find out which objects your filters admitted and which they refused — by name, with the reason ([ADR 0084](adr/0084-what-was-excluded-is-named-inside-the-cluster.md)) | **`/collection` on `127.0.0.1:9091`, controller only.** Bound to loopback, so no Service, no container port and no NetworkPolicy rule reaches it: the only way in is `kubectl port-forward`, which your own RBAC already gates on `create` of `pods/portforward` in the agent namespace. GET and HEAD only, newline-delimited JSON, and **no query is parsed** — the whole list is always answered, so an empty reply can never be mistaken for "nothing matched". This is the one place the agent names an object your filters excluded, and what it answers **never leaves the cluster**: nothing that can write a payload or open a connection can reach it, and a test on the import graph is what keeps that true. Off with one Helm value. An address that is not loopback is refused at startup rather than opened |
 | Node → controller reports | node → controller, in-cluster only | Deliver on-node Go build-info findings and captured profiles for aggregation (ADR 0010, ADR 0011), and ask which pods on this node passed your filters (ADR 0015) | Plain HTTP on the cluster network; the node always initiates and authenticates with a projected controller-audience ServiceAccount token, validated locally (no `TokenReview`). The report endpoints answer ack/error only. The **scope** query answers from the controller's already-filtered pod index, so it can only narrow what the node scans — it cannot name a pod your filters exclude. The **profiling-target** query is the one reply the node acts on; see [§7.2](#72-the-ebpf-cpu-profiler-opt-in-ebpf-profile-adr-0011) for what bounds it. The receiver is not reachable from outside the cluster, and a NetworkPolicy shipped with the chart restricts it to the node DaemonSet (ADR 0040 — it was described as shipping for months before it did, ADR 0039). Read that policy for what it is: it is enforced by your CNI, so on a cluster whose CNI does not implement NetworkPolicy it is inert, and it restricts who may be *heard*, never what a caller may *say*. The DaemonSet has **no** external egress. Honest disclosure: the token and the (already-filtered) facts travel in cleartext in-cluster, so a party who can sniff pod traffic can replay a node's token until it expires. What that buys is now bounded to **the node it was taken from**: the token names the node its bearer runs on, every endpoint refuses a request that speaks for a different one, and a fact can only join to a pod on the reporting node (ADR 0040). So a replayed or compromised node token can post fabricated inventory facts and profiles about workloads on *that* node, and read which of *that* node's pods passed your filters. It cannot read or write anything about any other node, it cannot reach the API, and it controls nothing |
 
 ---
@@ -1083,6 +1084,8 @@ rather than read back from the rules, so a grant defeated by a webhook reads as
 failing. No `SelfSubjectRulesReview` self-audit is performed: that is a `create`
 call (ADR 0054 §3). A 403 degrades with a log line, never a crash-loop.
 
+**And named, to you, inside your cluster.** The counts are what *we* receive. You can read the same decisions resolved to objects at `/collection` on the controller's loopback address ([§5](#5-network-access)): one line per namespace, pod and Job, whether it is collected, whether it may be profiled, which of your five controls refused it, and whether its workload was one the agent cannot read. It is computed on request from the same caches and the same filter code the agent collects with, so it cannot disagree with what it does. Nothing of it is transmitted, and nothing of it is remembered.
+
 **What those counts can and cannot tell us.** They say how much was excluded and
 under which of your five controls, never which object. One inference remains and
 is stated rather than hidden ([ADR 0039](adr/0039-stated-limits-are-measured-limits.md)):
diff --git a/internal/collector/decisions.go b/internal/collector/decisions.go
new file mode 100644
index 0000000..82b69cd
--- /dev/null
+++ b/internal/collector/decisions.go
@@ -0,0 +1,123 @@
+package collector
+
+import (
+	"sort"
+
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+// Decision is what the filter decided about one object, named. It is answered
+// only inside the cluster and never enters a payload (ADR 0084).
+type Decision struct {
+	Kind      string             `json:"kind"`
+	Namespace string             `json:"namespace,omitempty"`
+	Name      string             `json:"name"`
+	Collected bool               `json:"collected"`
+	Profiled  bool               `json:"profiled"`
+	Reason    ExclusionReason    `json:"reason,omitempty"`
+	Workload  UnresolvedWorkload `json:"workload,omitempty"`
+}
+
+// Decisions re-asks the filter about every object it gates and returns the
+// answers, named. Nothing is remembered to answer this: the caches hold every
+// namespace, pod and Job whether or not it was admitted, so the list is a
+// projection of what the informers already have (invariant 4).
+//
+// Counted nowhere. The counters describe the decisions the agent acted on, and
+// a read of this list is not one of them.
+func (w *PodWatcher) Decisions() []Decision {
+	out := w.namespaceDecisions()
+	out = append(out, w.podDecisions()...)
+	out = append(out, w.jobDecisions()...)
+	sort.Slice(out, func(i, j int) bool {
+		a, b := out[i], out[j]
+		if a.Kind != b.Kind {
+			return a.Kind < b.Kind
+		}
+		if a.Namespace != b.Namespace {
+			return a.Namespace < b.Namespace
+		}
+		return a.Name < b.Name
+	})
+	return out
+}
+
+// namespaceDecisions answers for the namespaces themselves. Their pods carry
+// the same answer one by one, but a namespace with no pods in it would
+// otherwise never show that the configuration reaches it.
+func (w *PodWatcher) namespaceDecisions() []Decision {
+	namespaces, err := w.nsLister.List(labels.Everything())
+	if err != nil {
+		return nil
+	}
+	out := make([]Decision, 0, len(namespaces))
+	for _, ns := range namespaces {
+		collected, reason := w.filter.AdmitNamespace(ns)
+		out = append(out, Decision{
+			Kind:      "Namespace",
+			Name:      ns.Name,
+			Collected: collected,
+			Profiled:  collected && ns.Annotations[ProfileAnnotation] != "false",
+			Reason:    reason,
+		})
+	}
+	return out
+}
+
+func (w *PodWatcher) podDecisions() []Decision {
+	pods, err := w.podLister.List(labels.Everything())
+	if err != nil {
+		return nil
+	}
+	out := make([]Decision, 0, len(pods))
+	for _, pod := range pods {
+		nsAnnotations := w.namespaceAnnotations(pod.Namespace)
+		workload := w.workloadAnnotations(pod)
+		collected, reason := w.filter.AdmitPod(pod, nsAnnotations, workload)
+		out = append(out, Decision{
+			Kind:      "Pod",
+			Namespace: pod.Namespace,
+			Name:      pod.Name,
+			Collected: collected,
+			Profiled:  collected && w.filter.AdmitProfiling(pod, nsAnnotations, workload),
+			Reason:    reason,
+			Workload:  workload.Unresolved,
+		})
+	}
+	return out
+}
+
+// jobDecisions answers for finished-Job facts, which are gated by a filter of
+// their own (ADR 0028 §1). A Job's pods answer for the pods; this answers for
+// `job_runs`, and the two can differ — the Job's pod template is a step only
+// the Job's decision has.
+func (w *PodWatcher) jobDecisions() []Decision {
+	jobs, err := w.jobLister.List(labels.Everything())
+	if err != nil {
+		return nil
+	}
+	out := make([]Decision, 0, len(jobs))
+	for _, job := range jobs {
+		workload := w.jobWorkloadAnnotations(job)
+		collected, reason := w.filter.AdmitJob(job, w.namespaceAnnotations(job.Namespace), workload)
+		out = append(out, Decision{
+			Kind:      "Job",
+			Namespace: job.Namespace,
+			Name:      job.Name,
+			Collected: collected,
+			Reason:    reason,
+			Workload:  workload.Unresolved,
+		})
+	}
+	return out
+}
+
+// namespaceAnnotations reads a namespace's annotations from the cache, where a
+// miss reads as none — the same reading the event path takes.
+func (w *PodWatcher) namespaceAnnotations(namespace string) map[string]string {
+	ns, err := w.nsLister.Get(namespace)
+	if err != nil {
+		return nil
+	}
+	return ns.Annotations
+}
diff --git a/internal/collector/decisions_test.go b/internal/collector/decisions_test.go
new file mode 100644
index 0000000..996e44e
--- /dev/null
+++ b/internal/collector/decisions_test.go
@@ -0,0 +1,208 @@
+package collector
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/RebuildStackCo/runtime-agent/internal/model"
+	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+// The list answers the question the counters cannot: which object, and under
+// which of the customer's controls. It is read inside the cluster and never
+// shipped (ADR 0084), so what it must get right is the reason attached to each
+// name — a list that named the wrong control would send someone editing an
+// object they never wrote on.
+
+func namespaceObj(name string, annotations map[string]string) *corev1.Namespace {
+	return &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: name, Annotations: annotations}}
+}
+
+// decisionsWatcher runs a watcher over the objects and waits until the filter
+// has decided about every pod among them.
+func decisionsWatcher(t *testing.T, filter *Filter, wantDecided int64, objects ...runtime.Object) *PodWatcher {
+	t.Helper()
+	clientset := fake.NewClientset(objects...)
+	watcher := NewPodWatcher(clientset, func(model.PodInfo) {})
+	watcher.SetFilter(filter)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	go func() { _ = watcher.Run(ctx) }()
+
+	waitFor(t, 5*time.Second, "the filter to decide about every pod", func() bool {
+		c := filter.Snapshot()
+		return c.PodsObserved+c.ExcludedNamespaceFilter+c.ExcludedNamespaceAnnotation+
+			c.ExcludedWorkloadAnnotation+c.ExcludedPodAnnotation >= wantDecided
+	})
+	return watcher
+}
+
+// find returns the decision about one object, so a test names what it asserts
+// about rather than an index into a sorted list.
+func find(t *testing.T, list []Decision, kind, namespace, name string) Decision {
+	t.Helper()
+	for _, d := range list {
+		if d.Kind == kind && d.Namespace == namespace && d.Name == name {
+			return d
+		}
+	}
+	t.Fatalf("no decision about %s %s/%s in %+v", kind, namespace, name, list)
+	return Decision{}
+}
+
+func TestEveryControlIsNamedOnTheObjectThatCarriesIt(t *testing.T) {
+	rs := &appsv1.ReplicaSet{ObjectMeta: metav1.ObjectMeta{
+		Namespace: "shop", Name: "web-7d9f", UID: "uid-web-7d9f",
+		OwnerReferences: []metav1.OwnerReference{controllerRef("Deployment", "web")},
+	}}
+	optedOutDeployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{
+		Namespace: "shop", Name: "web", UID: "uid-web",
+		Annotations: map[string]string{CollectAnnotation: "false"},
+	}}
+	owner := controllerRef("ReplicaSet", "web-7d9f")
+
+	byWorkload := pod("by-workload", &owner)
+	byPod := pod("by-pod", nil)
+	byPod.Annotations = map[string]string{CollectAnnotation: "false"}
+	collected := pod("collected", nil)
+	notProfiled := pod("not-profiled", nil)
+	notProfiled.Annotations = map[string]string{ProfileAnnotation: "false"}
+
+	inDeniedNamespace := pod("in-denied", nil)
+	inDeniedNamespace.Namespace = "kube-system"
+	inDeniedNamespace.UID = "uid-in-denied"
+
+	inOptedOutNamespace := pod("in-opted-out", nil)
+	inOptedOutNamespace.Namespace = "lab"
+	inOptedOutNamespace.UID = "uid-in-opted-out"
+
+	filter := NewFilter(nil, []string{"kube-system"})
+	watcher := decisionsWatcher(t, filter, 6,
+		namespaceObj("shop", nil),
+		namespaceObj("kube-system", nil),
+		namespaceObj("lab", map[string]string{CollectAnnotation: "false"}),
+		rs, optedOutDeployment,
+		byWorkload, byPod, collected, notProfiled, inDeniedNamespace, inOptedOutNamespace,
+	)
+
+	list := watcher.Decisions()
+	for _, want := range []struct {
+		namespace, name string
+		collected       bool
+		reason          ExclusionReason
+	}{
+		{"shop", "collected", true, ""},
+		{"shop", "not-profiled", true, ""},
+		{"shop", "by-workload", false, ExcludedByWorkloadAnnotation},
+		{"shop", "by-pod", false, ExcludedByPodAnnotation},
+		{"kube-system", "in-denied", false, ExcludedByNamespaceFilter},
+		{"lab", "in-opted-out", false, ExcludedByNamespaceAnnotation},
+	} {
+		got := find(t, list, "Pod", want.namespace, want.name)
+		if got.Collected != want.collected || got.Reason != want.reason {
+			t.Errorf("%s/%s = collected %v reason %q, want %v and %q",
+				want.namespace, want.name, got.Collected, got.Reason, want.collected, want.reason)
+		}
+	}
+
+	if d := find(t, list, "Pod", "shop", "not-profiled"); d.Profiled {
+		t.Error("the pod that refused the profiler is listed as profiled")
+	}
+	if d := find(t, list, "Pod", "shop", "collected"); !d.Profiled {
+		t.Error("a pod that refused nothing is listed as not profiled")
+	}
+	// An excluded pod is not profiled either, and saying so twice is how a
+	// reader learns that the narrower control did not have to be used.
+	if d := find(t, list, "Pod", "shop", "by-pod"); d.Profiled {
+		t.Error("an excluded pod is listed as profiled")
+	}
+}
+
+// A namespace nothing runs in still says whether the configuration reaches it.
+func TestAnEmptyNamespaceStillStatesItsAnswer(t *testing.T) {
+	filter := NewFilter([]string{"shop"}, nil)
+	watcher := decisionsWatcher(t, filter, 1,
+		namespaceObj("shop", nil),
+		namespaceObj("empty", nil),
+		pod("collected", nil),
+	)
+
+	list := watcher.Decisions()
+	if d := find(t, list, "Namespace", "", "empty"); d.Collected || d.Reason != ExcludedByNamespaceFilter {
+		t.Errorf("empty namespace = %+v, want excluded by the namespace filter", d)
+	}
+	if d := find(t, list, "Namespace", "", "shop"); !d.Collected {
+		t.Errorf("shop = %+v, want collected", d)
+	}
+}
+
+// The blind spot ADR 0028 counts is named here as well: this is where a
+// customer finds out that their workload annotation could not be read.
+func TestAPodWhoseControllerCannotBeReadSaysSo(t *testing.T) {
+	owner := controllerRef("Rollout", "web")
+	underRollout := pod("under-rollout", &owner)
+
+	filter := NewFilter(nil, nil)
+	watcher := decisionsWatcher(t, filter, 1, namespaceObj("shop", nil), underRollout)
+
+	d := find(t, watcher.Decisions(), "Pod", "shop", "under-rollout")
+	if !d.Collected {
+		t.Errorf("%+v, want collected: an unreadable controller fails open (ADR 0028)", d)
+	}
+	if d.Workload != WorkloadKindUnknown {
+		t.Errorf("workload = %q, want %q", d.Workload, WorkloadKindUnknown)
+	}
+}
+
+// A Job has a filter step its pods do not — the pod template, which is where an
+// opt-out written the pre-ADR-0028 way lives. The list answers for both.
+func TestAJobAnswersForItsOwnPayload(t *testing.T) {
+	job := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "nightly", UID: "uid-nightly"},
+		Spec: batchv1.JobSpec{Template: corev1.PodTemplateSpec{
+			ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{CollectAnnotation: "false"}},
+		}},
+	}
+	filter := NewFilter(nil, nil)
+	watcher := decisionsWatcher(t, filter, 0, namespaceObj("shop", nil), job)
+
+	waitFor(t, 5*time.Second, "the Job cache to fill", func() bool {
+		for _, d := range watcher.Decisions() {
+			if d.Kind == "Job" {
+				return true
+			}
+		}
+		return false
+	})
+	d := find(t, watcher.Decisions(), "Job", "shop", "nightly")
+	if d.Collected || d.Reason != ExcludedByObjectAnnotation {
+		t.Errorf("%+v, want excluded by the object's own annotation", d)
+	}
+}
+
+// Answering must not move a number. The counters describe what the agent acted
+// on, and reading the list is not one of those decisions (ADR 0054).
+func TestReadingTheListCountsNothing(t *testing.T) {
+	filter := NewFilter(nil, []string{"kube-system"})
+	denied := pod("in-denied", nil)
+	denied.Namespace = "kube-system"
+	watcher := decisionsWatcher(t, filter, 2,
+		namespaceObj("shop", nil), namespaceObj("kube-system", nil),
+		pod("collected", nil), denied,
+	)
+
+	before := filter.Snapshot()
+	for range 3 {
+		watcher.Decisions()
+	}
+	if after := filter.Snapshot(); after != before {
+		t.Errorf("coverage moved from %+v to %+v by being read", before, after)
+	}
+}
diff --git a/internal/collector/filter.go b/internal/collector/filter.go
index efb2c8b..eba2459 100644
--- a/internal/collector/filter.go
+++ b/internal/collector/filter.go
@@ -138,6 +138,20 @@ func (f *Filter) AdmitPod(pod *corev1.Pod, nsAnnotations map[string]string, work
 	return true, ""
 }
 
+// AdmitNamespace reports whether a namespace passes the two controls that
+// apply to it. The pods inside it are asked the same questions one by one and
+// reach the same answer; this is asked so that a namespace holding no pods
+// still says whether the configuration reaches it (ADR 0084).
+func (f *Filter) AdmitNamespace(ns *corev1.Namespace) (bool, ExclusionReason) {
+	if !f.namespaceAllowed(ns.Name) {
+		return false, ExcludedByNamespaceFilter
+	}
+	if ns.Annotations[CollectAnnotation] == "false" {
+		return false, ExcludedByNamespaceAnnotation
+	}
+	return true, ""
+}
+
 // AdmitProfiling reports whether an admitted pod may be profiled. It is asked
 // only of pods AdmitPod admitted, which is what makes this control the narrower
 // one rather than a second one beside it.
diff --git a/internal/collector/jobs.go b/internal/collector/jobs.go
index 31acec8..8f1e411 100644
--- a/internal/collector/jobs.go
+++ b/internal/collector/jobs.go
@@ -91,10 +91,7 @@ func (w *PodWatcher) resolveJobWorkload(job *batchv1.Job) model.WorkloadRef {
 // security.md documented before ADR 0028 wrote the annotation into the pod
 // template, and ignoring it would ship facts about a workload they refused.
 func (w *PodWatcher) admitJob(job *batchv1.Job, count bool) bool {
-	var nsAnnotations map[string]string
-	if ns, err := w.nsLister.Get(job.Namespace); err == nil {
-		nsAnnotations = ns.Annotations
-	}
+	nsAnnotations := w.namespaceAnnotations(job.Namespace)
 	workload := w.jobWorkloadAnnotations(job)
 	allowed, reason := w.filter.AdmitJob(job, nsAnnotations, workload)
 	if count {
diff --git a/internal/collector/pods.go b/internal/collector/pods.go
index 4eb8f03..04aae06 100644
--- a/internal/collector/pods.go
+++ b/internal/collector/pods.go
@@ -857,10 +857,7 @@ type admission struct {
 // opt-out reads the same three objects, and asking twice would double the
 // lookups for a question the first pass already has the answers to.
 func (w *PodWatcher) admit(pod *corev1.Pod, count bool) admission {
-	var nsAnnotations map[string]string
-	if ns, err := w.nsLister.Get(pod.Namespace); err == nil {
-		nsAnnotations = ns.Annotations
-	}
+	nsAnnotations := w.namespaceAnnotations(pod.Namespace)
 	workload := w.workloadAnnotations(pod)
 	allowed, reason := w.filter.AdmitPod(pod, nsAnnotations, workload)
 	profilable := allowed && w.filter.AdmitProfiling(pod, nsAnnotations, workload)
diff --git a/internal/config/config.go b/internal/config/config.go
index 3d2b38a..1dd5242 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -5,6 +5,7 @@ package config
 
 import (
 	"fmt"
+	"net"
 	"net/url"
 	"os"
 
@@ -24,6 +25,7 @@ type Config struct {
 	NodeIntake NodeIntake          `json:"nodeIntake"`
 	Profiling  ControllerProfiling `json:"profiling"`
 	Health     Health              `json:"health"`
+	Collection Collection          `json:"collection"`
 }
 
 // Health is where the controller answers the kubelet's two questions about it
@@ -42,6 +44,21 @@ type Health struct {
 // opened by omission.
 const DefaultHealthListenAddress = ":9090"
 
+// Collection is the in-cluster listener that answers which objects the filters
+// admitted and which they refused, by name (ADR 0084).
+type Collection struct {
+	// ListenAddress is where that listener binds (net/http form). Empty opens
+	// no listener. The host must be a loopback address: this is the one
+	// surface that names excluded objects, and Validate refuses to open it
+	// anywhere a pod in the cluster could reach.
+	ListenAddress string `json:"listenAddress"`
+}
+
+// DefaultCollectionListenAddress is the address the chart renders. Loopback,
+// so the only way in is a port-forward, whose permission Kubernetes already
+// defines.
+const DefaultCollectionListenAddress = "127.0.0.1:9091"
+
 // NodeConfig is the root of the node role's configuration file. It holds only
 // what the node enforces itself; everything else about profiling is the
 // controller's (see the note above).
@@ -292,9 +309,34 @@ func (c Config) Validate() error {
 	if err := validThirdPartySymbols(c.Profiling.ThirdPartySymbols); err != nil {
 		return err
 	}
+	if err := validCollectionListenAddress(c.Collection.ListenAddress); err != nil {
+		return err
+	}
 	return validBackendBaseURL(c.Backend.BaseURL)
 }
 
+// validCollectionListenAddress accepts an empty value, which opens nothing, and
+// otherwise an address whose host is loopback. A bare port or a wildcard host is
+// refused rather than normalized: what the listener answers is names the rest of
+// this agent spends its decisions keeping inside the cluster, and a value that
+// exposed it to every pod would be a decision made by a typo (ADR 0084).
+func validCollectionListenAddress(raw string) error {
+	if raw == "" {
+		return nil
+	}
+	host, _, err := net.SplitHostPort(raw)
+	if err != nil {
+		return fmt.Errorf("collection.listenAddress %q is not host:port: %w", raw, err)
+	}
+	if host == "localhost" {
+		return nil
+	}
+	if ip := net.ParseIP(host); ip != nil && ip.IsLoopback() {
+		return nil
+	}
+	return fmt.Errorf("collection.listenAddress %q must bind a loopback host; it names objects your filters excluded", raw)
+}
+
 // validBackendBaseURL accepts an empty value, which ships nothing, and
 // otherwise an absolute http or https URL. A relative or scheme-less address is
 // a startup failure rather than a shipper that builds unsendable requests every
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index fbe8040..72171ce 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -349,3 +349,38 @@ func TestTheShapeCarriesNoAddress(t *testing.T) {
 		t.Errorf("the configuration shape carries the backend address: %s", encoded)
 	}
 }
+
+// The listing names objects the filters excluded, so where it binds is the whole
+// of what keeps those names in the cluster. A host that is not loopback is a
+// startup failure rather than a value normalized into something safe: the
+// operator asked for something this agent does not offer, and finding that out
+// at startup is the point (ADR 0084).
+func TestTheCollectionListenerBindsLoopbackOrNothing(t *testing.T) {
+	for _, addr := range []string{"127.0.0.1:9091", "[::1]:9091", "localhost:9091"} {
+		cfg, err := Load(write(t, "collection:\n  listenAddress: \""+addr+"\"\n"))
+		if err != nil {
+			t.Errorf("Load(%q): %v", addr, err)
+			continue
+		}
+		if cfg.Collection.ListenAddress != addr {
+			t.Errorf("collection.listenAddress = %q, want %q", cfg.Collection.ListenAddress, addr)
+		}
+	}
+
+	for _, addr := range []string{":9091", "0.0.0.0:9091", "10.0.0.4:9091", "9091"} {
+		if _, err := Load(write(t, "collection:\n  listenAddress: \""+addr+"\"\n")); err == nil {
+			t.Errorf("Load accepted %q; a listener reachable from the cluster names excluded objects to it", addr)
+		} else if !strings.Contains(err.Error(), "collection.listenAddress") {
+			t.Errorf("the refusal of %q does not name the field: %v", addr, err)
+		}
+	}
+
+	// Empty opens nothing, the same shape the health listener has.
+	empty, err := Load(write(t, "spool:\n  dir: /tmp\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if empty.Collection.ListenAddress != "" {
+		t.Errorf("an unset collection.listenAddress became %q", empty.Collection.ListenAddress)
+	}
+}
diff --git a/internal/decisions/isolation_test.go b/internal/decisions/isolation_test.go
new file mode 100644
index 0000000..4f930a0
--- /dev/null
+++ b/internal/decisions/isolation_test.go
@@ -0,0 +1,74 @@
+package decisions
+
+import (
+	"go/parser"
+	"go/token"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// The names this package answers are the names every other decision in this
+// repository works to keep inside the cluster. What stops them from leaving is
+// not the handler's code — it is that nothing which can transmit knows this
+// package exists (ADR 0084). That is a property of the import graph, so it is
+// checked as one.
+
+// transmitters are the packages that can put bytes on the wire or in a spool
+// file. None of them may reach what is served here, and this package may not
+// reach them.
+var transmitters = []string{"internal/sink", "internal/shipper", "internal/journal"}
+
+// importsOf reads every Go file in dir, build tags and all: a file this walk
+// skipped would be a file free to import what the rest may not.
+func importsOf(t *testing.T, dir string) map[string][]string {
+	t.Helper()
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatalf("reading %s: %v", dir, err)
+	}
+	fset := token.NewFileSet()
+	out := make(map[string][]string)
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".go") {
+			continue
+		}
+		file, err := parser.ParseFile(fset, filepath.Join(dir, entry.Name()), nil, parser.ImportsOnly)
+		if err != nil {
+			t.Fatalf("parsing %s: %v", entry.Name(), err)
+		}
+		for _, imp := range file.Imports {
+			out[entry.Name()] = append(out[entry.Name()], strings.Trim(imp.Path.Value, `"`))
+		}
+	}
+	if len(out) == 0 {
+		t.Fatalf("%s holds no Go files; this check would pass vacuously", dir)
+	}
+	return out
+}
+
+func TestNothingThatCanTransmitKnowsThisPackage(t *testing.T) {
+	for _, dir := range transmitters {
+		for file, imports := range importsOf(t, filepath.Join("..", filepath.Base(dir))) {
+			for _, imp := range imports {
+				if strings.HasSuffix(imp, "/internal/decisions") {
+					t.Errorf("%s/%s imports the decision listing; what it answers names objects "+
+						"your filters excluded and must not reach a payload (ADR 0084)", dir, file)
+				}
+			}
+		}
+	}
+}
+
+func TestThisPackageReachesNothingThatCanTransmit(t *testing.T) {
+	for file, imports := range importsOf(t, ".") {
+		for _, imp := range imports {
+			for _, banned := range transmitters {
+				if strings.Contains(imp, banned) {
+					t.Errorf("%s imports %s; this listener answers inside the cluster only (ADR 0084)", file, imp)
+				}
+			}
+		}
+	}
+}
diff --git a/internal/decisions/server.go b/internal/decisions/server.go
new file mode 100644
index 0000000..639bd26
--- /dev/null
+++ b/internal/decisions/server.go
@@ -0,0 +1,116 @@
+// Package decisions answers, inside the cluster and only inside it, which
+// objects the agent's filters admitted and which they refused — by name, with
+// the reason (ADR 0084).
+//
+// It is the one surface of this agent that names an object the filters
+// excluded. That is why it is a listener of its own, bound to loopback, and
+// why nothing here is reachable from anywhere a payload could be built: the
+// names answered here never leave the cluster (CLAUDE.md invariant 6).
+package decisions
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/RebuildStackCo/runtime-agent/internal/collector"
+)
+
+// Path is where the list is served.
+const Path = "/collection"
+
+// Handler answers the list as newline-delimited JSON, one object per line, so
+// a reader can process a large cluster's answer as it arrives rather than
+// waiting for one document to be built and parsed.
+//
+// It reads nothing from the request but its method: no body, no query, no
+// header. A filter is `jq`'s job, and a parameter here would be a request this
+// endpoint acts on (CLAUDE.md invariant 1).
+func Handler(list func() []collector.Decision, synced func() bool, logger *slog.Logger) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet && r.Method != http.MethodHead {
+			w.Header().Set("Allow", "GET, HEAD")
+			http.Error(w, "read-only endpoint", http.StatusMethodNotAllowed)
+			return
+		}
+		// An unsynced cache would answer a short list that looks like a
+		// complete one, which is the failure this endpoint exists to prevent.
+		if !synced() {
+			http.Error(w, "informer caches are still syncing", http.StatusServiceUnavailable)
+			return
+		}
+		w.Header().Set("Content-Type", "application/x-ndjson; charset=utf-8")
+		w.WriteHeader(http.StatusOK)
+		if r.Method == http.MethodHead {
+			return
+		}
+		enc := json.NewEncoder(w)
+		for _, d := range list() {
+			if err := enc.Encode(d); err != nil {
+				// The status line is already sent, so the reader learns of the
+				// truncation from the stream ending early.
+				logger.Warn("collection listing cut short", "error", err)
+				return
+			}
+		}
+	})
+}
+
+// Server runs the listener until its context is canceled: one more long-lived
+// task in the controller's lifecycle, beside the health listener rather than
+// on it.
+type Server struct {
+	addr    string
+	handler http.Handler
+	logger  *slog.Logger
+}
+
+// New wires a listener on addr serving the list at Path and nothing else.
+func New(addr string, h http.Handler, logger *slog.Logger) *Server {
+	mux := http.NewServeMux()
+	mux.Handle(Path, h)
+	return &Server{addr: addr, handler: mux, logger: logger}
+}
+
+// Run serves until ctx is canceled, then shuts down. A listener that cannot
+// bind is a startup failure, as the health listener's is: an endpoint the
+// operator configured and that silently does not exist is worse than a refusal
+// to start.
+func (s *Server) Run(ctx context.Context) error {
+	srv := &http.Server{
+		Addr:              s.addr,
+		Handler:           s.handler,
+		ReadHeaderTimeout: 5 * time.Second,
+		ReadTimeout:       5 * time.Second,
+		// Long enough to write a large cluster's list to a reader that is
+		// paging it through a port-forward.
+		WriteTimeout: 60 * time.Second,
+		IdleTimeout:  30 * time.Second,
+	}
+
+	serveErr := make(chan error, 1)
+	go func() {
+		s.logger.Info("collection listener serving", "addr", s.addr, "path", Path)
+		serveErr <- srv.ListenAndServe()
+	}()
+
+	select {
+	case <-ctx.Done():
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := srv.Shutdown(shutdownCtx); err != nil {
+			return fmt.Errorf("collection listener shutdown: %w", err)
+		}
+		s.logger.Info("collection listener stopped")
+		return nil
+	case err := <-serveErr:
+		if errors.Is(err, http.ErrServerClosed) {
+			return nil
+		}
+		return fmt.Errorf("collection listener serve: %w", err)
+	}
+}
diff --git a/internal/decisions/server_test.go b/internal/decisions/server_test.go
new file mode 100644
index 0000000..81abeea
--- /dev/null
+++ b/internal/decisions/server_test.go
@@ -0,0 +1,125 @@
+package decisions
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/RebuildStackCo/runtime-agent/internal/collector"
+)
+
+func testHandler(list []collector.Decision, synced bool) http.Handler {
+	return Handler(
+		func() []collector.Decision { return list },
+		func() bool { return synced },
+		slog.New(slog.NewTextHandler(io.Discard, nil)),
+	)
+}
+
+var sample = []collector.Decision{
+	{Kind: "Namespace", Name: "shop", Collected: true, Profiled: true},
+	{Kind: "Pod", Namespace: "shop", Name: "web-a", Collected: true, Profiled: true},
+	{Kind: "Pod", Namespace: "shop", Name: "web-b", Collected: false,
+		Reason: collector.ExcludedByPodAnnotation},
+}
+
+func TestTheListIsOneObjectPerLine(t *testing.T) {
+	rec := httptest.NewRecorder()
+	testHandler(sample, true).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, Path, nil))
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", rec.Code)
+	}
+	if ct := rec.Header().Get("Content-Type"); !strings.HasPrefix(ct, "application/x-ndjson") {
+		t.Errorf("content type = %q, want ndjson", ct)
+	}
+
+	var got []collector.Decision
+	scan := bufio.NewScanner(rec.Body)
+	for scan.Scan() {
+		var d collector.Decision
+		if err := json.Unmarshal(scan.Bytes(), &d); err != nil {
+			t.Fatalf("line %q is not an object: %v", scan.Text(), err)
+		}
+		got = append(got, d)
+	}
+	if len(got) != len(sample) {
+		t.Fatalf("read %d lines, want %d", len(got), len(sample))
+	}
+	if got[2].Reason != collector.ExcludedByPodAnnotation {
+		t.Errorf("reason = %q, want the pod annotation", got[2].Reason)
+	}
+}
+
+// The reason is the point of the endpoint, so an absent one has to be absent
+// rather than an empty string a reader has to interpret.
+func TestACollectedObjectCarriesNoReason(t *testing.T) {
+	rec := httptest.NewRecorder()
+	testHandler(sample[:1], true).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, Path, nil))
+
+	if strings.Contains(rec.Body.String(), "reason") {
+		t.Errorf("a collected object was written with a reason field: %s", rec.Body.String())
+	}
+}
+
+// Nothing about a request changes what this endpoint does, and a method that
+// is not a read is refused rather than treated as one (invariant 1).
+func TestOnlyReadsAreAnswered(t *testing.T) {
+	for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} {
+		rec := httptest.NewRecorder()
+		testHandler(sample, true).ServeHTTP(rec, httptest.NewRequest(method, Path, strings.NewReader("{}")))
+		if rec.Code != http.StatusMethodNotAllowed {
+			t.Errorf("%s = %d, want 405", method, rec.Code)
+		}
+	}
+
+	rec := httptest.NewRecorder()
+	testHandler(sample, true).ServeHTTP(rec, httptest.NewRequest(http.MethodHead, Path, nil))
+	if rec.Code != http.StatusOK || rec.Body.Len() != 0 {
+		t.Errorf("HEAD = %d with %d bytes, want 200 and no body", rec.Code, rec.Body.Len())
+	}
+}
+
+// A query is not a filter here. Everything is answered whatever the URL says,
+// so a reader cannot come to believe an empty answer means nothing matched.
+func TestAQueryChangesNothing(t *testing.T) {
+	rec := httptest.NewRecorder()
+	testHandler(sample, true).ServeHTTP(rec,
+		httptest.NewRequest(http.MethodGet, Path+"?only=excluded&namespace=nowhere", nil))
+
+	if lines := strings.Count(strings.TrimSpace(rec.Body.String()), "\n") + 1; lines != len(sample) {
+		t.Errorf("read %d lines with a query, want all %d", lines, len(sample))
+	}
+}
+
+// A partial list is the failure this endpoint exists to prevent: it would look
+// exactly like a cluster with fewer objects in it.
+func TestAnUnsyncedCacheAnswersNothing(t *testing.T) {
+	rec := httptest.NewRecorder()
+	testHandler(sample, false).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, Path, nil))
+
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Errorf("status = %d, want 503 while the caches are filling", rec.Code)
+	}
+	if strings.Contains(rec.Body.String(), "shop") {
+		t.Errorf("an unsynced answer named an object: %s", rec.Body.String())
+	}
+}
+
+// Only Path is served. A listener that answered elsewhere would be a second
+// surface nobody decided on.
+func TestNothingElseIsServed(t *testing.T) {
+	srv := New("127.0.0.1:0", testHandler(sample, true),
+		slog.New(slog.NewTextHandler(io.Discard, nil)))
+
+	rec := httptest.NewRecorder()
+	srv.handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil))
+	if rec.Code != http.StatusNotFound {
+		t.Errorf("/ = %d, want 404", rec.Code)
+	}
+}