diff --git a/integration/k3sic/k3sic.go b/integration/k3sic/k3sic.go
new file mode 100644
index 00000000..6dc80563
--- /dev/null
+++ b/integration/k3sic/k3sic.go
@@ -0,0 +1,603 @@
+// Package k3sic wraps a single-container k3s cluster (server + agent) as a
+// privileged sibling container on the host docker daemon, like the
+// DERP-in-container wrapper in the dsic package.
+//
+// It exists so that integration tests can install the real Tailscale
+// Kubernetes operator (via its Helm chart) into a real Kubernetes cluster and
+// point it at an in-test Headscale. k3s bundles kubectl and we run helm inside
+// the container through Execute, so the host dev shell needs no kube tooling.
+//
+// The operator is pointed at Headscale over plain HTTP (see
+// hsic.WithoutTLS), so the operator and proxy pods need no CA: there is no
+// image baking and no CoreDNS hostname mapping. To run instead against a TLS
+// Headscale with a private CA, see tls-ca-baking.md.
+//
+// The harness runs as sibling containers on the host docker daemon (the
+// test-suite container has the host docker socket bind-mounted); it is NOT
+// docker-in-docker. We therefore run the purpose-built single-container k3s image
+// as one more privileged sibling joined to the scenario networks, rather than using
+// k3d/kind which shell out to the docker daemon and fight the sibling model.
+// Privileged is the harness-wide norm here, not a k3s-specific escalation: the
+// tsic (client) and dsic (DERP) containers run privileged too (see
+// dockertestutil.DockerAllowNetworkAdministration).
+package k3sic
+
+import (
+	"archive/tar"
+	"compress/gzip"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/juanfont/headscale/hscontrol/capver"
+	"github.com/juanfont/headscale/integration/dockertestutil"
+	"github.com/juanfont/headscale/integration/integrationutil"
+	"github.com/ory/dockertest/v3"
+	"github.com/ory/dockertest/v3/docker"
+	"tailscale.com/util/rands"
+)
+
+const (
+	k3sicHashLength = 6
+
+	// K3sImage is the single-container k3s server+agent image, pinned on ghcr (no
+	// anonymous Docker Hub rate limit). Pinned rather than resolved from the k3s
+	// stable channel because that channel floats the k8s minor unpredictably and
+	// would outrun the cgroup-v2 and br_netfilter workarounds below. Bump by hand.
+	K3sImage = "ghcr.io/k3s-io/k3s:v1.35.5-k3s1"
+
+	// operatorImageRepo and proxyImageRepo are the ghcr-hosted operator and
+	// proxy images. ghcr is used instead of Docker Hub to avoid anonymous pull
+	// rate limits; the pods pull these directly.
+	operatorImageRepo = "ghcr.io/tailscale/k8s-operator"
+	proxyImageRepo    = "ghcr.io/tailscale/tailscale"
+
+	dockerExecuteTimeout = 300 * time.Second
+
+	// helmVersionFallback is used when the latest helm release cannot be
+	// resolved at runtime (see resolveHelmVersion). The image ships no helm and
+	// cannot fetch it itself, so we inject a binary that matches the container
+	// arch.
+	helmVersionFallback = "v3.19.1"
+
+	// kubeconfigPath is where k3s writes the kubeconfig (see RunOptions.Env);
+	// helm needs it pointed explicitly, kubectl finds it by default.
+	kubeconfigPath = "/etc/rancher/k3s/k3s.yaml"
+
+	// kubectlBin is the in-container kubectl the k3s image ships on PATH.
+	kubectlBin = "kubectl"
+
+	// shellBin is the in-container shell used for compound commands.
+	shellBin = "/bin/sh"
+
+	// tailscaleNamespace is where the operator and its proxies are installed.
+	tailscaleNamespace = "tailscale"
+
+	// kubeSystemNamespace holds CoreDNS and the rest of the k3s system addons.
+	kubeSystemNamespace = "kube-system"
+)
+
+var (
+	errHelmDownload     = errors.New("helm download failed")
+	errHelmNotInTarball = errors.New("helm binary not found in release tarball")
+
+	errNoKubeDNSEndpoints = errors.New("kube-dns Service has no ready endpoints yet")
+)
+
+// OperatorImageTag is the image tag the operator and proxy images use, derived
+// from the Tailscale minor Headscale tracks via capver (e.g. "v1.98"). The
+// operator shares the Tailscale release train, so this keeps the images and the
+// Helm chart in lockstep with the client versions Headscale is tested against,
+// without a hand-pinned constant. The tag is a rolling tag within the minor.
+func OperatorImageTag() string {
+	return capver.TailscaleLatestMajorMinor(1, false)[0]
+}
+
+// OperatorImage and ProxyImage are the ghcr operator/proxy image references the
+// test wires into the Helm chart.
+func OperatorImage() string { return operatorImageRepo + ":" + OperatorImageTag() }
+func ProxyImage() string    { return proxyImageRepo + ":" + OperatorImageTag() }
+
+// OperatorChartVersion is the Helm chart version constraint matching the derived
+// minor; helm resolves the latest patch in that line. The top-level loginServer
+// value the operator needs to target Headscale instead of the Tailscale SaaS
+// first shipped in chart 1.98.4.
+func OperatorChartVersion() string {
+	return strings.TrimPrefix(OperatorImageTag(), "v") + ".*"
+}
+
+// K3sInContainer represents a k3s cluster running in a single privileged
+// container (K3sInContainer, hence k3sic).
+type K3sInContainer struct {
+	hostname string
+
+	pool      *dockertest.Pool
+	container *dockertest.Resource
+	networks  []*dockertest.Network
+}
+
+// New starts a new [K3sInContainer] joined to the given networks.
+func New(
+	pool *dockertest.Pool,
+	networks []*dockertest.Network,
+) (*K3sInContainer, error) {
+	hash := rands.HexString(k3sicHashLength)
+
+	// Include the run ID in the hostname for easier identification of which
+	// test run owns this container, matching the dsic/tsic convention.
+	runID := dockertestutil.GetIntegrationRunID()
+
+	var hostname string
+
+	if runID != "" {
+		runIDShort := runID[len(runID)-6:]
+		hostname = fmt.Sprintf("k3s-%s-%s", runIDShort, hash)
+	} else {
+		hostname = "k3s-" + hash
+	}
+
+	k := &K3sInContainer{
+		hostname: hostname,
+		pool:     pool,
+		networks: networks,
+	}
+
+	// Pull the k3s image (ghcr, not built) via PullWithAuth, as hsic does for
+	// prebuilt/pulled images.
+	err := dockertestutil.PullWithAuth(pool, K3sImage)
+	if err != nil {
+		return nil, fmt.Errorf("pulling %s: %w", K3sImage, err)
+	}
+
+	repo, tag, ok := strings.Cut(K3sImage, ":")
+	if !ok {
+		return nil, fmt.Errorf("invalid k3s image reference %q", K3sImage) //nolint:err113
+	}
+
+	runOptions := &dockertest.RunOptions{
+		Name:       hostname,
+		Repository: repo,
+		Tag:        tag,
+		Networks:   networks,
+		// "server" runs both the control plane and a built-in agent in one
+		// container. --disable traefik/servicelb/metrics-server keeps the
+		// cluster lean: the test only needs the API server and the ability to
+		// schedule the operator pods. --tls-san pins the hostname into the
+		// apiserver cert (not strictly needed since we exec kubectl in-container,
+		// but harmless and future-proof).
+		Cmd: []string{
+			"server",
+			"--disable", "traefik",
+			"--disable", "servicelb",
+			"--disable", "metrics-server",
+			"--disable-network-policy",
+			"--snapshotter", "native",
+			"--tls-san", hostname,
+		},
+		Env: []string{
+			"K3S_KUBECONFIG_OUTPUT=" + kubeconfigPath,
+			"K3S_KUBECONFIG_MODE=0644",
+		},
+	}
+
+	// Stamp the run-id label or the reaper leaks the container.
+	dockertestutil.DockerAddIntegrationLabels(runOptions, "k3s")
+
+	// dockertest does not handle pre-existing containers well; make sure a
+	// stale one with this name is gone first.
+	err = pool.RemoveContainerByName(hostname)
+	if err != nil {
+		return nil, err
+	}
+
+	container, err := pool.RunWithOptions(
+		runOptions,
+		dockertestutil.DockerRestartPolicy,
+		// Privileged + NET_ADMIN: k3s manages iptables/ipvs, mounts cgroups and
+		// runs containerd. This is the same knob dsic uses for the DERP server.
+		dockertestutil.DockerAllowNetworkAdministration,
+		withK3sHostConfig,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("%s starting k3s container: %w", hostname, err)
+	}
+
+	log.Printf("Created %s container\n", hostname)
+
+	k.container = container
+
+	// Make ClusterIP DNAT work before k3s programs kube-proxy rules; without it
+	// in-cluster DNS times out on hosts where br_netfilter is not preloaded.
+	k.ensureBridgeNetfilter()
+
+	return k, nil
+}
+
+// withK3sHostConfig sets the HostConfig knobs k3s needs beyond
+// privileged/NET_ADMIN: a tmpfs on /run and /var/run, which the k3s image
+// expects.
+//
+// It deliberately does NOT bind-mount the host /sys/fs/cgroup. On a cgroup-v2
+// host, bind-mounting the host cgroup tree into a container that keeps its own
+// (private) cgroup namespace makes the cgroup root visible to the container
+// disagree with the namespace runc places workload pods under. The kubelet can
+// then start (the apiserver and node go Ready), but every workload pod fails to
+// create its sandbox with "failed to apply cgroup configuration: ...
+// cgroup.procs: no such file or directory", so nothing the operator schedules
+// ever runs. Leaving the bind-mount off lets the privileged k3s entrypoint set
+// up cgroup-v2 delegation within its own namespace, which it is
+// designed to do, and workload pods schedule normally.
+func withK3sHostConfig(config *docker.HostConfig) {
+	config.Tmpfs = map[string]string{
+		"/run":     "",
+		"/var/run": "",
+	}
+
+	// Bind the host kernel modules read-only so the container can load
+	// br_netfilter (see ensureBridgeNetfilter). Without bridge netfilter,
+	// kube-proxy's ClusterIP DNAT rules do not apply to bridged pod-to-pod
+	// traffic, so kube-dns (and every other Service) is unreachable from pods —
+	// the in-cluster DNS timeout seen on the arm64 CI runner. The container
+	// shares the host kernel, so the modules match.
+	config.Binds = append(config.Binds, "/lib/modules:/lib/modules:ro")
+}
+
+// ensureBridgeNetfilter loads br_netfilter and enables the sysctls that make
+// kube-proxy's ClusterIP DNAT apply to bridged pod-to-pod traffic. On a host
+// where the module is already loaded (e.g. the amd64 dev box, where Docker
+// loads it for bridge networks) these are no-ops; on the arm64 CI runner the
+// module is absent and pods cannot reach any Service IP — kube-dns included —
+// so in-cluster DNS times out. Best-effort: k3s also loads the module, and a
+// genuinely missing module surfaces in DumpDiagnostics rather than here.
+func (k *K3sInContainer) ensureBridgeNetfilter() {
+	for _, cmd := range []string{
+		"modprobe br_netfilter || true",
+		"sysctl -w net.bridge.bridge-nf-call-iptables=1 || true",
+		"sysctl -w net.bridge.bridge-nf-call-ip6tables=1 || true",
+		"sysctl -w net.ipv4.ip_forward=1 || true",
+	} {
+		out, stderr, err := k.Execute([]string{shellBin, "-c", cmd})
+		if err != nil {
+			log.Printf("[k3s] %q failed: %v (stdout: %s, stderr: %s)", cmd, err, out, stderr)
+		}
+	}
+}
+
+// Hostname returns the hostname of the [K3sInContainer].
+func (k *K3sInContainer) Hostname() string {
+	return k.hostname
+}
+
+// ID returns the docker container ID of the [K3sInContainer].
+func (k *K3sInContainer) ID() string {
+	return k.container.Container.ID
+}
+
+// ConnectToNetwork connects the cluster container to an additional network.
+func (k *K3sInContainer) ConnectToNetwork(network *dockertest.Network) error {
+	return k.container.ConnectToNetwork(network)
+}
+
+// Execute runs a command inside the k3s container and returns its stdout.
+// kubectl and the k3s-bundled tools (and helm, once installed via
+// [K3sInContainer.InstallHelm]) are on PATH. KUBECONFIG is exported so helm,
+// which (unlike the image's kubectl) does not default to the k3s config, can
+// reach the cluster.
+func (k *K3sInContainer) Execute(command []string) (string, string, error) {
+	return dockertestutil.ExecuteCommand(
+		k.container,
+		command,
+		[]string{"KUBECONFIG=" + kubeconfigPath},
+		dockertestutil.ExecuteCommandTimeout(dockerExecuteTimeout),
+	)
+}
+
+// WriteFile saves a file inside the container.
+func (k *K3sInContainer) WriteFile(path string, data []byte) error {
+	return integrationutil.WriteFileToContainer(k.pool, k.container, path, data)
+}
+
+// WaitForRunning blocks until the cluster is ready to schedule DNS-dependent
+// workloads: the kube-apiserver is serving, the single node reports Ready, and
+// in-cluster DNS is servable (CoreDNS rolled out with a backed kube-dns
+// Service). Gating on DNS here pins a missing-DNS failure at its source rather
+// than letting the operator crashloop on an opaque lookup timeout.
+func (k *K3sInContainer) WaitForRunning() error {
+	log.Printf("waiting for k3s API server in %s to be ready", k.hostname)
+
+	err := k.pool.Retry(func() error {
+		// `kubectl get --raw=/readyz` returns "ok" once the apiserver is up.
+		out, _, err := k.Execute([]string{
+			kubectlBin, "get", "--raw=/readyz",
+		})
+		if err != nil {
+			return fmt.Errorf("k3s apiserver not ready: %w", err)
+		}
+
+		if !strings.Contains(out, "ok") {
+			return fmt.Errorf("k3s apiserver readyz returned %q", strings.TrimSpace(out)) //nolint:err113
+		}
+
+		// Wait for the node object to exist and be Ready before returning so
+		// pods can actually be scheduled.
+		nodeOut, _, err := k.Execute([]string{
+			kubectlBin, "get", "nodes", "--no-headers",
+		})
+		if err != nil {
+			return fmt.Errorf("k3s node not ready: %w", err)
+		}
+
+		if !strings.Contains(nodeOut, " Ready") {
+			return fmt.Errorf("k3s node not Ready yet: %q", strings.TrimSpace(nodeOut)) //nolint:err113
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	return k.waitForClusterDNS()
+}
+
+// InstallHelm installs the helm binary into the container so the operator can be
+// installed. The k3s image ships kubectl but not helm, has no curl, and its
+// busybox wget cannot do HTTPS, so we download helm in the test process (which
+// has network egress) and inject the binary. helm's own HTTPS client then
+// fetches the operator chart from inside the container.
+func (k *K3sInContainer) InstallHelm() error {
+	bin, err := fetchHelmBinary(resolveHelmVersion(), runtime.GOARCH)
+	if err != nil {
+		return fmt.Errorf("fetching helm: %w", err)
+	}
+
+	err = k.WriteFile("/usr/local/bin/helm", bin)
+	if err != nil {
+		return fmt.Errorf("writing helm binary: %w", err)
+	}
+
+	// WriteFile uploads with mode 0; make it readable+executable.
+	_, stderr, err := k.Execute([]string{"chmod", "0755", "/usr/local/bin/helm"})
+	if err != nil {
+		return fmt.Errorf("chmod helm (stderr: %s): %w", stderr, err)
+	}
+
+	return nil
+}
+
+// waitForClusterDNS blocks until CoreDNS is rolled out and the kube-dns Service
+// has at least one ready endpoint — i.e. in-cluster name resolution is actually
+// servable, which every workload (starting with the operator) depends on. k3s
+// deploys CoreDNS via its addon manager shortly after the node reports Ready, so
+// the deployment may not exist yet; retry until it does before checking rollout.
+func (k *K3sInContainer) waitForClusterDNS() error {
+	err := k.pool.Retry(func() error {
+		_, stderr, err := k.Execute([]string{
+			kubectlBin, "-n", kubeSystemNamespace, "get", "deployment", "coredns",
+		})
+		if err != nil {
+			return fmt.Errorf("coredns deployment not present yet (stderr: %s): %w", stderr, err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("waiting for coredns deployment to appear: %w", err)
+	}
+
+	_, stderr, err := k.Execute([]string{
+		kubectlBin, "-n", kubeSystemNamespace, "rollout", "status",
+		"deployment/coredns", "--timeout=150s",
+	})
+	if err != nil {
+		return fmt.Errorf("coredns did not become available (stderr: %s): %w", stderr, err)
+	}
+
+	return k.pool.Retry(func() error {
+		// A populated endpoint set means a CoreDNS pod is serving :53 and
+		// kube-proxy has a backend to DNAT the kube-dns ClusterIP to; empty means
+		// in-cluster lookups will time out no matter how long a client waits.
+		out, stderr, err := k.Execute([]string{
+			kubectlBin, "-n", kubeSystemNamespace, "get", "endpoints", "kube-dns",
+			"-o", "jsonpath={.subsets[*].addresses[*].ip}",
+		})
+		if err != nil {
+			return fmt.Errorf("reading kube-dns endpoints (stderr: %s): %w", stderr, err)
+		}
+
+		if strings.TrimSpace(out) == "" {
+			return errNoKubeDNSEndpoints
+		}
+
+		return nil
+	})
+}
+
+// ConfigureCoreDNSHost makes in-cluster pods resolve hostname to ip via CoreDNS.
+// The operator targets Headscale's control plane by IP, but the embedded DERP map
+// references Headscale by hostname; without this, the proxy pods cannot resolve
+// the DERP server, never connect to it, and — since they only advertise
+// unreachable pod-network endpoints — get no data path to nodes outside the
+// cluster. It installs a coredns-custom ConfigMap (a k3s-native extension point:
+// keys ending in .server become additional server blocks), which CoreDNS's reload
+// plugin picks up without a restart.
+func (k *K3sInContainer) ConfigureCoreDNSHost(hostname, ip string) error {
+	manifest := fmt.Sprintf(`apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: coredns-custom
+  namespace: kube-system
+data:
+  headscale.server: |
+    %s {
+      hosts {
+        %s %s
+        fallthrough
+      }
+    }
+`, hostname, ip, hostname)
+
+	return k.ApplyManifest("coredns-custom", manifest)
+}
+
+// DumpDiagnostics logs cluster state useful for debugging a failed operator
+// install: pod status across namespaces and the operator's own logs and events.
+// Best-effort — every command's failure is logged, not returned.
+func (k *K3sInContainer) DumpDiagnostics() {
+	for _, c := range [][]string{
+		{kubectlBin, "get", "pods", "-A", "-o", "wide"},
+		{kubectlBin, "-n", tailscaleNamespace, "get", "events", "--sort-by=.lastTimestamp"},
+		{kubectlBin, "-n", tailscaleNamespace, "describe", "pods"},
+		{kubectlBin, "-n", tailscaleNamespace, "logs", "deployment/operator", "--tail=200"},
+		// A crashlooping operator's fatal error is in the previous container.
+		{kubectlBin, "-n", tailscaleNamespace, "logs", "deployment/operator", "--previous", "--tail=200"},
+		{kubectlBin, "-n", tailscaleNamespace, "get", "statefulsets,pods", "-o", "wide"},
+		// Proxy pods' tailscaled logs: DERP-connection and registration failures
+		// (the data-path culprits) surface here, not in the operator log.
+		{
+			shellBin, "-c",
+			"for p in $(kubectl -n " + tailscaleNamespace + " get pods -o name | grep /ts-); do " +
+				"echo \"== $p ==\"; kubectl -n " + tailscaleNamespace +
+				" logs $p -c tailscale --tail=80 2>&1; done",
+		},
+		// In-cluster DNS: the operator's first dependency. A "lookup
+		// kubernetes.default.svc ... i/o timeout" crash means CoreDNS is not
+		// serving, so capture its pod state, logs (Corefile parse errors land
+		// here), and the Service endpoints.
+		{kubectlBin, "-n", kubeSystemNamespace, "get", "pods", "-l", "k8s-app=kube-dns", "-o", "wide"},
+		{kubectlBin, "-n", kubeSystemNamespace, "logs", "-l", "k8s-app=kube-dns", "--tail=100"},
+		{kubectlBin, "-n", kubeSystemNamespace, "get", "endpoints", "kube-dns", "-o", "wide"},
+		// Host network state behind a ClusterIP-unreachable DNS timeout: whether
+		// br_netfilter is loaded and the call-iptables/forward sysctls are on, and
+		// whether kube-proxy actually programmed the kube-dns DNAT rule. If CoreDNS
+		// is healthy (above) but these are missing, the fault is the Service DNAT
+		// path, not DNS.
+		{shellBin, "-c", "lsmod | grep -E 'br_netfilter|nf_conntrack' || echo 'br_netfilter NOT loaded'"},
+		{shellBin, "-c", "sysctl net.bridge.bridge-nf-call-iptables net.ipv4.ip_forward 2>&1 || true"},
+		{shellBin, "-c", "iptables-save -t nat 2>/dev/null | grep -iE 'KUBE-SERVICES|kube-dns|10.43.0.10' | head -40 || echo 'no kube-dns nat rules'"},
+	} {
+		out, stderr, err := k.Execute(c)
+		label := strings.Join(c, " ")
+
+		if err != nil {
+			log.Printf("[k3s diag] %s failed: %v (stderr: %s)", label, err, stderr)
+			continue
+		}
+
+		log.Printf("[k3s diag] %s:\n%s", label, out)
+	}
+}
+
+// resolveHelmVersion returns the latest published helm release tag (e.g.
+// "v3.19.1"), falling back to [helmVersionFallback] if it cannot be resolved.
+// get.helm.sh is not Docker Hub and has no anonymous rate limit, so a "rolling"
+// latest is cheap; the fallback keeps a broken release from breaking CI.
+func resolveHelmVersion() string {
+	req, err := http.NewRequestWithContext(
+		context.Background(), http.MethodGet, "https://get.helm.sh/helm-latest-version", nil)
+	if err != nil {
+		return helmVersionFallback
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return helmVersionFallback
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return helmVersionFallback
+	}
+
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 32))
+	if err != nil {
+		return helmVersionFallback
+	}
+
+	version := strings.TrimSpace(string(body))
+	if !strings.HasPrefix(version, "v") {
+		return helmVersionFallback
+	}
+
+	return version
+}
+
+// fetchHelmBinary downloads the helm release tarball for version and goarch and
+// returns the helm binary bytes.
+func fetchHelmBinary(version, goarch string) ([]byte, error) {
+	url := fmt.Sprintf("https://get.helm.sh/helm-%s-linux-%s.tar.gz", version, goarch)
+
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, url, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("%w: %s returned status %d", errHelmDownload, url, resp.StatusCode)
+	}
+
+	gz, err := gzip.NewReader(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	defer gz.Close()
+
+	want := "linux-" + goarch + "/helm"
+	tr := tar.NewReader(gz)
+
+	for {
+		hdr, err := tr.Next()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		if hdr.Name == want {
+			return io.ReadAll(tr)
+		}
+	}
+
+	return nil, fmt.Errorf("%w: %s", errHelmNotInTarball, want)
+}
+
+// Shutdown saves the container log and then runs k3s-killall in-container so
+// k3s's own child processes/containers (containerd-shims, pods) do not leak,
+// before purging the container itself.
+func (k *K3sInContainer) Shutdown() error {
+	err := k.SaveLog("/tmp/control")
+	if err != nil {
+		log.Printf("saving log from %s: %s", k.hostname, err)
+	}
+
+	// k3s spawns containerd and a tree of child processes inside this
+	// container; the bundled k3s-killall.sh tears them down. Best-effort: the
+	// Purge below removes the container regardless.
+	_, _, err = k.Execute([]string{shellBin, "-c", "k3s-killall.sh || true"})
+	if err != nil {
+		log.Printf("running k3s-killall in %s: %s", k.hostname, err)
+	}
+
+	return k.pool.Purge(k.container)
+}
+
+// SaveLog saves the container stdout/stderr logs to a path on the host.
+func (k *K3sInContainer) SaveLog(path string) error {
+	_, _, err := dockertestutil.SaveLog(k.pool, k.container, path)
+
+	return err
+}
diff --git a/integration/k3sic/operator.go b/integration/k3sic/operator.go
new file mode 100644
index 00000000..e9eee170
--- /dev/null
+++ b/integration/k3sic/operator.go
@@ -0,0 +1,243 @@
+package k3sic
+
+import (
+	"fmt"
+	"strings"
+)
+
+// This file holds the reusable building blocks for driving the Tailscale
+// Kubernetes operator inside the cluster: installing it and applying the CRs and
+// workloads a test needs. Tests compose these methods rather than embedding
+// kubectl/helm invocations, so adding a new operator test is a few method calls.
+
+// ApplyManifest writes manifest into the container as /tmp/<name>.yaml and
+// kubectl-applies it. It is the building block the helpers below use, and is
+// exported so tests can apply ad-hoc manifests without a bespoke method.
+func (k *K3sInContainer) ApplyManifest(name, manifest string) error {
+	path := "/tmp/" + name + ".yaml"
+
+	err := k.WriteFile(path, []byte(manifest))
+	if err != nil {
+		return fmt.Errorf("writing manifest %s: %w", name, err)
+	}
+
+	_, stderr, err := k.Execute([]string{kubectlBin, "apply", "-f", path})
+	if err != nil {
+		return fmt.Errorf("applying manifest %s (stderr: %s): %w", name, stderr, err)
+	}
+
+	return nil
+}
+
+// InstallOperator installs the Tailscale Kubernetes operator via Helm into the
+// tailscale namespace, pointed at loginServer with the given OAuth client
+// credentials. loginServer is used by the operator for both the control plane
+// and the management API; for an in-test Headscale pass its HTTP endpoint by IP
+// (hsic.HeadscaleInContainer.GetIPEndpoint) so the pods need no DNS or CA. The
+// operator and proxy images come from ghcr at the capver-derived tag. Blocks
+// (helm --wait) until the operator deployment is available.
+func (k *K3sInContainer) InstallOperator(loginServer, clientID, clientSecret string) error {
+	repoAdd := "helm repo add tailscale https://pkgs.tailscale.com/helmcharts && helm repo update"
+
+	_, stderr, err := k.Execute([]string{shellBin, "-c", repoAdd})
+	if err != nil {
+		return fmt.Errorf("helm repo add/update (stderr: %s): %w", stderr, err)
+	}
+
+	_, stderr, err = k.Execute([]string{
+		kubectlBin, "create", "namespace", tailscaleNamespace,
+	})
+	if err != nil {
+		return fmt.Errorf("creating %s namespace (stderr: %s): %w", tailscaleNamespace, stderr, err)
+	}
+
+	// Precreate the operator-oauth Secret with --from-literal instead of the
+	// chart's oauth.clientId/clientSecret: the chart interpolates those unquoted,
+	// so an all-digit credential renders as a YAML number and the apiserver
+	// rejects it. --from-literal always stores strings. The chart uses a Secret
+	// named operator-oauth when oauth.clientId is unset.
+	_, stderr, err = k.Execute([]string{
+		kubectlBin, "-n", tailscaleNamespace, "create", "secret", "generic", "operator-oauth",
+		"--from-literal=client_id=" + clientID,
+		"--from-literal=client_secret=" + clientSecret,
+	})
+	if err != nil {
+		return fmt.Errorf("creating operator-oauth secret (stderr: %s): %w", stderr, err)
+	}
+
+	opRepo, opTag, _ := strings.Cut(OperatorImage(), ":")
+	proxyRepo, proxyTag, _ := strings.Cut(ProxyImage(), ":")
+
+	const set = "--set-string"
+
+	install := []string{
+		"helm", "upgrade", "--install", "tailscale-operator",
+		"tailscale/tailscale-operator",
+		"--version", OperatorChartVersion(),
+		"--namespace", tailscaleNamespace,
+		set, "loginServer=" + loginServer,
+		set, "operatorConfig.image.repository=" + opRepo,
+		set, "operatorConfig.image.tag=" + opTag,
+		set, "proxyConfig.image.repository=" + proxyRepo,
+		set, "proxyConfig.image.tag=" + proxyTag,
+		"--wait", "--timeout", "5m",
+	}
+
+	_, stderr, err = k.Execute(install)
+	if err != nil {
+		k.DumpDiagnostics()
+		return fmt.Errorf("helm install operator (stderr: %s): %w", stderr, err)
+	}
+
+	// hsic serves the embedded DERP without TLS, but a proxy dials DERP over HTTPS
+	// and so cannot relay through it. Pods on the k3s pod network can only reach
+	// off-cluster nodes via DERP (their only endpoint is an unreachable pod IP),
+	// so without this the ingress/egress proxies get no data path. The ProxyClass
+	// injects TS_DEBUG_DERP_WS_CLIENT + TS_DEBUG_USE_DERP_HTTP, switching proxies to
+	// plain-HTTP websocket DERP. The proxy-creating helpers below reference it.
+	return k.applyDERPWebsocketProxyClass()
+}
+
+// DERPWebsocketProxyClass is the ProxyClass [InstallOperator] creates to make
+// operator proxies reach the embedded (non-TLS) DERP over websocket. Proxy
+// resources reference it via spec.proxyClass / the tailscale.com/proxy-class
+// annotation.
+const DERPWebsocketProxyClass = "headscale-derp-ws" //nolint:gosec // G101 false positive: a ProxyClass name, not a credential
+
+func (k *K3sInContainer) applyDERPWebsocketProxyClass() error {
+	manifest := fmt.Sprintf(`apiVersion: tailscale.com/v1alpha1
+kind: ProxyClass
+metadata:
+  name: %s
+spec:
+  statefulSet:
+    pod:
+      tailscaleContainer:
+        env:
+          - name: TS_DEBUG_DERP_WS_CLIENT
+            value: "true"
+          - name: TS_DEBUG_USE_DERP_HTTP
+            value: "true"
+`, DERPWebsocketProxyClass)
+
+	return k.ApplyManifest("proxyclass-"+DERPWebsocketProxyClass, manifest)
+}
+
+// DeployConnector applies a Connector CR advertising an egress subnet router for
+// advertiseRoutes, tagged with tags. The operator provisions a proxy and
+// registers it as a node in Headscale.
+func (k *K3sInContainer) DeployConnector(name string, tags, advertiseRoutes []string) error {
+	manifest := fmt.Sprintf(`apiVersion: tailscale.com/v1alpha1
+kind: Connector
+metadata:
+  name: %s
+spec:
+  proxyClass: %s
+  tags:
+%s
+  subnetRouter:
+    advertiseRoutes:
+%s
+`, name, DERPWebsocketProxyClass, yamlList(tags, 4), yamlList(advertiseRoutes, 6))
+
+	return k.ApplyManifest("connector-"+name, manifest)
+}
+
+// DeployProxyGroup applies a ProxyGroup CR of the given type ("ingress" or
+// "egress") with replicas proxies tagged with tags. ProxyGroups are the current
+// way to run a pool of operator proxies for HA ingress/egress.
+func (k *K3sInContainer) DeployProxyGroup(name, proxyType string, replicas int, tags []string) error {
+	manifest := fmt.Sprintf(`apiVersion: tailscale.com/v1alpha1
+kind: ProxyGroup
+metadata:
+  name: %s
+spec:
+  type: %s
+  replicas: %d
+  proxyClass: %s
+  tags:
+%s
+`, name, proxyType, replicas, DERPWebsocketProxyClass, yamlList(tags, 4))
+
+	return k.ApplyManifest("proxygroup-"+name, manifest)
+}
+
+// DeployEchoServer deploys a minimal HTTP server (agnhost, served from
+// registry.k8s.io to avoid Docker Hub rate limits) labelled app=<name> with a
+// ClusterIP Service of the same name on port 80. Use it as the in-cluster target
+// for connectivity tests; expose it to the tailnet with [ExposeServiceToTailnet].
+func (k *K3sInContainer) DeployEchoServer(name string) error {
+	manifest := fmt.Sprintf(`apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: %s
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: %s
+  template:
+    metadata:
+      labels:
+        app: %s
+    spec:
+      containers:
+        - name: echo
+          image: registry.k8s.io/e2e-test-images/agnhost:2.47
+          args: ["netexec", "--http-port=80"]
+          ports:
+            - containerPort: 80
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: %s
+spec:
+  selector:
+    app: %s
+  ports:
+    - port: 80
+      targetPort: 80
+`, name, name, name, name, name)
+
+	return k.ApplyManifest("echo-"+name, manifest)
+}
+
+// ExposeServiceToTailnet creates a tailscale LoadBalancer Service named
+// "<name>-ts" that exposes the pods labelled app=<name> to the tailnet, tagged
+// with tags. The operator provisions an ingress proxy and registers a node, so a
+// node outside the cluster (a regular tsic client) can reach the service over
+// the tailnet.
+func (k *K3sInContainer) ExposeServiceToTailnet(name string, tags []string) error {
+	manifest := fmt.Sprintf(`apiVersion: v1
+kind: Service
+metadata:
+  name: %s-ts
+  annotations:
+    tailscale.com/tags: "%s"
+    tailscale.com/proxy-class: %s
+spec:
+  type: LoadBalancer
+  loadBalancerClass: tailscale
+  selector:
+    app: %s
+  ports:
+    - port: 80
+      targetPort: 80
+`, name, strings.Join(tags, ","), DERPWebsocketProxyClass, name)
+
+	return k.ApplyManifest("expose-"+name, manifest)
+}
+
+// yamlList renders items as a YAML block sequence indented by indent spaces,
+// e.g. "    - tag:k8s". Returns "" for an empty list.
+func yamlList(items []string, indent int) string {
+	var b strings.Builder
+
+	pad := strings.Repeat(" ", indent)
+	for _, item := range items {
+		fmt.Fprintf(&b, "%s- %s\n", pad, item)
+	}
+
+	return strings.TrimRight(b.String(), "\n")
+}
diff --git a/integration/k3sic/tls-ca-baking.md b/integration/k3sic/tls-ca-baking.md
new file mode 100644
index 00000000..7d2507f3
--- /dev/null
+++ b/integration/k3sic/tls-ca-baking.md
@@ -0,0 +1,70 @@
+# Running the operator against a private-CA TLS Headscale
+
+`TestK8sOperator` points the Tailscale Kubernetes operator at an **HTTP**
+Headscale (`hsic.WithoutTLS`, `loginServer = http://<ip>:<port>`). That keeps
+the harness small: the operator and proxy pods need no CA, so there is no image
+baking, no containerd import, and no CoreDNS hostname mapping.
+
+This note records how to run the same test against a **TLS** Headscale serving a
+private CA, in case a future test needs to exercise realistic TLS. It is not
+wired up; reconstruct it from here.
+
+## Why it is involved
+
+tailscaled/tsnet verify the control connection against Go's
+`x509.SystemCertPool` (on the Alpine images: `/etc/ssl/certs/ca-certificates.crt`).
+A private CA must therefore be in the pod's **system trust store**.
+
+As of the tailscale operator chart on `main`, there is no supported way to hand
+a CA _file_ to a proxy pod:
+
+- `ProxyClass.spec.statefulSet.pod` has no `volumes`.
+- `ProxyClass...tailscaleContainer` has no `volumeMounts`, and its `env` is a
+  reduced schema (`name`/`value` only — no `valueFrom`/`envFrom`).
+- `operatorConfig` exposes `extraEnv` but no `extraVolumes`.
+
+`SSL_CERT_FILE`/`SSL_CERT_DIR` _are_ honoured by tailscaled, but they need a
+file that cannot be projected in. So the CA has to be baked into the images.
+
+## The recipe
+
+1. Serve Headscale with TLS (the `hsic` default) and grab `headscale.GetCert()`.
+   Pass it to the cluster so in-container `helm`/`kubectl` trust it.
+
+2. Bake the CA into derived operator and proxy images. For each of
+   `tailscale/k8s-operator:<tag>` and `tailscale/tailscale:<tag>`, build:
+
+   ```Dockerfile
+   FROM <base>
+   COPY headscale-ca.crt /usr/local/share/ca-certificates/headscale-ca.crt
+   RUN update-ca-certificates
+   ```
+
+   Build on the host docker daemon (it has egress to pull the base images),
+   `ExportImage` the result to a docker-format tarball, stream it into the k3s
+   container, and import it into the kubelet's containerd namespace:
+
+   ```
+   ctr --namespace k8s.io images import <tarball>
+   ```
+
+   Tag the derived images `headscale.local/...:<tag>-ca` and run them with
+   `imagePullPolicy: Never` (the `headscale.local/` prefix is never resolved by
+   a registry).
+
+3. Wire the derived images into the chart via `operatorConfig.image` /
+   `proxyConfig.image`. The operator's proxy StatefulSet template hard-codes
+   `imagePullPolicy: Always`, so a `ProxyClass` must override the proxy image
+   **and** set `imagePullPolicy: Never`; Connectors must reference that
+   ProxyClass explicitly (`defaultProxyClass` does not apply to them).
+
+4. Pods resolve Headscale through CoreDNS, not the container's `/etc/hosts`, and
+   the cert's only SAN is the Headscale hostname (dialing by IP fails TLS
+   verification). Install a `coredns-custom` ConfigMap mapping the hostname to
+   the Headscale IP and gate on the `kube-dns` Service having ready endpoints
+   before starting the operator.
+
+The full implementation lived in `integration/k3sic/k3sic.go` and
+`integration/k8s_operator_test.go` before the switch to HTTP; recover it from
+git history (`PrepareTailscaleImages`, `bakeAndImportImage`, `caBuildContext`,
+`ConfigureCoreDNSHost`) if needed.
diff --git a/integration/k8s_operator_test.go b/integration/k8s_operator_test.go
new file mode 100644
index 00000000..7af22ae8
--- /dev/null
+++ b/integration/k8s_operator_test.go
@@ -0,0 +1,266 @@
+package integration
+
+import (
+	"fmt"
+	"net/netip"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	clientv1 "github.com/juanfont/headscale/gen/client/v1"
+	policyv2 "github.com/juanfont/headscale/hscontrol/policy/v2"
+	"github.com/juanfont/headscale/integration/hsic"
+	"github.com/juanfont/headscale/integration/integrationutil"
+	"github.com/juanfont/headscale/integration/k3sic"
+	"github.com/juanfont/headscale/integration/tsic"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"tailscale.com/tailcfg"
+)
+
+const (
+	tagK8sOperator = "tag:k8s-operator"
+	tagK8s         = "tag:k8s"
+)
+
+// k8sOperatorPolicy is the tagOwners policy the Tailscale Kubernetes operator
+// requires: tag:k8s-operator is self/admin-owned, and the operator
+// (tag:k8s-operator) owns tag:k8s so it can mint auth keys for the proxy nodes
+// it spins up. The wildcard ACL lets the in-cluster proxies and the out-of-cluster
+// tsic client reach each other for the connectivity checks.
+func k8sOperatorPolicy() *policyv2.Policy {
+	return &policyv2.Policy{
+		TagOwners: policyv2.TagOwners{
+			tagK8sOperator: policyv2.Owners{},
+			tagK8s:         policyv2.Owners{new(policyv2.Tag(tagK8sOperator))},
+		},
+		ACLs: []policyv2.ACL{
+			{
+				Action:  "accept",
+				Sources: []policyv2.Alias{policyv2.Wildcard},
+				Destinations: []policyv2.AliasWithPorts{
+					{Alias: policyv2.Wildcard, Ports: []tailcfg.PortRange{tailcfg.PortRangeAny}},
+				},
+			},
+		},
+	}
+}
+
+// TestK8sOperator verifies that the real Tailscale Kubernetes operator, installed
+// into a real k3s cluster via its Helm chart and pointed at an in-test Headscale,
+// can authenticate with OAuth client credentials, mint auth keys, and register
+// nodes that then interoperate with a regular tailnet node.
+//
+// The operator targets Headscale over plain HTTP by IP (hsic.WithoutTLS +
+// GetIPEndpoint), so the operator and proxy pods need no CA and no DNS entry for
+// Headscale. See integration/k3sic/tls-ca-baking.md for the TLS variant.
+//
+// The cluster-side steps are reusable building blocks on k3sic.K3sInContainer
+// (InstallOperator, DeployConnector, DeployEchoServer, ExposeServiceToTailnet,
+// DeployProxyGroup), so further operator scenarios are a few method calls.
+//
+// Run it with `go run ./cmd/hi run "TestK8sOperator"`.
+func TestK8sOperator(t *testing.T) {
+	IntegrationSkip(t)
+
+	// One regular user+node provides the out-of-cluster tailnet peer used by the
+	// connectivity subtest; the operator registers its own nodes on top.
+	spec := ScenarioSpec{
+		Users:        []string{"k8s-user"},
+		NodesPerUser: 1,
+	}
+
+	scenario, err := NewScenario(spec)
+	require.NoError(t, err)
+
+	defer scenario.ShutdownAssertNoPanics(t)
+
+	err = scenario.CreateHeadscaleEnv(
+		// The tsic client reaches the in-cluster proxy only via DERP (no direct
+		// path to the k3s pod network), and hsic's embedded DERP is non-TLS, so the
+		// client must reach DERP over plain-HTTP websockets — as the proxy pods do.
+		[]tsic.Option{tsic.WithDERPOverHTTP()},
+		hsic.WithTestName("k8soperator"),
+		hsic.WithoutTLS(),
+		hsic.WithACLPolicy(k8sOperatorPolicy()),
+	)
+	require.NoError(t, err)
+
+	headscale, err := scenario.Headscale()
+	require.NoError(t, err)
+
+	// Mint an OAuth client for the operator: devices:core + auth_keys scopes,
+	// tagged tag:k8s-operator. CreateOAuthClient mints the admin API key and calls
+	// the v2 keys API itself.
+	clientID, clientSecret, err := headscale.CreateOAuthClient(
+		t.Context(),
+		[]string{"devices:core", "auth_keys"},
+		[]string{tagK8sOperator},
+	)
+	require.NoError(t, err, "creating OAuth client (server-side OAuth must be implemented)")
+	require.NotEmpty(t, clientID)
+	require.NotEmpty(t, clientSecret)
+
+	// Bring up the k3s cluster on the scenario networks.
+	k3s, err := k3sic.New(scenario.Pool(), scenario.Networks())
+	require.NoError(t, err)
+
+	defer func() {
+		shutdownErr := k3s.Shutdown()
+		if shutdownErr != nil {
+			t.Logf("shutting down k3s: %s", shutdownErr)
+		}
+	}()
+
+	// Registered after Shutdown so it runs first (defers are LIFO): dump cluster
+	// state while it is still up if anything below fails.
+	defer func() {
+		if t.Failed() {
+			k3s.DumpDiagnostics()
+		}
+	}()
+
+	require.NoError(t, k3s.WaitForRunning())
+	require.NoError(t, k3s.InstallHelm())
+
+	// The operator reaches the control plane by IP, but the embedded DERP map
+	// references Headscale by hostname; teach CoreDNS to resolve it so the proxy
+	// pods can connect to DERP and get a data path to nodes outside the cluster.
+	hsIP := headscale.GetIPInNetwork(scenario.Networks()[0])
+	require.NoError(t, k3s.ConfigureCoreDNSHost(headscale.GetHostname(), hsIP))
+
+	// loginServer is the in-cluster-reachable HTTP endpoint by IP; the operator
+	// uses it for both the control plane and the management API.
+	loginServer := headscale.GetIPEndpoint()
+	require.NoError(t, k3s.InstallOperator(loginServer, clientID, clientSecret))
+
+	t.Run("operator-registers", func(t *testing.T) {
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			nodes, err := headscale.ListNodes()
+			assert.NoError(c, err)
+			assert.True(c, hasNodeWithTag(nodes, tagK8sOperator),
+				"expected a node tagged %s registered by the operator, got %s",
+				tagK8sOperator, describeNodes(nodes))
+		}, integrationutil.ScaledTimeout(180*time.Second), 2*time.Second,
+			"operator node should register and be tagged "+tagK8sOperator)
+	})
+
+	t.Run("egress-connector", func(t *testing.T) {
+		require.NoError(t, k3s.DeployConnector("k8s-egress", []string{tagK8s}, []string{"10.40.0.0/14"}))
+
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			nodes, err := headscale.ListNodes()
+			assert.NoError(c, err)
+			assert.True(c, hasNodeWithTag(nodes, tagK8s),
+				"expected a proxy node tagged %s registered by the operator, got %s",
+				tagK8s, describeNodes(nodes))
+		}, integrationutil.ScaledTimeout(180*time.Second), 2*time.Second,
+			"egress proxy node should register and be tagged "+tagK8s)
+	})
+
+	t.Run("ingress-service-reachable-from-tailnet", func(t *testing.T) {
+		require.NoError(t, k3s.DeployEchoServer("echo"))
+		require.NoError(t, k3s.ExposeServiceToTailnet("echo", []string{tagK8s}))
+
+		clients, err := scenario.ListTailscaleClients("k8s-user")
+		require.NoError(t, err)
+		require.NotEmpty(t, clients)
+
+		// The operator registers the ingress proxy as a tailnet node named after
+		// the exposed Service (<namespace>-<service>, here default-echo-ts). Read
+		// its IP from Headscale rather than the Service's LoadBalancer status,
+		// which the operator does not populate against Headscale.
+		var svcIP string
+
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			nodes, err := headscale.ListNodes()
+			assert.NoError(c, err)
+
+			ip, ok := nodeIPv4ByName(nodes, "echo")
+			assert.True(c, ok, "ingress proxy node for echo should register, got %s", describeNodes(nodes))
+
+			svcIP = ip
+		}, integrationutil.ScaledTimeout(180*time.Second), 2*time.Second,
+			"operator should register an ingress proxy node for the exposed service")
+
+		// The out-of-cluster node reaches the in-cluster service through the proxy
+		// over the tailnet. /hostname is an agnhost endpoint that returns the
+		// backend pod's hostname, proving the request reached the service.
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			body, err := clients[0].Curl("http://" + svcIP + "/hostname")
+			assert.NoError(c, err)
+			assert.NotEmpty(c, body, "expected a response from the exposed service")
+		}, integrationutil.ScaledTimeout(120*time.Second), 2*time.Second,
+			"tsic node should reach the k8s-exposed service over the tailnet")
+	})
+
+	t.Run("proxy-group", func(t *testing.T) {
+		nodes, err := headscale.ListNodes()
+		require.NoError(t, err)
+
+		before := countNodesWithTag(nodes, tagK8s)
+
+		const replicas = 2
+
+		require.NoError(t, k3s.DeployProxyGroup("ts-ingress", "ingress", replicas, []string{tagK8s}))
+
+		// A ProxyGroup runs a pool of proxies; each replica registers its own node.
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			nodes, err := headscale.ListNodes()
+			assert.NoError(c, err)
+			assert.GreaterOrEqual(c, countNodesWithTag(nodes, tagK8s), before+replicas,
+				"expected %d more %s nodes from the ProxyGroup, got %s",
+				replicas, tagK8s, describeNodes(nodes))
+		}, integrationutil.ScaledTimeout(180*time.Second), 2*time.Second,
+			"ProxyGroup replicas should each register a node tagged "+tagK8s)
+	})
+}
+
+// hasNodeWithTag reports whether any node carries the given tag.
+func hasNodeWithTag(nodes []*clientv1.Node, tag string) bool {
+	return countNodesWithTag(nodes, tag) > 0
+}
+
+// countNodesWithTag counts the nodes carrying the given tag.
+func countNodesWithTag(nodes []*clientv1.Node, tag string) int {
+	count := 0
+
+	for _, node := range nodes {
+		if slices.Contains(node.Tags, tag) {
+			count++
+		}
+	}
+
+	return count
+}
+
+// nodeIPv4ByName returns the IPv4 of the first node whose given name contains
+// substr, identifying an operator-registered proxy by the Service/Connector it
+// fronts (e.g. "echo" matches the default-echo-ts ingress proxy).
+func nodeIPv4ByName(nodes []*clientv1.Node, substr string) (string, bool) {
+	for _, node := range nodes {
+		if !strings.Contains(node.GivenName, substr) {
+			continue
+		}
+
+		for _, ip := range node.IpAddresses {
+			addr, err := netip.ParseAddr(ip)
+			if err == nil && addr.Is4() {
+				return ip, true
+			}
+		}
+	}
+
+	return "", false
+}
+
+// describeNodes renders a compact name->tags summary for failure messages.
+func describeNodes(nodes []*clientv1.Node) string {
+	parts := make([]string, 0, len(nodes))
+	for _, node := range nodes {
+		parts = append(parts, fmt.Sprintf("%s%v", node.Name, node.Tags))
+	}
+
+	return "[" + strings.Join(parts, " ") + "]"
+}