@@ -1,6 +1,6 @@
 """Generic shell-operator hook utilities shared across all hooks.
 
-Provides binding context I/O and status patching via kubectl.
+Provides binding context I/O and CR status patching.
 """
 
 from __future__ import annotations
@@ -9,10 +9,14 @@
 import json
 import logging
 import os
-import subprocess
 import sys
 from typing import Any
 
+from kubernetes import client as k8s_client
+from kubernetes.client.exceptions import ApiException
+
+from openstack_sync.utils import load_kubernetes_config
+
 LOG = logging.getLogger(__name__)
 
 
@@ -181,27 +185,74 @@ def _status_is_current(
     return True
 
 
+#: Memoised CustomObjectsApi, so one config load serves every patch in a run.
+_custom_objects_api: Any = None
+
+
+def _customobjects_api() -> Any:
+    """Return the shared ``CustomObjectsApi``, configuring the client once."""
+    global _custom_objects_api
+    if _custom_objects_api is None:
+        load_kubernetes_config()
+        _custom_objects_api = k8s_client.CustomObjectsApi()
+    return _custom_objects_api
+
+
+def crd_request_target(crd_api_version: str, crd_resource: str) -> tuple[str, str, str]:
+    """Return ``(group, version, plural)`` for addressing a CR.
+
+    Derived from the two environment values the chart already injects, rather
+    than from new ones: ``<prefix>_CRD_API_VERSION`` is ``<group>/<version>``
+    and ``<prefix>_CRD_RESOURCE`` is ``<plural>.<group>``, which between them
+    carry everything the API needs.
+
+    Raises:
+        ValueError: When either value is not in the expected shape.
+    """
+    group, _, version = crd_api_version.partition("/")
+    plural = crd_resource.partition(".")[0]
+    if not group or not version or not plural:
+        raise ValueError(
+            f"cannot derive a CRD request target from "
+            f"api_version={crd_api_version!r} and resource={crd_resource!r}"
+        )
+    return group, version, plural
+
+
+def _api_error_detail(exc: ApiException, max_body: int = 512) -> str:
+    """Return a one-line description of *exc* for a log message."""
+    body = str(exc.body or "").strip().replace("\n", " ")
+    detail = f"HTTP {exc.status} {exc.reason or ''}".strip()
+    if not body:
+        return detail
+    return f"{detail}: {truncate_message(body, max_body)}"
+
+
 def patch_resource_status(
     *,
     name: str,
     namespace: str | None,
     generation: int | None,
     sync_status: str,
     message: str,
+    crd_api_version: str,
     crd_resource: str,
     crd_kind: str,
     status_enabled: bool,
     current_status: dict[str, Any] | None = None,
 ) -> None:
-    """Patch the status subresource of a CR via kubectl.
+    """Patch the status subresource of a CR.
 
     Args:
         name: CR metadata.name.
-        namespace: CR metadata.namespace (optional).
+        namespace: CR metadata.namespace. Required to address the object; the
+            patch is skipped with a warning when it is absent.
         generation: CR metadata.generation for observedGeneration (optional).
         sync_status: One of ``"Synced"`` or ``"Failed"``.
         message: Human-readable detail for the status message.
-        crd_resource: Fully-qualified CRD resource name for kubectl (e.g.
+        crd_api_version: CRD API version as ``<group>/<version>`` (e.g.
+            ``neutron.understack.rackspace.net/v1alpha1``).
+        crd_resource: Fully-qualified CRD resource name (e.g.
             ``neutronrouterflavors.neutron.understack.rackspace.net``).
         crd_kind: CRD kind used in log messages (e.g. ``NeutronRouterFlavor``).
         status_enabled: When False the function returns immediately.
@@ -231,32 +282,46 @@ def patch_resource_status(
     if generation is not None:
         status["observedGeneration"] = generation
 
-    command = [
-        "kubectl",
-        "patch",
-        crd_resource,
-        name,
-        "--type",
-        "merge",
-        "--subresource",
-        "status",
-        "-p",
-        json.dumps({"status": status}, sort_keys=True),
-    ]
-    if namespace:
-        command.extend(["-n", namespace])
+    if not namespace:
+        LOG.warning(
+            "unable to patch %s status for %s; no namespace to address it in",
+            crd_kind,
+            name,
+        )
+        return
 
     try:
-        result = subprocess.run(  # noqa: S603,S607
-            command,
-            capture_output=True,
-            check=False,
-            text=True,
-        )
-    except FileNotFoundError:
-        LOG.warning("kubectl not found; unable to patch %s status", crd_kind)
+        group, version, plural = crd_request_target(crd_api_version, crd_resource)
+    except ValueError as exc:
+        LOG.warning("unable to patch %s status for %s: %s", crd_kind, name, exc)
         return
 
-    if result.returncode != 0:
-        error = (result.stderr or result.stdout or "unknown error").strip()
-        LOG.warning("failed to patch %s status for %s: %s", crd_kind, name, error)
+    # The client's default content type here is merge-patch+json, which is what
+    # `kubectl patch --type merge --subresource status` sent before this moved
+    # off kubectl, so the request on the wire is unchanged.
+    try:
+        _customobjects_api().patch_namespaced_custom_object_status(
+            group=group,
+            version=version,
+            namespace=namespace,
+            plural=plural,
+            name=name,
+            body={"status": status},
+        )
+    except ApiException as exc:
+        LOG.warning(
+            "failed to patch %s status for %s: %s",
+            crd_kind,
+            name,
+            _api_error_detail(exc),
+        )
+    except Exception as exc:  # noqa: BLE001
+        # Config load and transport failures reach here. A status patch is
+        # reporting, never the work itself, so it must not fail the reconcile.
+        LOG.warning(
+            "failed to patch %s status for %s: %s: %s",
+            crd_kind,
+            name,
+            type(exc).__name__,
+            exc,
+        )
@@ -502,6 +502,7 @@ def _patch_status(
         generation=resource.generation,
         sync_status=sync_status,
         message=message,
+        crd_api_version=config.crd_api_version,
         crd_resource=config.crd_resource,
         crd_kind=config.crd_kind,
         status_enabled=config.status_enabled,
 
@@ -17,12 +17,20 @@
 from openstack.config import loader as os_config_loader
 
 
-def read_secret_key(secret_name: str, secret_key: str, namespace: str) -> str:
-    """Read a single key from a Kubernetes Secret and return its decoded value.
+def load_kubernetes_config() -> None:
+    """Configure the Kubernetes client for wherever this is running.
 
-    Configures the client automatically:
     - Inside a cluster: uses the pod's service-account token.
     - Outside a cluster: falls back to the local kubeconfig (development).
+    """
+    try:
+        k8s_config.load_incluster_config()
+    except k8s_config.ConfigException:
+        k8s_config.load_kube_config()
+
+
+def read_secret_key(secret_name: str, secret_key: str, namespace: str) -> str:
+    """Read a single key from a Kubernetes Secret and return its decoded value.
 
     Args:
         secret_name: Name of the Kubernetes Secret to read.
@@ -35,10 +43,7 @@ def read_secret_key(secret_name: str, secret_key: str, namespace: str) -> str:
     Raises:
         KeyError: When ``secret_key`` is not present in the Secret's data.
     """
-    try:
-        k8s_config.load_incluster_config()
-    except k8s_config.ConfigException:
-        k8s_config.load_kube_config()
+    load_kubernetes_config()
 
     v1 = client.CoreV1Api()
     secret = v1.read_namespaced_secret(name=secret_name, namespace=namespace)
@@ -89,7 +94,7 @@ def get_openstack_connection(secret_name: str, cloud_name: str) -> Any:
     # openstack.connect(cloud=name) resolves the named cloud from files on
     # disk, which fails inside a container with no clouds.yaml present.
     # Instead, parse the YAML from the Secret and inject it directly into
-    # the SDK config loader — no filesystem writes required.
+    # the SDK config loader -- no filesystem writes required.
     loader = os_config_loader.OpenStackConfig(
         config_files=[],
         vendor_files=[],
 
@@ -767,8 +767,8 @@ def test_run_sync_connects_with_each_resources_own_credentials():
 def test_run_sync_forwards_crd_identity_and_current_status_to_the_patch():
     """Forward everything patch_resource_status needs.
 
-    The CRD identity targets kubectl, and the current status decides whether the
-    patch can be skipped.
+    The CRD identity addresses the object on the API, and the current status
+    decides whether the patch can be skipped.
     """
     config = make_hook_config(status_enabled=True)
     plugin = StubPlugin(config)