feat: helm_list_releases — release status reachable over MCP/agent

docs/mcp.md

@@ -7,7 +7,8 @@ Start with `korvid --mcp` (or set `mcp: {enabled: true}` in
 external agents — VS Code Copilot Chat, Claude Code, Cursor, Zed — over a
 [Streamable HTTP MCP](https://modelcontextprotocol.io) server bound to
 `127.0.0.1:7878` (`mcp: {port: N}` to change).  External hosts can list
-resources, fetch manifests, logs, and events, and drive the TUI you are
+resources, fetch manifests, logs, events, and helm release status, and
+drive the TUI you are
 looking at (navigate, filter, open logs/describe).  Write tools are **not**
 exposed: cluster mutations stay behind the in-TUI confirmation dialog.
 An opt-in *proposal* flow lets external agents queue writes for your review
@@ -63,6 +64,7 @@ mirrors those reads in the TUI so you can watch the assistant work:
 | `get_logs` | log pane on that pod/container |
 | `diagnose_pod` | describe pane on the pod |
 | `list_operators` | navigate to subscriptions |
+| `helm_list_releases` | navigate to the helm release browser |
 
 Off by default — screen hijacking mid-task is worse than invisibility.
 Enable at startup with `mcp: {follow: true}` in the config, or live with

src/korvid/agent/profiles.py

@@ -83,6 +83,10 @@
     "list_operators": (
         "List OLM operator packages and installed subscriptions with their status. Read-only."
     ),
+    "helm_list_releases": (
+        "List installed Helm releases with revision, status, chart and app "
+        "version. Read-only; parsed from cluster Secrets."
+    ),
     "open_logs": "Open the live log pane for a pod on the user's screen.",
     "resize_pod": (
         "Request an in-place CPU/memory resize of a running pod (Kubernetes "

src/korvid/evals/fake_kube.py

@@ -20,6 +20,7 @@
 from korvid.evals.scenario import SCENARIO_NOW, TIMESTAMP_PATTERN, Scenario
 from korvid.k8s.discovery import ResourceMeta
 from korvid.k8s.errors import ApiStatusError
+from korvid.k8s.helm import HELM_SECRET_TYPE, HelmReleaseSummary, release_from_secret
 from korvid.k8s.logs import LogLine
 from korvid.k8s.models import GenericSummary, summary_for
 from korvid.k8s.reads import ReadOps
@@ -119,6 +120,24 @@ async def get_object(
                 return deepcopy(manifest)
         raise ApiStatusError(404, f"{meta.plural} {namespace or ''}/{name} not found")
 
+    async def list_helm_releases(self, namespace: str | None) -> list[HelmReleaseSummary]:
+        """Latest revision per release from helm-owned Secrets in the scenario."""
+        latest: dict[tuple[str, str], HelmReleaseSummary] = {}
+        for manifest in self._objects:
+            if str(manifest.get("type") or "") != HELM_SECRET_TYPE:
+                continue
+            metadata = manifest.get("metadata") or {}
+            if (metadata.get("labels") or {}).get("owner") != "helm":
+                continue
+            if namespace is not None and str(metadata.get("namespace") or "") != namespace:
+                continue
+            release = release_from_secret(manifest)
+            key = (release.namespace, release.name)
+            current = latest.get(key)
+            if current is None or release.revision > current.revision:
+                latest[key] = release
+        return sorted(latest.values(), key=lambda r: (r.namespace, r.name))
+
     async def list_events_for(
         self,
         namespace: str,

src/korvid/k8s/client.py

@@ -595,6 +595,23 @@ async def watch_helm_releases(
             for out in tracker.apply(event_type, release_from_secret(secret)):
                 yield out
 
+    async def list_helm_releases(self, namespace: str | None) -> list[HelmReleaseSummary]:
+        """Latest revision per release, LIST-only (the helm_list_releases
+        tool, issue #161): same Secret parsing as the browser's synthetic
+        kind — no helm binary involved."""
+        if self._api is None:
+            raise RuntimeError("connect() first")
+        base = self._helm_secrets_base(namespace)
+        data = await self._request_json(f"{base}?{urlencode(self._helm_secrets_query())}")
+        latest: dict[tuple[str, str], HelmReleaseSummary] = {}
+        for item in data.get("items", []):
+            release = release_from_secret(item)
+            key = (release.namespace, release.name)
+            current = latest.get(key)
+            if current is None or release.revision > current.revision:
+                latest[key] = release
+        return sorted(latest.values(), key=lambda r: (r.namespace, r.name))
+
     async def watch_helm_revisions(
         self, namespace: str | None
     ) -> AsyncIterator[tuple[str, HelmRevisionSummary]]:

src/korvid/k8s/reads.py

@@ -14,6 +14,7 @@
 from typing import Any
 
 from korvid.k8s.discovery import ResourceMeta
+from korvid.k8s.helm import HelmReleaseSummary
 from korvid.k8s.logs import LogLine
 from korvid.k8s.models import GenericSummary
 
@@ -31,6 +32,10 @@ async def get_object(
     ) -> dict[str, Any]:
         """Fetch the raw manifest for a single object; 404 → ApiStatusError."""
 
+    @abc.abstractmethod
+    async def list_helm_releases(self, namespace: str | None) -> list[HelmReleaseSummary]:
+        """Latest revision per helm release (Secret-parsed, no helm binary)."""
+
     @abc.abstractmethod
     async def list_events_for(
         self,

src/korvid/tools/executor.py

@@ -645,6 +645,18 @@ async def _list_resources(self, args: dict[str, Any]) -> str:
             lines.append(line)
         return "\n".join(lines)
 
+    async def _helm_list_releases(self, args: dict[str, Any]) -> str:
+        """Installed helm releases with status (issue #161): parsed from the
+        cluster's release Secrets - same path as the TUI's helm browser, so
+        the tool line and the table always agree."""
+        namespace: str | None = args.get("namespace")
+        releases = await self._kube.list_helm_releases(namespace)
+        if not releases:
+            return "(none)"
+        return "\n".join(
+            f"{r.namespace}/{r.name}  -  age={r.age()}  {summary_facts(r)}" for r in releases
+        )
+
     async def _list_operators(self, args: dict[str, Any]) -> str:
         """Catalog packages + installed subscriptions, straight from the
         cluster's own OLM objects (issue #29: no hardcoded operator

src/korvid/tools/follow.py

@@ -31,6 +31,7 @@
         "get_logs",
         "get_events",
         "list_operators",
+        "helm_list_releases",
         "diagnose_pod",
     }
 )
@@ -99,8 +100,18 @@ async def mirror_read(ui: UIBridge, tool: str, args: Mapping[str, Any]) -> str |
         return None
 
 
+#: Reads that mirror as a plain view navigation: tool -> view alias.
+_NAVIGATE_MIRRORS: dict[str, str] = {
+    "list_operators": "subscriptions",
+    "helm_list_releases": "helm",
+}
+
+
 async def _mirror(ui: UIBridge, tool: str, args: Mapping[str, Any]) -> str | None:
     namespace = _str_or_none(args.get("namespace"))
+    view = _NAVIGATE_MIRRORS.get(tool)
+    if view is not None:
+        return await ui.agent_navigate(view, namespace or "all")
     if tool == "list_resources":
         kind = _str_or_none(args.get("kind"))
         if kind is None:
@@ -130,6 +141,4 @@ async def _mirror(ui: UIBridge, tool: str, args: Mapping[str, Any]) -> str | Non
         if pod is None:
             return None
         return await ui.agent_open_describe("pods", pod, namespace)
-    if tool == "list_operators":
-        return await ui.agent_navigate("subscriptions", namespace or "all")
     return None

src/korvid/tools/registry.py

@@ -396,6 +396,38 @@ def mcp_tool_schemas(*, write_proposals: bool = False) -> list[dict[str, Any]]:
             },
         },
     ),
+    ToolDef(
+        name="helm_list_releases",
+        effect="cluster_read",
+        dispatch="_helm_list_releases",
+        surfaces=_ALL_SURFACES,
+        schema={
+            "type": "function",
+            "function": {
+                "name": "helm_list_releases",
+                "description": (
+                    "List installed Helm releases with their status: one line"
+                    " per release - revision, status (deployed/failed/"
+                    "pending-…), chart and app version. Read-only, parsed"
+                    " from the cluster's own release Secrets (no helm binary"
+                    " involved); installing or upgrading a release is done by"
+                    " the user through the UI."
+                ),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "namespace": {
+                            "type": "string",
+                            "description": (
+                                "Namespace to scope releases to. Omit for all namespaces."
+                            ),
+                        },
+                    },
+                    "required": [],
+                },
+            },
+        },
+    ),
     ToolDef(
         name="diagnose_pod",
         effect="cluster_read",

tests/k8s/test_helm.py

@@ -511,3 +511,39 @@ async def test_unknown_release_raises_not_found(self) -> None:
             pytest.raises(ApiStatusError, match="not found"),
         ):
             await client.get_helm_release_components("default", "ghost")
+
+
+class TestListHelmReleases:
+    """LIST-only release listing for the helm_list_releases tool (#161)."""
+
+    async def test_latest_revision_per_release(self) -> None:
+        client = KubeClient()
+        list_resp: dict[str, Any] = {
+            "items": [
+                _secret("web", 1, "superseded"),
+                _secret("web", 3, "deployed"),
+                _secret("api", 2, "failed"),
+            ]
+        }
+        with (
+            patch.object(client, "_api", MagicMock()),
+            patch.object(client, "_request_json", AsyncMock(return_value=list_resp)),
+        ):
+            releases = await client.list_helm_releases("default")
+        assert [(r.name, r.revision, r.status) for r in releases] == [
+            ("api", 2, "failed"),
+            ("web", 3, "deployed"),
+        ]
+
+    async def test_cluster_wide_when_namespace_is_none(self) -> None:
+        client = KubeClient()
+        request_json = AsyncMock(return_value={"items": []})
+        with (
+            patch.object(client, "_api", MagicMock()),
+            patch.object(client, "_request_json", request_json),
+        ):
+            assert await client.list_helm_releases(None) == []
+        assert request_json.await_args is not None
+        path = request_json.await_args.args[0]
+        assert path.startswith("/api/v1/secrets?")
+        assert "labelSelector=owner%3Dhelm" in path  # helm-owned Secrets only

tests/tools/test_executor.py

@@ -34,6 +34,7 @@ def test_read_tools_schema_names() -> None:
         "get_logs",
         "get_events",
         "list_operators",
+        "helm_list_releases",
         "diagnose_pod",
     ]
 

tests/tools/test_follow.py

@@ -107,6 +107,7 @@ async def test_every_followable_tool_reaches_the_bridge(tool: str) -> None:
         "get_logs": {"pod": "x", "namespace": "d"},
         "diagnose_pod": {"pod": "x", "namespace": "d"},
         "list_operators": {},
+        "helm_list_releases": {},
     }[tool]
     ui = FakeBridge()
     result = await mirror_read(ui, tool, args)
@@ -137,3 +138,10 @@ def test_read_summary_sanitizes_and_bounds_hostile_arguments() -> None:
     assert "\x1b" not in line
     assert "\u202e" not in line  # bidi override cannot reorder the toast
     assert len(line) <= 200
+
+
+async def test_helm_list_releases_mirrors_as_the_helm_view() -> None:
+    ui = FakeBridge()
+    result = await mirror_read(ui, "helm_list_releases", {"namespace": "prod"})
+    assert result is not None
+    assert ui.calls == [("navigate", {"view": "helm", "namespace": "prod"})]

tests/tools/test_list_resources.py

@@ -280,3 +280,52 @@ def test_helm_revision_facts_include_app_version() -> None:
         app_version="2.7.1",
     )
     assert "app_version=2.7.1" in facts(s)
+
+
+# ---------------------------------------------------------------------------
+# helm_list_releases (issue #161)
+# ---------------------------------------------------------------------------
+
+
+async def test_helm_list_releases_renders_release_facts() -> None:
+    from korvid.k8s.helm import HelmReleaseSummary
+
+    class HelmKube:
+        async def list_helm_releases(self, namespace: str | None) -> list[HelmReleaseSummary]:
+            assert namespace == "prod"
+            return [
+                HelmReleaseSummary(
+                    name="web",
+                    namespace="prod",
+                    kind="HelmRelease",
+                    created="",
+                    revision=3,
+                    status="deployed",
+                    chart="web-1.2.3",
+                    app_version="2.7.1",
+                )
+            ]
+
+    ex = ToolExecutor(HelmKube(), {})  # type: ignore[arg-type]  # read-only fake
+    out = await ex.execute("helm_list_releases", {"namespace": "prod"})
+    assert "prod/web" in out
+    assert "revision=3" in out
+    assert "status=deployed" in out
+    assert "chart=web-1.2.3" in out
+    assert "app_version=2.7.1" in out
+
+
+async def test_helm_list_releases_empty_and_error() -> None:
+    class EmptyKube:
+        async def list_helm_releases(self, namespace: str | None) -> list[Any]:
+            return []
+
+    ex = ToolExecutor(EmptyKube(), {})  # type: ignore[arg-type]  # read-only fake
+    assert await ex.execute("helm_list_releases", {}) == "(none)"
+
+    class ExplodingKube:
+        async def list_helm_releases(self, namespace: str | None) -> list[Any]:
+            raise RuntimeError("boom")
+
+    ex = ToolExecutor(ExplodingKube(), {})  # type: ignore[arg-type]  # read-only fake
+    assert (await ex.execute("helm_list_releases", {})).startswith("ERROR:")

tests/tools/test_registry.py

@@ -231,6 +231,7 @@ def test_validate_dispatch_targets_rejects_write_tool_naming_executor_method() -
     "get_logs",
     "get_events",
     "list_operators",
+    "helm_list_releases",
     "diagnose_pod",
 ]
 _UI_ORDER = ["navigate", "set_filter", "open_logs", "open_describe", "drill_down"]