fix: list-only kinds (OLM packagemanifests) poll instead of dying in the watch loop (#141)

src/korvid/core/watch.py

@@ -31,6 +31,13 @@ def _is_forbidden(exc: Exception) -> bool:
     return isinstance(exc, ApiStatusError) and exc.status == 403
 
 
+def _is_method_not_allowed(exc: Exception) -> bool:
+    """405: the endpoint exists but will never serve this verb — as
+    deterministic as a 403 (issue #141), so retries gain nothing. Unlike a
+    403 the rows a successful LIST already delivered stay valid."""
+    return isinstance(exc, ApiStatusError) and exc.status == 405
+
+
 class WatchManager:
     def __init__(
         self,
@@ -108,7 +115,7 @@ async def _watch_loop(self, kind: str, scope: str) -> Exception | None:
             except asyncio.CancelledError:
                 raise
             except Exception as exc:  # report + retry, never die silently
-                if _is_forbidden(exc):
+                if _is_forbidden(exc) or _is_method_not_allowed(exc):
                     return exc
                 failures += 1
                 logger.exception(

src/korvid/k8s/client.py

@@ -44,6 +44,11 @@
 
 logger = logging.getLogger(__name__)
 
+#: Re-LIST cadence for kinds without a watch endpoint (OLM's packageserver,
+#: issue #141): catalog-ish content changes rarely, so a slow poll keeps the
+#: view fresh without hammering an aggregated API.
+LIST_POLL_INTERVAL = 30.0
+
 
 def _path_segment(value: str) -> str:
     """Percent-encode *value* for safe use as a single URL path segment.
@@ -419,6 +424,13 @@ async def _watch_pods_cluster(self) -> AsyncIterator[tuple[str, PodSummary]]:
         except k8s_client.exceptions.ApiException as exc:
             raise ApiStatusError(int(exc.status or 0), str(exc.reason or "")) from exc
 
+    def _list_path(self, meta: ResourceMeta, namespace: str | None) -> str:
+        """LIST/WATCH path for a kind; cluster-scoped kinds have no
+        namespaced path regardless of scope."""
+        if namespace is not None and meta.namespaced:
+            return f"{meta.api_base}/namespaces/{_path_segment(namespace)}/{meta.plural}"
+        return f"{meta.api_base}/{meta.plural}"
+
     async def watch_objects(
         self, meta: ResourceMeta, namespace: str | None
     ) -> AsyncIterator[tuple[str, GenericSummary]]:
@@ -427,22 +439,31 @@ async def watch_objects(
         Contract mirrors watch_pods: pre-existing items are yielded as ADDED
         first, then live watch events from the snapshot resourceVersion.
         ApiException is wrapped as ApiStatusError at both the LIST and watch phases.
+
+        Kinds whose server offers no watch (``meta.watchable`` False, or a
+        server that advertises watch and then rejects it with 405 - OLM's
+        packageserver, issue #141) degrade to periodic re-LIST diffing: the
+        stream stays alive and incremental, so the view keeps rendering
+        without the clear/retry/die loop.
         """
         if self._api is None:
             raise RuntimeError("connect() first")
 
         # LIST phase --------------------------------------------------------
-        # Cluster-scoped kinds have no namespaced path regardless of scope.
-        if namespace is not None and meta.namespaced:
-            list_path = f"{meta.api_base}/namespaces/{_path_segment(namespace)}/{meta.plural}"
-        else:
-            list_path = f"{meta.api_base}/{meta.plural}"
-
+        list_path = self._list_path(meta, namespace)
         data = await self._request_json(list_path)
 
         resource_version: str | None = (data.get("metadata") or {}).get("resourceVersion")
+        known: dict[str, GenericSummary] = {}
         for item in data.get("items", []):
-            yield ("ADDED", self._object_summary(meta, item))
+            summary = self._object_summary(meta, item)
+            known[f"{summary.namespace}/{summary.name}"] = summary
+            yield ("ADDED", summary)
+
+        if not meta.watchable:
+            async for event in self._poll_objects(meta, list_path, known):
+                yield event
+            return
 
         # Watch phase -------------------------------------------------------
         watch_kwargs: dict[str, Any] = {}
@@ -459,8 +480,49 @@ async def watch_objects(
                         str(event["type"]),
                         self._object_summary(meta, event["raw_object"]),
                     )
-        except k8s_client.exceptions.ApiException as exc:
-            raise ApiStatusError(int(exc.status or 0), str(exc.reason or "")) from exc
+        except (k8s_client.exceptions.ApiException, ApiStatusError) as exc:
+            # The raw-watch adapter surfaces HTTP errors as ApiStatusError
+            # (via _raise_for_status); the kubernetes client's own paths
+            # raise ApiException - both carry .status/.reason, and the 405
+            # fallback must catch both.
+            status = int(getattr(exc, "status", 0) or 0)
+            if status != 405:
+                # Preserve .body: same-status disambiguation (PDB denial vs
+                # APF throttling) depends on it; ApiException carries one too.
+                raise ApiStatusError(
+                    status,
+                    str(getattr(exc, "reason", "") or ""),
+                    str(getattr(exc, "body", "") or ""),
+                ) from exc
+            # Discovery advertised watch but the server refuses it: as
+            # deterministic as it gets - poll instead of letting the
+            # manager burn retries clearing and re-seeding the store.
+            logger.info("%s rejects watch (405); falling back to LIST polling", meta.plural)
+            async for event in self._poll_objects(meta, list_path, known):
+                yield event
+
+    async def _poll_objects(
+        self, meta: ResourceMeta, list_path: str, known: dict[str, GenericSummary]
+    ) -> AsyncIterator[tuple[str, GenericSummary]]:
+        """Endless re-LIST diff stream for kinds without a watch endpoint.
+
+        Each round upserts every present row (ADDED doubles as MODIFIED in
+        the store) and emits DELETED for rows that vanished since the last
+        round, so the table stays incremental - never cleared. *known* is
+        seeded with the initial LIST's rows.
+        """
+        while True:
+            await asyncio.sleep(LIST_POLL_INTERVAL)
+            data = await self._request_json(list_path)
+            current: dict[str, GenericSummary] = {}
+            for item in data.get("items", []):
+                summary = self._object_summary(meta, item)
+                current[f"{summary.namespace}/{summary.name}"] = summary
+                yield ("ADDED", summary)
+            for key, old in known.items():
+                if key not in current:
+                    yield ("DELETED", old)
+            known = current
 
     async def list_objects(self, meta: ResourceMeta, namespace: str | None) -> list[GenericSummary]:
         """LIST any resource kind and return GenericSummary items.
@@ -470,11 +532,7 @@ async def list_objects(self, meta: ResourceMeta, namespace: str | None) -> list[
         """
         if self._api is None:
             raise RuntimeError("connect() first")
-        if namespace is not None and meta.namespaced:
-            list_path = f"{meta.api_base}/namespaces/{_path_segment(namespace)}/{meta.plural}"
-        else:
-            list_path = f"{meta.api_base}/{meta.plural}"
-        data = await self._request_json(list_path)
+        data = await self._request_json(self._list_path(meta, namespace))
         return [self._object_summary(meta, item) for item in data.get("items", [])]
 
     async def get_object(
@@ -1384,7 +1442,11 @@ async def _request_json(
             raise ApiStatusError(int(exc.status or 0), str(exc.reason or "")) from exc
 
     async def discover_resources(self) -> list[ResourceMeta]:
-        """Return every list+watch-able resource from /api/v1 and /apis.
+        """Return every LIST-able resource from /api/v1 and /apis.
+
+        Kinds without a watch verb (aggregated APIs like OLM's
+        packageserver) are included with ``watchable=False`` — the watch
+        source keeps them fresh by polling (issue #141).
 
         Group version lists are fetched concurrently — sequential fetching adds
         one RTT per API group and dominates startup on clusters with many CRDs.
@@ -1457,7 +1519,7 @@ def _parse_resource_list(data: dict[str, Any], *, group: str, version: str) -> l
         verbs: list[str] = r.get("verbs", [])
         if not isinstance(name, str) or not isinstance(kind, str) or namespaced is None:
             continue  # malformed entry must not kill discovery
-        if "/" in name or "list" not in verbs or "watch" not in verbs:
+        if "/" in name or "list" not in verbs:
             continue
         out.append(
             ResourceMeta(
@@ -1467,6 +1529,9 @@ def _parse_resource_list(data: dict[str, Any], *, group: str, version: str) -> l
                 version,
                 bool(namespaced),
                 tuple(r.get("shortNames") or ()),
+                # list-only aggregated APIs (OLM's packageserver) stay
+                # discoverable; the watch source polls them (issue #141).
+                watchable="watch" in verbs,
             )
         )
     return out

src/korvid/k8s/discovery.py

@@ -20,6 +20,10 @@ class ResourceMeta:
     #: (e.g. the helm browser reads Secrets). Permission probes target this;
     #: None means the view has no backing API resource to probe.
     backing: tuple[str, str] | None = None
+    #: False for kinds whose server offers `list` but not `watch` (aggregated
+    #: APIs like OLM's packageserver, issue #141): the watch source keeps
+    #: them fresh by periodic re-LIST diffing instead of a watch stream.
+    watchable: bool = True
 
     @property
     def api_base(self) -> str:

tests/core/test_watch.py

@@ -280,6 +280,28 @@ def _ns_pod(name: str, ns: str) -> PodSummary:
     return PodSummary(name=name, namespace=ns, phase="Running", ready="1/1", restarts=0, node=None)
 
 
+async def test_405_reports_once_without_retries_and_keeps_listed_rows() -> None:
+    """405 Method Not Allowed is as deterministic as 403 (the server will
+    never start supporting watch mid-retry) - but unlike a 403, the rows the
+    successful LIST already delivered stay visible (issue #141)."""
+    store = ResourceStore()
+    errors: list[str] = []
+    attempts: list[int] = []
+
+    async def source(kind: str, scope: str) -> AsyncIterator[tuple[str, Summary]]:
+        attempts.append(1)
+        yield ("ADDED", _ns_pod("listed", "olm"))
+        raise ApiStatusError(405, "Method Not Allowed")
+
+    mgr = WatchManager(store, source, on_error=errors.append, retry_delay=0, max_retries=5)
+    await mgr.start("packagemanifests", "olm")
+    await asyncio.sleep(0.05)
+    assert len(attempts) == 1  # deterministic: no retry burn
+    assert len(errors) == 1
+    # the LISTed rows survive - only authorization denials purge the bucket
+    assert [s.name for s in store.get("packagemanifests", "olm")] == ["listed"]
+
+
 async def test_cluster_scope_403_reports_once_without_retries() -> None:
     """A Forbidden cluster-scope watch is deterministic: one attempt, one
     report, no retry loop and no per-namespace watch tasks."""

tests/k8s/test_client.py

@@ -327,6 +327,152 @@ async def test_watch_objects_cluster_scoped_kind_ignores_namespace() -> None:
     assert called_path == "/api/v1/nodes"
 
 
+# ---------------------------------------------------------------------------
+# watch_objects — list-only kinds poll instead of watching (issue #141)
+# ---------------------------------------------------------------------------
+
+
+def _pkg_meta() -> ResourceMeta:
+    return ResourceMeta(
+        "PackageManifest",
+        "packagemanifests",
+        "packages.operators.coreos.com",
+        "v1",
+        True,
+        watchable=False,
+    )
+
+
+async def _take(gen: Any, n: int) -> list[tuple[str, str]]:
+    """First *n* (event, name) pairs from an endless watch generator."""
+    out: list[tuple[str, str]] = []
+    async for ev, s in gen:
+        out.append((ev, s.name))
+        if len(out) >= n:
+            break
+    return out
+
+
+async def test_unwatchable_kind_polls_lists_and_diffs_instead_of_watching() -> None:
+    """A kind discovered without the watch verb (OLM packageserver) must be
+    kept fresh by periodic re-LIST diffing: upserts for present rows, a
+    DELETED for vanished ones - and the Watch API is never touched."""
+    client = KubeClient()
+    meta = _pkg_meta()
+    snapshots = [
+        {"metadata": {}, "items": [_generic("etcd"), _generic("kafka")]},
+        {"metadata": {}, "items": [_generic("etcd"), _generic("postgres")]},
+    ]
+    request_json_mock = AsyncMock(side_effect=snapshots)
+    watch_factory = MagicMock()
+
+    with (
+        patch.object(client, "_api", MagicMock()),
+        patch.object(client, "_request_json", request_json_mock),
+        patch.object(client_mod, "LIST_POLL_INTERVAL", 0.0),
+        patch("korvid.k8s.client.k8s_watch.Watch", watch_factory),
+    ):
+        events = await _take(client.watch_objects(meta, "olm"), 5)
+
+    assert events[:2] == [("ADDED", "etcd"), ("ADDED", "kafka")]
+    # Second LIST round: upserts for present rows, DELETED for the vanished one.
+    assert ("ADDED", "postgres") in events[2:]
+    assert ("DELETED", "kafka") in events[2:]
+    watch_factory.assert_not_called()
+
+
+async def test_watch_405_falls_back_to_list_polling() -> None:
+    """A server that advertises watch but rejects it with 405 (aggregated
+    API drift) degrades to polling instead of dying in the retry loop.
+    The raw-watch adapter surfaces the refusal as ApiStatusError (via
+    _raise_for_status), so that exact type must be caught."""
+    client = KubeClient()
+    meta = _deploy_meta()
+    snapshots = [
+        {"metadata": {"resourceVersion": "1"}, "items": [_generic("dep-a")]},
+        {"metadata": {}, "items": [_generic("dep-a"), _generic("dep-b")]},
+    ]
+    request_json_mock = AsyncMock(side_effect=snapshots)
+    fake_watch = _FakeWatch([], raise_at=0, raise_exc=ApiStatusError(405, "Method Not Allowed"))
+
+    with (
+        patch.object(client, "_api", MagicMock()),
+        patch.object(client, "_request_json", request_json_mock),
+        patch.object(client_mod, "LIST_POLL_INTERVAL", 0.0),
+        patch("korvid.k8s.client.k8s_watch.Watch", return_value=fake_watch),
+    ):
+        events = await _take(client.watch_objects(meta, "default"), 3)
+
+    assert events[0] == ("ADDED", "dep-a")
+    assert ("ADDED", "dep-b") in events[1:]
+
+
+async def test_watch_405_api_exception_also_falls_back_to_polling() -> None:
+    """The kubernetes client's own ApiException(405) takes the same
+    fallback (both exception types cross the watch stream)."""
+    client = KubeClient()
+    meta = _deploy_meta()
+    snapshots = [
+        {"metadata": {"resourceVersion": "1"}, "items": [_generic("dep-a")]},
+        {"metadata": {}, "items": [_generic("dep-a"), _generic("dep-b")]},
+    ]
+    request_json_mock = AsyncMock(side_effect=snapshots)
+    fake_watch = _FakeWatch(
+        [], raise_at=0, raise_exc=ApiException(status=405, reason="Method Not Allowed")
+    )
+
+    with (
+        patch.object(client, "_api", MagicMock()),
+        patch.object(client, "_request_json", request_json_mock),
+        patch.object(client_mod, "LIST_POLL_INTERVAL", 0.0),
+        patch("korvid.k8s.client.k8s_watch.Watch", return_value=fake_watch),
+    ):
+        events = await _take(client.watch_objects(meta, "default"), 3)
+
+    assert events[0] == ("ADDED", "dep-a")
+    assert ("ADDED", "dep-b") in events[1:]
+
+
+async def test_watch_non_405_api_status_error_still_raises() -> None:
+    """A non-405 ApiStatusError from the raw adapter propagates with its
+    status, reason and body intact - the body disambiguates same-status
+    responses (PDB denial vs APF throttling)."""
+    client = KubeClient()
+    meta = _deploy_meta()
+    list_resp: dict[str, Any] = {"metadata": {"resourceVersion": "9"}, "items": []}
+    fake_watch = _FakeWatch(
+        [], raise_at=0, raise_exc=ApiStatusError(410, "Gone", '{"kind":"Status"}')
+    )
+
+    with (
+        patch.object(client, "_api", MagicMock()),
+        patch.object(client, "_request_json", AsyncMock(return_value=list_resp)),
+        patch("korvid.k8s.client.k8s_watch.Watch", return_value=fake_watch),
+        pytest.raises(ApiStatusError, match="Gone") as excinfo,
+    ):
+        async for _ in client.watch_objects(meta, "default"):
+            pass
+    assert excinfo.value.body == '{"kind":"Status"}'
+
+
+async def test_watch_non_405_api_exception_still_raises() -> None:
+    """Only the deterministic 405 falls back to polling: other watch errors
+    keep propagating so the WatchManager's retry/report loop stays in charge."""
+    client = KubeClient()
+    meta = _deploy_meta()
+    list_resp: dict[str, Any] = {"metadata": {"resourceVersion": "9"}, "items": []}
+    fake_watch = _FakeWatch([], raise_at=0, raise_exc=ApiException(status=500, reason="boom"))
+
+    with (
+        patch.object(client, "_api", MagicMock()),
+        patch.object(client, "_request_json", AsyncMock(return_value=list_resp)),
+        patch("korvid.k8s.client.k8s_watch.Watch", return_value=fake_watch),
+        pytest.raises(ApiStatusError, match="boom"),
+    ):
+        async for _ in client.watch_objects(meta, "default"):
+            pass
+
+
 # ---------------------------------------------------------------------------
 # get_object
 # ---------------------------------------------------------------------------