refactor: the vocabulary of a collected fact is a package (ADR 0065)

cmd/agent/main.go

@@ -33,6 +33,7 @@ import (
 	"github.com/RebuildStackCo/runtime-agent/internal/inventory"
 	"github.com/RebuildStackCo/runtime-agent/internal/journal"
 	"github.com/RebuildStackCo/runtime-agent/internal/metadata"
+	"github.com/RebuildStackCo/runtime-agent/internal/model"
 	"github.com/RebuildStackCo/runtime-agent/internal/nodeauth"
 	"github.com/RebuildStackCo/runtime-agent/internal/nodeintake"
 	"github.com/RebuildStackCo/runtime-agent/internal/nodeprofile"
@@ -179,7 +180,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 
 	filter := collector.NewFilter(cfg.Filters.Namespaces.Allow, cfg.Filters.Namespaces.Deny)
 
-	podWatcher := collector.NewPodWatcher(clientset, func(p collector.PodInfo) {
+	podWatcher := collector.NewPodWatcher(clientset, func(p model.PodInfo) {
 		logger.Info("pod observed",
 			"namespace", p.Namespace,
 			"pod", p.Name,
@@ -191,7 +192,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 			"containers", p.Containers,
 		)
 	})
-	podWatcher.OnOOMKill(func(o collector.OOMKill) {
+	podWatcher.OnOOMKill(func(o model.OOMKill) {
 		logger.Warn("oom kill observed",
 			"namespace", o.Namespace,
 			"pod", o.Pod,
@@ -213,7 +214,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 	// emitting one payload per restart: a crash loop must not put the spool's
 	// file count under its own control (ADR 0020).
 	restartJournal := journal.NewRestarts(collector.UsageWindowLength)
-	podWatcher.OnContainerRestart(func(r collector.ContainerRestart) {
+	podWatcher.OnContainerRestart(func(r model.ContainerRestart) {
 		logger.Info("container restarted",
 			"namespace", r.Namespace,
 			"pod", r.Pod,
@@ -230,7 +231,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 	// preempted or evicted once, so the accumulator holds records rather than
 	// counters (ADR 0021).
 	disruptionJournal := journal.NewDisruptions(collector.UsageWindowLength)
-	podWatcher.OnPodDisruption(func(d collector.PodDisruption) {
+	podWatcher.OnPodDisruption(func(d model.PodDisruption) {
 		logger.Warn("pod disrupted",
 			"namespace", d.Namespace,
 			"pod", d.Pod,
@@ -248,7 +249,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 	// supply for a workload that ran for part of a window (ADR 0029).
 	jobJournal := journal.NewJobRuns(collector.UsageWindowLength)
 	nodeJournal := journal.NewNodeEvents(collector.UsageWindowLength)
-	podWatcher.OnJobFinished(func(r collector.JobRun) {
+	podWatcher.OnJobFinished(func(r model.JobRun) {
 		logger.Info("job run finished",
 			"namespace", r.Namespace,
 			"workload_kind", r.Workload.Kind,
@@ -264,7 +265,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 		jobJournal.Observe(r)
 	})
 	podWatcher.SetFilter(filter)
-	nodeWatcher := collector.NewNodeWatcher(clientset, func(n collector.NodeInfo) {
+	nodeWatcher := collector.NewNodeWatcher(clientset, func(n model.NodeInfo) {
 		logger.Info("node observed",
 			"node", n.Name,
 			"instance_type", n.InstanceType,
@@ -281,7 +282,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 			"taints", len(n.Taints),
 		)
 	})
-	nodeWatcher.OnNodeLifecycle(func(e collector.NodeLifecycle) {
+	nodeWatcher.OnNodeLifecycle(func(e model.NodeLifecycle) {
 		event := "node left"
 		if e.Joined {
 			event = "node joined"
@@ -375,7 +376,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 	if goStore != nil && cfg.Profiling.Pprof.Enabled {
 		prober = pprofprobe.New(func(c pprofprobe.Candidate) (string, bool) {
 			ip, ok := podWatcher.PodAddress(c.Namespace,
-				collector.WorkloadRef{Kind: c.WorkloadKind, Name: c.WorkloadName},
+				model.WorkloadRef{Kind: c.WorkloadKind, Name: c.WorkloadName},
 				c.Container, c.ImageDigest)
 			if !ok {
 				return "", false
@@ -390,7 +391,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 	if prober != nil && cfg.Profiling.Pprof.Pull && spool != nil {
 		address := func(c pprofprobe.Candidate) (string, bool) {
 			ip, ok := podWatcher.PodAddress(c.Namespace,
-				collector.WorkloadRef{Kind: c.WorkloadKind, Name: c.WorkloadName},
+				model.WorkloadRef{Kind: c.WorkloadKind, Name: c.WorkloadName},
 				c.Container, c.ImageDigest)
 			if !ok {
 				return "", false

docs/adr/0065-the-vocabulary-is-a-package.md

@@ -0,0 +1,96 @@
+# 0065. The vocabulary of a collected fact is a package, and it does not know Kubernetes
+
+Date: 2026-09-03
+
+Status: Accepted
+
+Moves the plain-data types that describe collected facts — `WorkloadRef`,
+`PodInfo`, `NodeInfo`, the journal events, the policy reductions, the coverage
+counters — out of `internal/collector` into `internal/model`. Nothing else
+changes: the golden payload bytes are identical, which is the property that
+makes a change this wide reviewable at all.
+
+## Context
+
+`internal/collector` had grown into three packages sharing one name: the
+vocabulary of collected facts, the Kubernetes adapter that fills it in
+(informers, listers, the kubelet poller), and the reductions that shrink an
+object to what ships. Only the middle one needs a cluster.
+
+The vocabulary is what every other package speaks. `journal` windows restart
+counts, `metadata` aggregates pod shapes, `revisions` orders ReplicaSets,
+`sink` serializes all of it — none of them opens a connection, and every one of
+them imported `collector` to name a type. Measured on the tree before this
+change:
+
+| package | packages in its transitive closure | of them client-go |
+| --- | --- | --- |
+| `rollup` | 1 | 0 |
+| `journal`, `metadata`, `revisions` | 458 | 261 |
+| `sink` | 486 | 261 |
+
+`rollup` is the control. It is window arithmetic that never needed to name a
+workload, so it stayed free — which is the evidence that the weight above is
+not the price of the domain but the price of where the names happened to live.
+Nothing about windowing a restart count requires 261 packages of Kubernetes
+client; it requires the identifier `WorkloadRef`.
+
+Build weight is the symptom and not the reason. The reason is that no boundary
+stood between a Kubernetes type and a payload struct. `PodInfo` and its
+neighbours were already free of `corev1` — every field is a string, a number or
+a pointer to one — but that was a convention held up by review, in the one
+direction where a lapse is not a refactor but a wire-format change. The
+extractors that read `corev1` and the structs they fill sat in the same files,
+so the only thing separating them was which line a reader was on.
+
+## Decision
+
+`internal/model` holds the vocabulary: 48 exported declarations, 894 lines, no
+imports beyond `time` and `reflect`. A test in the package asserts that last
+part, because it is the whole point and nothing else in the build would fail if
+it stopped being true.
+
+`internal/collector` keeps what needs a cluster — the watchers, the kubelet
+poller, the filter, and every reduction from an API object down to a model type
+— and drops from 5801 to 4946 lines and from 95 to 48 exported declarations. It
+is still three packages in one; splitting the adapter from the reductions is a
+separate change this one does not make.
+
+Two methods could not travel unchanged, and the split is along the same seam:
+
+- `NodeInfo.sameAs` became `model.NodeInfo.SameAs`. Whether two collected views
+  are the same fact is a question about the type, so it goes with the type, and
+  exporting it is the cost of that.
+- `ResourceAmounts.empty` became `amountsEmpty` in `collector`. Whether a
+  reduced amount set is worth emitting is a question the reduction asks, not a
+  property of the value.
+
+Alternatives considered and rejected:
+
+- **Aliases in `collector`** (`type WorkloadRef = model.WorkloadRef`) would have
+  made the diff a tenth of the size and let consumers migrate one at a time. It
+  leaves two names for one type — and it leaves `journal` free to keep importing
+  `collector`, which is the thing being prevented. A boundary that permits the
+  old path is a suggestion.
+- **Splitting `collector` by concern first** (watch / kubelet / reduce) is the
+  larger and more interesting change, but each of those parts still needs the
+  vocabulary, so this one comes first regardless.
+- **Leaving it** was viable — nothing was broken. It is rejected because the
+  boundary that was missing is the one between "what the cluster said" and "what
+  we ship", and that is the boundary the whole agent is about.
+
+## Consequences
+
+The packages that reduce, window and serialize facts no longer link a
+Kubernetes client: `journal`, `metadata` and `revisions` go from 458 packages
+to 2, `sink` from 486 to 37, and `inventory`, `targeting` and `pprofprobe` fall
+with them. `cmd/agent` is now the only package that imports `collector`.
+
+A payload type can no longer quietly acquire a `corev1` field, because the
+package it lives in cannot see `corev1`.
+
+What is given up: one more hop. A reader of `collector` who wants a type's
+definition now changes files, and a fact's shape and the code that fills it are
+no longer adjacent. The doc comments went with the types, so the explanation
+travels with the shape rather than with the extraction — which is the right half
+to keep them next to, but it is a real loss for whoever is reading the extractor.

docs/adr/README.md

@@ -122,3 +122,4 @@ line to paste.
 - [0062. The counters are subtracted on the node, and never leave without their interval](0062-the-counters-and-the-window-they-cover.md)
 - [0063. A profile the collector dominates is still a profile](0063-a-profile-of-the-collector-is-still-a-profile.md)
 - [0064. The node is described, and its comings and goings recorded](0064-the-node-is-described-and-its-comings-recorded.md)
+- [0065. The vocabulary of a collected fact is a package, and it does not know Kubernetes](0065-the-vocabulary-is-a-package.md)

internal/collector/cachetransform_test.go

@@ -6,6 +6,7 @@ import (
 	"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"
@@ -225,7 +226,7 @@ func TestTheStrippedObjectIsWhatTheListersServe(t *testing.T) {
 	p.Spec.Containers[0].Args = []string{"--token=t0ps3cret"}
 	p.Annotations = map[string]string{lastAppliedAnnotation: secretish}
 
-	watcher := NewPodWatcher(fake.NewClientset(p), func(PodInfo) {})
+	watcher := NewPodWatcher(fake.NewClientset(p), func(model.PodInfo) {})
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	go func() { _ = watcher.Run(ctx) }()

internal/collector/cadvisor_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/RebuildStackCo/runtime-agent/internal/model"
 	"github.com/RebuildStackCo/runtime-agent/internal/rollup"
 )
 
@@ -115,7 +116,7 @@ func TestThrottlingDeltaRules(t *testing.T) {
 // own start, exactly as any first observation is.
 func TestThrottlingDoesNotCrossPodRecreation(t *testing.T) {
 	resolver := stubResolver{
-		"uid-old": {namespace: "shop", name: "web-abc", workload: WorkloadRef{Kind: "StatefulSet", Name: "web"}},
+		"uid-old": {namespace: "shop", name: "web-abc", workload: model.WorkloadRef{Kind: "StatefulSet", Name: "web"}},
 	}
 	p := testPoller(resolver)
 	first := usageTestStart.Add(30 * time.Second)
@@ -139,7 +140,7 @@ func TestThrottlingDoesNotCrossPodRecreation(t *testing.T) {
 	delete(resolver, "uid-old")
 	resolver["uid-new"] = podIndexEntry{
 		namespace: "shop", name: "web-abc",
-		workload: WorkloadRef{Kind: "StatefulSet", Name: "web"},
+		workload: model.WorkloadRef{Kind: "StatefulSet", Name: "web"},
 	}
 	ingest(3, 10, second, newPodStart, second)
 

internal/collector/filter.go

@@ -4,6 +4,7 @@ import (
 	"strings"
 	"sync/atomic"
 
+	"github.com/RebuildStackCo/runtime-agent/internal/model"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 )
@@ -203,36 +204,9 @@ func (f *Filter) countUnresolvedWorkload(reason UnresolvedWorkload) {
 	}
 }
 
-// Coverage is an aggregate snapshot of what the filter admitted and
-// excluded, counted once per pod appearance. It is the seed of the coverage
-// report (docs/security.md §11): full information about what is collected,
-// only counts about what is not.
-type Coverage struct {
-	PodsObserved                int64 `json:"pods_observed"`
-	ExcludedNamespaceFilter     int64 `json:"excluded_namespace_filter"`
-	ExcludedNamespaceAnnotation int64 `json:"excluded_namespace_annotation"`
-	ExcludedWorkloadAnnotation  int64 `json:"excluded_workload_annotation"`
-	ExcludedPodAnnotation       int64 `json:"excluded_pod_annotation"`
-	// WorkloadUnknownKind and WorkloadNotCached are the blind spot: pods
-	// admitted without their workload-level opt-out being checked. They are
-	// kept apart because they mean different things — the first is a standing
-	// property of a cluster running operators the agent does not read, the
-	// second should sit at zero.
-	WorkloadUnknownKind int64 `json:"workload_unknown_kind"`
-	WorkloadNotCached   int64 `json:"workload_not_cached"`
-
-	// Job counters, kept apart from the pod ones on purpose: one Job produces
-	// many pods, so a shared counter would answer neither question.
-	JobsObserved                    int64 `json:"jobs_observed"`
-	JobsExcludedNamespaceFilter     int64 `json:"jobs_excluded_namespace_filter"`
-	JobsExcludedNamespaceAnnotation int64 `json:"jobs_excluded_namespace_annotation"`
-	JobsExcludedWorkloadAnnotation  int64 `json:"jobs_excluded_workload_annotation"`
-	JobsExcludedAnnotation          int64 `json:"jobs_excluded_annotation"`
-}
-
 // Snapshot returns the current coverage counters.
-func (f *Filter) Snapshot() Coverage {
-	return Coverage{
+func (f *Filter) Snapshot() model.Coverage {
+	return model.Coverage{
 		PodsObserved:                f.podsObserved.Load(),
 		ExcludedNamespaceFilter:     f.excludedNamespaceFilter.Load(),
 		ExcludedNamespaceAnnotation: f.excludedNamespaceAnnotation.Load(),

internal/collector/filter_test.go

@@ -6,6 +6,7 @@ import (
 	"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"
@@ -110,16 +111,16 @@ func TestWatcherGatesPodsAndCountsCoverage(t *testing.T) {
 	clientset := fake.NewClientset(optOutNs, deniedPod, optedOutPod, annotatedPod, collected)
 
 	var mu sync.Mutex
-	seen := make(map[string]PodInfo)
-	watcher := NewPodWatcher(clientset, func(p PodInfo) {
+	seen := make(map[string]model.PodInfo)
+	watcher := NewPodWatcher(clientset, func(p model.PodInfo) {
 		mu.Lock()
 		seen[p.Namespace+"/"+p.Name] = p
 		mu.Unlock()
 	})
 	filter := NewFilter(nil, []string{"pr-*"})
 	watcher.SetFilter(filter)
-	oomEvents := make(chan OOMKill, 16)
-	watcher.OnOOMKill(func(o OOMKill) { oomEvents <- o })
+	oomEvents := make(chan model.OOMKill, 16)
+	watcher.OnOOMKill(func(o model.OOMKill) { oomEvents <- o })
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -149,7 +150,7 @@ func TestWatcherGatesPodsAndCountsCoverage(t *testing.T) {
 	}
 	mu.Unlock()
 
-	want := Coverage{
+	want := model.Coverage{
 		PodsObserved:                1,
 		ExcludedNamespaceFilter:     1,
 		ExcludedNamespaceAnnotation: 1,
@@ -243,7 +244,7 @@ func TestWorkloadAnnotationExcludesWithoutTouchingPodTemplates(t *testing.T) {
 
 	var mu sync.Mutex
 	seen := map[string]bool{}
-	watcher := NewPodWatcher(clientset, func(p PodInfo) {
+	watcher := NewPodWatcher(clientset, func(p model.PodInfo) {
 		mu.Lock()
 		seen[p.Name] = true
 		mu.Unlock()