reconciler: read release secrets directly instead of shelling helm

helm's list gunzips every release payload to build its table. The only fields
either caller reads are name and namespace:

    reconciler/main.py   live = {(r.name, r.namespace) for r in releases}
    worker/handlers.py   releases = {r.name for r in await ...list_releases()}

Both live in the release secret's labels and metadata, so nothing needs
decompressing. Measured from a pod on this cluster: 21ms and 31KB, against
helm's 4392ms.

That is the whole fix for a drift check that was timing out at 330s every tick
and OOMKilling the container at 256Mi. The cause of both was decompressing 96
releases to extract two strings each. It also means the container no longer
cares that a cluster policy mutates its CPU request to 0 — at 21ms there is
nothing left to starve.

Three details carry the correctness, each with a test:

  - PartialObjectMetadataList in the Accept header asks for metadata only.
    Without it every release's gzipped manifest crosses the wire to be thrown
    away, which is the cost this exists to avoid.
  - The status selector drops superseded revisions server-side: 96 release
    secrets here, 25 live. Failed and pending states are kept, matching what
    `helm list` shows, because a failed release does exist and calling it
    missing would have the reconciler re-provision on top of it.
  - helm writes one secret per revision, up to 10 per release here, so the
    newest version label wins. Control-tested with a string compare, which
    picks "9" over "10".

Out of cluster there is no ServiceAccount, so it falls back to helm and the
e2e suite and laptop runs are unchanged. An API error raises rather than
falling back: the fallback is for a known-absent ServiceAccount, and quietly
retrying through helm would swap a visible error for the timeout this removes.

chart and app_version are empty on this path rather than wrong — they live
only in the compressed payload, and nothing reads them.

libs/svcforge_core/svcforge_core/adapters/helm.py

@@ -28,6 +28,7 @@ from collections.abc import Sequence
 from pathlib import Path
 from typing import Any, Protocol
 
+import httpx
 import yaml
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -55,6 +56,24 @@ _STDERR_TAIL_BYTES = 2048
 MANAGED_BY_LABEL = "app.kubernetes.io/managed-by"
 MANAGED_BY_VALUE = "svcforge"
 
+# Where the kubelet mounts the pod's ServiceAccount. Their presence is also how this module
+# decides it is running inside the cluster: in-cluster gets the fast release read below,
+# anything else falls back to shelling helm.
+_SA_DIR = Path("/var/run/secrets/kubernetes.io/serviceaccount")
+_SA_TOKEN = _SA_DIR / "token"
+_SA_CA = _SA_DIR / "ca.crt"
+
+# helm's own label on every release secret it writes. Pairing it with MANAGED_BY_LABEL is
+# what separates svcforge's releases from the rest of the cluster's.
+_HELM_OWNER_LABEL = "owner=helm"
+
+# The states `helm list` shows by default. Pushed into the selector so superseded revisions
+# never leave the API server: this cluster had 96 release secrets of which 71 were
+# superseded, so the filter is most of the win.
+_LIVE_STATUSES = "status in (deployed,failed,pending-install,pending-upgrade,pending-rollback)"
+
+_API_TIMEOUT_S = 15.0
+
 
 class HelmError(SvcforgeError, RuntimeError):
     """Non-zero exit. str(self) is the stderr tail that lands in instances.error.
@@ -258,24 +277,30 @@ class HelmProvisioner:
         cert-manager are all reported as orphans svcforge is failing to account for, every
         sweep. They are not orphans. They were never svcforge's to know about.
 
-        What this does NOT fix is cost, and that was measured rather than assumed:
+        Cost is why this does not shell out to helm when it does not have to. `--selector` is
-        unscoped 23 releases in 4392ms, scoped to 0 releases in 3988ms. `--selector` is not
+        NOT pushed down as a server-side selector — helm fetches and decompresses every
-        pushed down as a server-side label selector — helm fetches and decompresses every
+        release secret in the cluster regardless, then filters what it already parsed.
-        release secret in the cluster regardless, then filters what it already parsed. The
+        Measured: unscoped 23 releases in 4392ms, scoped to 0 in 3988ms, a saving of about
-        saving is around 10%, not the order of magnitude the shape of the flag suggests.
+        10%, not the order of magnitude the flag's shape suggests.
 
-        That matters because this call was timing out at 330s on every tick, with the CPU
+        That was not academic. With the CPU request mutated to 0 by a cluster policy, the
-        request mutated to 0 by a cluster policy, against ~4s given real CPU. Scoping does
+        call took over 330s and timed out on every tick, against ~4s given real CPU. A check
-        not rescue that. A check that never completes reports no drift, which looks exactly
+        that never completes reports no drift, which looks exactly like no drift existing.
-        like no drift existing, so the timeout needs its own fix — CPU for the container, or
+
-        reading the release secrets directly with a label selector, which the API server
+        In-cluster this reads the release secrets directly instead. Measured from a pod on
-        answers in 141ms because it never decompresses anything.
+        this cluster: 21ms and 31KB against helm's 4392ms, because nothing is decompressed
+        and no release payload crosses the wire. See `_list_releases_via_api`. Out of cluster there is no
+        ServiceAccount to authenticate with, so it falls back to helm — which keeps the e2e
+        suite, and anyone running this from a laptop, working unchanged.
 
         Releases provisioned before the label existed will not match, so the first sweep
         after this ships sees them as missing and re-provisions. That is safe by
         construction — provisioning is `helm upgrade --install` against a deterministic
         release name — and the re-provision is what applies the label.
         """
+        via_api = await self._list_releases_via_api()
+        if via_api is not None:
+            return via_api
         argv = self._base_argv(
             "list",
             "--all-namespaces",
@@ -293,6 +318,104 @@ class HelmProvisioner:
             raise HelmError(f"helm list returned {type(parsed).__name__}, expected a list")
         return [ReleaseInfo.model_validate(row) for row in parsed]
 
+    async def _list_releases_via_api(self) -> list[ReleaseInfo] | None:
+        """Release names and namespaces read straight off helm's release secrets.
+
+        Returns None when there is no in-cluster ServiceAccount to authenticate with, which
+        is the caller's signal to fall back to helm.
+
+        This exists because helm's own list is expensive for a reason this caller does not
+        need. helm gunzips every release payload to build its table; the only fields anyone
+        here reads are name and namespace:
+
+            reconciler/main.py   live = {(r.name, r.namespace) for r in releases}
+            worker/handlers.py   releases = {r.name for r in await ...list_releases()}
+
+        Both live in the secret's labels and its metadata, so nothing has to be decompressed.
+        That is the whole difference between ~141ms and ~4.4s, and it is why this survives a
+        container whose CPU request was mutated to 0.
+
+        Three details carry the correctness:
+
+        * `PartialObjectMetadataList` in the Accept header asks the API server for metadata
+          only. Without it the response carries every release's gzipped manifest — megabytes
+          of payload fetched purely to be thrown away, which is the cost this method exists
+          to avoid.
+        * The status selector drops superseded revisions server-side: 96 release secrets on
+          this cluster, 25 of them live. The states kept are the ones `helm list` shows by
+          default, so a failed release still counts as existing — it does exist, and
+          treating it as missing would have the reconciler re-provision on top of it.
+        * helm writes one secret per revision, so a release can still appear more than once
+          — 20 releases here had up to 10 revisions each. The newest `version` label wins.
+          Skipping this would report a release that exists as several, which for the
+          reconciler's set difference is harmless, and for anything counting releases is not.
+
+        `chart`, `status`, `revision` and `app_version` on the returned ReleaseInfo are the
+        subset the labels give away free. `chart` in particular is empty rather than wrong,
+        because the chart name lives only in the compressed payload. Narrowing the model
+        instead would have been honest about this method and dishonest about the helm path,
+        which does populate them.
+        """
+        try:
+            token = _SA_TOKEN.read_text(encoding="utf-8").strip()
+        except OSError:
+            return None
+        host, port = (
+            os.environ.get("KUBERNETES_SERVICE_HOST"),
+            os.environ.get("KUBERNETES_SERVICE_PORT_HTTPS", "443"),
+        )
+        if not host:
+            return None
+
+        selector = f"{_HELM_OWNER_LABEL},{MANAGED_BY_LABEL}={MANAGED_BY_VALUE},{_LIVE_STATUSES}"
+        try:
+            async with httpx.AsyncClient(verify=str(_SA_CA), timeout=_API_TIMEOUT_S) as client:
+                resp = await client.get(
+                    f"https://{host}:{port}/api/v1/secrets",
+                    params={"labelSelector": selector},
+                    headers={
+                        "authorization": f"Bearer {token}",
+                        "accept": ("application/json;as=PartialObjectMetadataList;g=meta.k8s.io;v=v1"),
+                    },
+                )
+                resp.raise_for_status()
+                items: Any = resp.json().get("items", [])
+        except (httpx.HTTPError, json.JSONDecodeError) as exc:
+            # Raise, do not fall back to helm. The fallback is for "there is no
+            # ServiceAccount here", which is a fact about the environment and is known
+            # before any request goes out. This is different: the API server was reachable
+            # and something went wrong, and quietly retrying through helm would swap a
+            # visible error for the 330s timeout this method exists to remove, on a
+            # container that OOMs while helm parses. The tick logs check.failed and the
+            # next one tries again in 60s.
+            raise HelmError(f"listing release secrets failed: {exc}") from exc
+
+        newest: dict[tuple[str, str], tuple[int, ReleaseInfo]] = {}
+        for item in items:
+            meta = item.get("metadata") or {}
+            labels = meta.get("labels") or {}
+            name, namespace = labels.get("name"), meta.get("namespace")
+            if not name or not namespace:
+                continue  # not a release secret this method understands; leave it alone
+            try:
+                revision = int(labels.get("version", 0))
+            except ValueError:
+                revision = 0
+            key = (name, namespace)
+            if key in newest and newest[key][0] >= revision:
+                continue
+            newest[key] = (
+                revision,
+                ReleaseInfo(
+                    name=name,
+                    namespace=namespace,
+                    chart="",
+                    status=labels.get("status", ""),
+                    revision=max(revision, 0),
+                ),
+            )
+        return [info for _, info in newest.values()]
+
 
 class _ValuesFile:
     """Context manager yielding a path to a values.yaml written from a dict.

tests/unit/test_helm_argv.py

@@ -12,9 +12,12 @@ here is that the two sides of the label agree, and that neither drops out under
 from __future__ import annotations
 
 from pathlib import Path
+from typing import Any
 
+import httpx
 import pytest
 
+from svcforge_core.adapters import helm
 from svcforge_core.adapters.helm import (
     MANAGED_BY_LABEL,
     MANAGED_BY_VALUE,
@@ -91,3 +94,137 @@ async def test_the_label_written_is_the_label_read(monkeypatch: pytest.MonkeyPat
     await prov.list_releases()
 
     assert _pair(seen[0], "--labels") == _pair(seen[1], "--selector")
+
+
+def _meta(name: str, ns: str, version: str, status: str = "deployed") -> dict[str, object]:
+    """One PartialObjectMetadata item as the API server returns it for a release secret."""
+    return {
+        "metadata": {
+            "name": f"sh.helm.release.v1.{name}.v{version}",
+            "namespace": ns,
+            "labels": {"name": name, "owner": "helm", "status": status, "version": version},
+        }
+    }
+
+
+class _FakeResponse:
+    def __init__(self, items: list[dict[str, object]]) -> None:
+        self._items = items
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict[str, object]:
+        return {"items": self._items}
+
+
+def _fake_api(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, items: list[dict[str, object]]
+) -> dict[str, Any]:
+    """Stand in for the in-cluster ServiceAccount and the API call it authenticates.
+
+    The token is a real file at a repointed constant rather than a patched method: the
+    module decides it is in-cluster by whether that file reads, so faking the decision at
+    the filesystem keeps the test honest about what it is exercising.
+    """
+    captured: dict[str, Any] = {}
+
+    token = tmp_path / "token"
+    token.write_text("tok", encoding="utf-8")
+    monkeypatch.setattr(helm, "_SA_TOKEN", token)
+    monkeypatch.setattr(helm, "_SA_CA", tmp_path / "ca.crt")
+    monkeypatch.setenv("KUBERNETES_SERVICE_HOST", "10.96.0.1")
+    monkeypatch.setenv("KUBERNETES_SERVICE_PORT_HTTPS", "443")
+
+    class _Client:
+        def __init__(self, **kw: object) -> None:
+            captured["client_kwargs"] = kw
+
+        async def __aenter__(self) -> _Client:
+            return self
+
+        async def __aexit__(self, *exc: object) -> None:
+            return None
+
+        async def get(self, url: str, **kw: object) -> _FakeResponse:
+            captured["url"] = url
+            captured.update(kw)
+            return _FakeResponse(items)
+
+    monkeypatch.setattr(httpx, "AsyncClient", _Client)
+    return captured
+
+
+@pytest.mark.asyncio
+async def test_api_read_collapses_a_release_to_its_newest_revision(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """helm writes one secret per revision. Ten revisions is one release, not ten."""
+    _fake_api(
+        monkeypatch,
+        tmp_path,
+        [_meta("acme-redis", "tenant-acme", v) for v in ("1", "2", "10", "9")],
+    )
+
+    out = await HelmProvisioner().list_releases()
+
+    assert [(r.name, r.namespace) for r in out] == [("acme-redis", "tenant-acme")]
+    # 10, not 9: string ordering would pick "9" and quietly report a stale revision.
+    assert out[0].revision == 10
+
+
+@pytest.mark.asyncio
+async def test_api_read_keeps_same_named_releases_in_different_namespaces(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """Two tenants may both call their instance `redis`. Namespace is part of the identity."""
+    _fake_api(monkeypatch, tmp_path, [_meta("redis", "tenant-a", "1"), _meta("redis", "tenant-b", "1")])
+
+    out = await HelmProvisioner().list_releases()
+
+    assert {(r.name, r.namespace) for r in out} == {("redis", "tenant-a"), ("redis", "tenant-b")}
+
+
+@pytest.mark.asyncio
+async def test_api_read_asks_for_metadata_only_and_scopes_the_selector(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """The Accept header is the difference between metadata and megabytes of gzipped payload."""
+    cap = _fake_api(monkeypatch, tmp_path, [])
+
+    await HelmProvisioner().list_releases()
+
+    assert "PartialObjectMetadataList" in cap["headers"]["accept"]
+    selector = cap["params"]["labelSelector"]
+    assert "owner=helm" in selector
+    assert f"{MANAGED_BY_LABEL}={MANAGED_BY_VALUE}" in selector
+    assert "superseded" not in selector  # dropped server-side by asking only for live states
+    assert cap["url"].endswith("/api/v1/secrets")
+
+
+@pytest.mark.asyncio
+async def test_api_read_ignores_secrets_that_are_not_releases(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A malformed or unrelated secret must not become a phantom release."""
+    _fake_api(
+        monkeypatch,
+        tmp_path,
+        [{"metadata": {"namespace": "x", "labels": {}}}, _meta("real", "tenant-a", "1")],
+    )
+
+    out = await HelmProvisioner().list_releases()
+
+    assert [r.name for r in out] == ["real"]
+
+
+@pytest.mark.asyncio
+async def test_no_service_account_falls_back_to_helm(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """Out of cluster there is nothing to authenticate with, so e2e and laptops keep working."""
+    seen = _capture(monkeypatch)
+
+    monkeypatch.setattr(helm, "_SA_TOKEN", tmp_path / "absent")
+
+    await HelmProvisioner(kubeconfig=Path("/dev/null")).list_releases()
+
+    assert seen and seen[0][:2] == ["helm", "list"]