fix(models): lora model revision re-pull logic

services/core/models/src/nmp/core/models/controllers/backends/common.py

@@ -55,6 +55,8 @@ class DeploymentConfigView:
     image_name: Optional[str] = None
     image_tag: Optional[str] = None
     health_check_path: Optional[str] = None
+    run_as_user: Optional[int] = None
+    run_as_group: Optional[int] = None
     additional_envs: Optional[Dict[str, str]] = None
     additional_args: Optional[List[str]] = None
     k8s_nim_operator_config: Optional[K8sNIMOperatorConfig] = None
@@ -78,6 +80,8 @@ def deployment_config_view(config: Optional[_DeploymentConfigLike]) -> Deploymen
         image_name=getattr(executor, "image_name", None),
         image_tag=getattr(executor, "image_tag", None),
         health_check_path=getattr(executor, "health_check_path", None),
+        run_as_user=getattr(executor, "run_as_user", None),
+        run_as_group=getattr(executor, "run_as_group", None),
         additional_envs=getattr(executor, "additional_envs", None),
         additional_args=getattr(executor, "additional_args", None),
         k8s_nim_operator_config=getattr(executor, "k8s_nim_operator_config", None),

services/core/models/src/nmp/core/models/controllers/backends/k8s_nim_operator/reconcilers/k8s.py

@@ -37,6 +37,7 @@
 from nmp.core.models.app.constants import MODEL_MANAGED_BY_LABEL, MODEL_MANAGED_BY_MODELS_CONTROLLER
 from nmp.core.models.controllers.backends import generic_compiler, vllm_compiler
 from nmp.core.models.controllers.backends.backends import DeploymentStatusUpdate
+from nmp.core.models.controllers.backends.common import DeploymentConfigView
 from nmp.core.models.controllers.backends.engine import (
     ENGINE_GENERIC,
     ENGINE_VLLM,
@@ -137,7 +138,7 @@ async def create(self, resolved: ResolvedDeployment) -> DeploymentStatusUpdate:
             if resolved.files_hf_url is None:
                 raise ValueError(f"Cannot create {engine} deployment: Files HF endpoint was not resolved")
 
-            user_id, group_id = self._pod_user(engine)
+            user_id, group_id = self._pod_user(engine, view)
             pvc = vllm_k8s_compiler.compile_pvc(
                 resource_name=resource_name,
                 workspace=deployment.workspace,
@@ -449,16 +450,25 @@ def _project_deployment_readiness(self, resource_name: str) -> DeploymentStatusU
     # Engine-parameterized serving objects (shared by vLLM + generic)
     # ------------------------------------------------------------------
 
-    def _pod_user(self, engine: str) -> tuple[Optional[int], Optional[int]]:
-        """Pod securityContext uid/gid for an engine.
+    def _pod_user(self, engine: str, view: DeploymentConfigView) -> tuple[Optional[int], Optional[int]]:
+        """Pod securityContext uid/gid for a deployment.
 
-        vLLM pins its image's user (2000/0); a generic container runs as its own
-        image's user (unset), since we can't assume an arbitrary image tolerates a
-        forced uid/gid.
+        An explicit ``executor_config.run_as_user`` / ``run_as_group`` always wins
+        (each independently). Otherwise the engine default applies: vLLM pins its
+        image's user (2000/0); a generic container runs as its own image's user
+        (unset), since we can't assume an arbitrary image tolerates a forced
+        uid/gid.
         """
         if engine == ENGINE_VLLM:
-            return self._backend_config.default_vllm_user_id, self._backend_config.default_vllm_group_id
-        return None, None
+            default_user: Optional[int] = self._backend_config.default_vllm_user_id
+            default_group: Optional[int] = self._backend_config.default_vllm_group_id
+        else:
+            default_user = None
+            default_group = None
+
+        user_id = view.run_as_user if view.run_as_user is not None else default_user
+        group_id = view.run_as_group if view.run_as_group is not None else default_group
+        return user_id, group_id
 
     def _serving_spec(
         self,
@@ -478,7 +488,7 @@ def _serving_spec(
         resource_name = resolved.resource_name
         health_path = resolve_health_path(engine, view)
         startup_grace = self._backend_config.default_startup_probe_grace_period_seconds or 600
-        user_id, group_id = self._pod_user(engine)
+        user_id, group_id = self._pod_user(engine, view)
 
         if engine == ENGINE_GENERIC:
             image_name, image_tag = generic_compiler.resolve_generic_image(view)
@@ -718,14 +728,42 @@ def _find_job_pod_name(self, job_name: str) -> str | None:
             return None
 
     def _existing_model_source(self, resource_name: str) -> str | None:
-        """Read the model-source annotation off the existing puller Job, if any."""
+        """Read the recorded model-source for a deployment, from the Job or the PVC.
+
+        The model source is stamped as an annotation on BOTH the puller Job and
+        the PVC at create time. The Job is the primary source while it exists, but
+        it is deleted at P3 (to release the RWO volume) once weights are pulled --
+        so for an already-serving deployment the Job is gone and we must read the
+        annotation from the (long-lived) PVC instead. Without the PVC fallback a
+        later model-revision change would read ``None`` here, skip the re-pull
+        branch, and serve stale weights.
+
+        Returns ``None`` only when neither object carries the annotation (or
+        neither exists). Non-404 API errors are propagated rather than swallowed.
+        """
+        # Primary: the puller Job (present during/just after the pull phase).
         try:
             job = self._batch_v1.read_namespaced_job(
                 name=vllm_k8s_compiler.pull_job_name(resource_name), namespace=self._k8s_namespace
             )
-        except k8s_client.exceptions.ApiException:
-            return None
-        annotations = (job.metadata.annotations or {}) if job.metadata else {}
+            annotations = (job.metadata.annotations or {}) if job.metadata else {}
+            source = annotations.get(vllm_k8s_compiler.MODEL_SOURCE_ANNOTATION)
+            if source:
+                return source
+        except k8s_client.exceptions.ApiException as e:
+            if e.status != 404:
+                raise
+
+        # Fallback: the PVC (survives Job deletion at P3).
+        try:
+            pvc = self._core_v1.read_namespaced_persistent_volume_claim(
+                name=vllm_k8s_compiler.pvc_name(resource_name), namespace=self._k8s_namespace
+            )
+        except k8s_client.exceptions.ApiException as e:
+            if e.status == 404:
+                return None
+            raise
+        annotations = (pvc.metadata.annotations or {}) if pvc.metadata else {}
         return annotations.get(vllm_k8s_compiler.MODEL_SOURCE_ANNOTATION)
 
     def _delete_serving_resources(self, resource_name: str) -> list[str]:

services/core/models/src/nmp/core/models/schemas.py

@@ -1421,6 +1421,25 @@ class ContainerExecutorConfig(BaseModel):
         max_length=constants.MAX_LENGTH_255,
     )
 
+    # Pod securityContext user override (k8s backend only). When unset, the engine
+    # picks an appropriate default (vLLM pins its image's user; generic runs as the
+    # image's own user). Set these to run an arbitrary container as a specific
+    # uid/gid -- e.g. a generic image that requires a particular user. Ignored by
+    # the docker backend, which does not set a container user.
+    run_as_user: Optional[int] = Field(
+        default=None,
+        ge=0,
+        description="Pod securityContext runAsUser (uid) for the serving container (k8s backend only). "
+        "If unset, the engine default applies (vLLM pins its image's user; generic uses the image's "
+        "own user). Ignored by the docker backend.",
+    )
+    run_as_group: Optional[int] = Field(
+        default=None,
+        ge=0,
+        description="Pod securityContext runAsGroup (gid) for the serving container (k8s backend only). "
+        "If unset, the engine default applies. Ignored by the docker backend.",
+    )
+
     # Escape hatches -- for anything not surfaced as a first-class field above
     additional_envs: Optional[Dict[str, str]] = Field(
         default=None, description="Additional environment variables for the deployment"