fix(serve): fix two P0 resource-reuse bugs (instance-type reuse + Bedrock permission fail-open)

sagemaker-serve/src/sagemaker/serve/model_builder.py

@@ -2089,11 +2089,19 @@ def _reused_endpoint_matches_config(
             if not variants:
                 return True
             variant = variants[0]
+            # InstanceType lives on the variant, so check it before the IC
+            # early-return below -- otherwise an IC mismatch is silently reused.
+            if (
+                instance_type
+                and variant.get("InstanceType")
+                and instance_type != variant["InstanceType"]
+            ):
+                return False
             model_name = variant.get("ModelName")
             if not model_name:
-                # IC-based endpoint — can't validate container config from the
-                # variant. Proceed with reuse (the Model was already matched by
-                # tag in _find_reusable_model).
+                # IC-based endpoint: no ModelName to validate container config
+                # against, and instance type is already checked above. Reuse it
+                # (the Model was matched by tag in _find_reusable_model).
                 return True
             model_desc = sagemaker_client.describe_model(ModelName=model_name)
             # Check PrimaryContainer and fallback to Containers list if PrimaryContainer is empty
@@ -2118,9 +2126,6 @@ def _reused_endpoint_matches_config(
         if self.image_uri and container.get("Image") and self.image_uri != container["Image"]:
             return False
 
-        if instance_type and variant.get("InstanceType") and instance_type != variant["InstanceType"]:
-            return False
-
         return True
 
     def _convert_model_data_source_to_local(self, model_data_source):
@@ -5414,16 +5419,21 @@ def deploy(
                 "inference_component_name (IC update). The flag is ignored."
             )
 
-        # Resource reuse is opt-in per call. When requested, reuse an existing
-        # endpoint built from the same model source (with matching config). If
-        # build() already resolved one (build's reuse gate), use that; otherwise
-        # discover it here.
+        # Resource reuse is opt-in per call. build() may have cached a candidate
+        # in _reused_endpoint_name, but without deploy-time context (instance_type
+        # is a deploy() arg), so re-validate the cache before trusting it and fall
+        # back to a fresh discovery on a miss or mismatch.
         if reuse_resources and not is_inference_component_deploy:
-            reusable_endpoint = getattr(
-                self, "_reused_endpoint_name", None
-            ) or self._find_reusable_endpoint(
-                instance_type=instance_type or self.instance_type
-            )
+            requested_instance_type = instance_type or self.instance_type
+            cached_endpoint = getattr(self, "_reused_endpoint_name", None)
+            if cached_endpoint and self._reused_endpoint_matches_config(
+                cached_endpoint, instance_type=requested_instance_type
+            ):
+                reusable_endpoint = cached_endpoint
+            else:
+                reusable_endpoint = self._find_reusable_endpoint(
+                    instance_type=requested_instance_type
+                )
             if reusable_endpoint:
                 if endpoint_name and endpoint_name != reusable_endpoint:
                     logger.warning(

sagemaker-serve/src/sagemaker/serve/model_reuse.py

@@ -18,6 +18,8 @@
 import time
 from typing import Callable, Optional
 
+from botocore.exceptions import ClientError
+
 logger = logging.getLogger(__name__)
 
 MODEL_SOURCE_TAG_KEY = "sagemaker.amazonaws.com/model-source"
@@ -29,6 +31,7 @@
 _ACTIVE_STATUSES = {"Active", "InService"}
 _CREATING_STATUSES = {"Creating"}
 _FAILED_STATUSES = {"Failed"}
+_ACCESS_DENIED_CODE = "AccessDeniedException"
 
 
 def normalize_tag_value(value: str) -> str:
@@ -43,6 +46,32 @@ def normalize_tag_value(value: str) -> str:
     return f"{value[:_TAG_TRUNCATE_PREFIX_LENGTH]}-{hash_suffix}"
 
 
+def _reraise_if_access_denied(error: ClientError, permission: str) -> None:
+    """Surface a denied reuse-discovery call as an actionable PermissionError.
+
+    Called only from ``reuse_resources=True`` discovery paths. A denied
+    discovery call otherwise looks the same as "nothing to reuse", so reuse
+    falls through to creating the resource. But the resource created on a prior
+    run is still there under the same deterministic name, so the create fails
+    with a confusing ``ValidationException: ... already exists`` that never
+    mentions the real cause. Re-raising here names the missing IAM permission
+    instead.
+
+    Args:
+        error: The ClientError raised by a discovery call.
+        permission: The IAM action reuse discovery requires (named in the message).
+
+    Raises:
+        PermissionError: If ``error`` is an ``AccessDeniedException``.
+    """
+    if error.response.get("Error", {}).get("Code") == _ACCESS_DENIED_CODE:
+        raise PermissionError(
+            f"reuse_resources=True requires the '{permission}' permission to discover "
+            f"reusable resources. Add it to the execution role's policy, or set "
+            f"reuse_resources=False to always create new resources."
+        ) from error
+
+
 def find_existing_bedrock_model(
     bedrock_client,
     source_id: str,
@@ -70,6 +99,10 @@ def find_existing_bedrock_model(
     tag_value = normalize_tag_value(source_id)
     try:
         resource_arn = _find_bedrock_model_arn_by_tag(bedrock_client, tag_value)
+    except ClientError as e:
+        _reraise_if_access_denied(e, "bedrock:ListTagsForResource")
+        logger.warning("Could not list Bedrock custom models: %s. Proceeding without.", e)
+        return None
     except Exception as e:
         logger.warning("Could not list Bedrock custom models: %s. Proceeding without.", e)
         return None
@@ -105,6 +138,12 @@ def find_active_bedrock_deployment_for_model(bedrock_client, model_arn: str) ->
             next_token = response.get("nextToken")
             if not next_token:
                 return None
+    except ClientError as e:
+        _reraise_if_access_denied(e, "bedrock:ListCustomModelDeployments")
+        logger.warning(
+            "Could not list Bedrock custom model deployments: %s. Proceeding without.", e
+        )
+        return None
     except Exception as e:
         logger.warning(
             "Could not list Bedrock custom model deployments: %s. Proceeding without.", e
@@ -139,6 +178,10 @@ def find_existing_sagemaker_endpoint(
     tag_value = normalize_tag_value(source_id)
     try:
         resource_arn = _find_sagemaker_endpoint_arn_by_tag(sagemaker_client, tag_value)
+    except ClientError as e:
+        _reraise_if_access_denied(e, "sagemaker:ListTags")
+        logger.warning("Could not list SageMaker endpoints: %s. Proceeding without.", e)
+        return None
     except Exception as e:
         logger.warning("Could not list SageMaker endpoints: %s. Proceeding without.", e)
         return None

sagemaker-serve/tests/unit/test_model_builder.py

@@ -1220,6 +1220,87 @@ def test_reuse_does_not_intercept_inference_component_deploy(
         mock_endpoint_get.assert_not_called()
         mock_deploy.assert_called_once()
 
+    def _stub_endpoint_config_client(self, instance_type):
+        # Describe an endpoint whose production variant runs on `instance_type`.
+        client = self.mock_session.sagemaker_client
+        client.describe_endpoint.return_value = {"EndpointConfigName": "cfg"}
+        client.describe_endpoint_config.return_value = {
+            "ProductionVariants": [{"ModelName": "m", "InstanceType": instance_type}]
+        }
+        client.describe_model.return_value = {
+            "PrimaryContainer": {"Environment": {}, "Image": "img:1"}
+        }
+        return client
+
+    @patch("sagemaker.serve.model_builder.Endpoint.get")
+    @patch("sagemaker.serve.model_builder.find_existing_sagemaker_endpoint")
+    def test_deploy_ignores_cached_endpoint_when_instance_type_mismatches(
+        self, mock_find, mock_endpoint_get
+    ):
+        # A build-time cached endpoint must be re-validated against the deploy-time
+        # instance_type, not returned blindly. Cache is g5.xlarge; deploy asks p4d.
+        self._stub_endpoint_config_client(instance_type="ml.g5.xlarge")
+        mock_find.return_value = None
+
+        with (
+            patch.object(ModelBuilder, "_resolve_model_source_id", return_value="s3://bucket/model"),
+            patch.object(ModelBuilder, "_is_model_customization", return_value=False),
+            patch.object(ModelBuilder, "_get_deploy_wrapper") as mock_get_wrapper,
+            patch.object(ModelBuilder, "add_tags"),
+        ):
+            mock_deploy = Mock(return_value=Mock())
+            mock_get_wrapper.return_value = mock_deploy
+
+            builder = self._make_builder()
+            builder.built_model = Mock()
+            builder.region = "us-west-2"
+            builder._reused_endpoint_name = "cached-ep"
+
+            builder.deploy(
+                endpoint_name="new-ep",
+                instance_type="ml.p4d.24xlarge",
+                reuse_resources=True,
+            )
+
+        # Instance-type mismatch: fall through and create a new endpoint.
+        mock_endpoint_get.assert_not_called()
+        mock_deploy.assert_called_once()
+
+    @patch("sagemaker.serve.model_builder.Endpoint.get")
+    @patch("sagemaker.serve.model_builder.find_existing_sagemaker_endpoint")
+    def test_deploy_reuses_cached_endpoint_when_instance_type_matches(
+        self, mock_find, mock_endpoint_get
+    ):
+        # A matching instance_type still reuses the cached endpoint (no over-correction).
+        self._stub_endpoint_config_client(instance_type="ml.p4d.24xlarge")
+        mock_endpoint = Mock()
+        mock_endpoint_get.return_value = mock_endpoint
+
+        with (
+            patch.object(ModelBuilder, "_resolve_model_source_id", return_value="s3://bucket/model"),
+            patch.object(ModelBuilder, "_is_model_customization", return_value=False),
+            patch.object(ModelBuilder, "_get_deploy_wrapper") as mock_get_wrapper,
+            patch.object(ModelBuilder, "add_tags"),
+        ):
+            mock_deploy = Mock(return_value=Mock())
+            mock_get_wrapper.return_value = mock_deploy
+
+            builder = self._make_builder()
+            builder.built_model = Mock()
+            builder.region = "us-west-2"
+            builder._reused_endpoint_name = "cached-ep"
+
+            result = builder.deploy(
+                endpoint_name="new-ep",
+                instance_type="ml.p4d.24xlarge",
+                reuse_resources=True,
+            )
+
+        # Cache hit: endpoint reused, not recreated, and no fresh discovery scan.
+        mock_deploy.assert_not_called()
+        mock_find.assert_not_called()
+        assert result == mock_endpoint
+
 
 class TestReusedEndpointMatchesConfig(unittest.TestCase):
     """Tests for ModelBuilder._reused_endpoint_matches_config."""
@@ -1300,3 +1381,37 @@ def test_matches_when_describe_fails(self):
         client.describe_endpoint.side_effect = Exception("boom")
         builder.sagemaker_session.sagemaker_client = client
         assert builder._reused_endpoint_matches_config("ep") is True
+
+    def _stub_ic_sagemaker_client(self, builder, instance_type):
+        # IC-based endpoints carry no ModelName on the variant, only InstanceType.
+        client = Mock()
+        client.describe_endpoint.return_value = {"EndpointConfigName": "cfg"}
+        client.describe_endpoint_config.return_value = {
+            "ProductionVariants": [{"InstanceType": instance_type}]
+        }
+        builder.sagemaker_session.sagemaker_client = client
+        return client
+
+    def test_ic_endpoint_mismatch_on_instance_type(self):
+        # Regression: the IC early-return previously skipped the instance_type check.
+        builder = self._make_builder()
+        client = self._stub_ic_sagemaker_client(builder, instance_type="ml.g5.xlarge")
+        assert (
+            builder._reused_endpoint_matches_config("ep", instance_type="ml.p4d.24xlarge") is False
+        )
+        # Caught from the variant alone, without describing a model.
+        client.describe_model.assert_not_called()
+
+    def test_ic_endpoint_matches_on_instance_type(self):
+        # An IC endpoint whose instance_type matches is still reusable.
+        builder = self._make_builder()
+        self._stub_ic_sagemaker_client(builder, instance_type="ml.p4d.24xlarge")
+        assert (
+            builder._reused_endpoint_matches_config("ep", instance_type="ml.p4d.24xlarge") is True
+        )
+
+    def test_ic_endpoint_reusable_when_no_instance_type_requested(self):
+        # No instance_type requested: an IC endpoint stays reusable.
+        builder = self._make_builder()
+        self._stub_ic_sagemaker_client(builder, instance_type="ml.g5.xlarge")
+        assert builder._reused_endpoint_matches_config("ep") is True

sagemaker-serve/tests/unit/test_model_reuse.py

@@ -16,9 +16,12 @@
 import pytest
 from unittest.mock import Mock, patch, call
 
+from botocore.exceptions import ClientError
+
 from sagemaker.serve.model_reuse import (
     MODEL_SOURCE_TAG_KEY,
     normalize_tag_value,
+    find_active_bedrock_deployment_for_model,
     find_existing_bedrock_model,
     find_existing_sagemaker_endpoint,
     build_source_tag,
@@ -28,6 +31,22 @@
 )
 
 
+def _access_denied_error(operation, action):
+    """Build a ClientError mirroring a real AWS AccessDeniedException (403)."""
+    return ClientError(
+        {
+            "Error": {
+                "Code": "AccessDeniedException",
+                "Message": (
+                    f"User is not authorized to perform: {action} because no "
+                    f"identity-based policy allows the {action} action"
+                ),
+            }
+        },
+        operation,
+    )
+
+
 @pytest.fixture
 def boto_session():
     return Mock()
@@ -193,6 +212,40 @@ def test_find_existing_bedrock_model_returns_none_when_no_models(boto_session, b
     assert result is None
 
 
+def test_find_existing_bedrock_model_raises_on_access_denied(boto_session, bedrock_client):
+    # A denied tag read must surface the missing permission, not fall through to None.
+    bedrock_client.list_custom_models.return_value = {"modelSummaries": [{"modelArn": SAMPLE_ARN}]}
+    bedrock_client.list_tags_for_resource.side_effect = _access_denied_error(
+        "ListTagsForResource", "bedrock:ListTagsForResource"
+    )
+
+    with pytest.raises(PermissionError, match="bedrock:ListTagsForResource"):
+        find_existing_bedrock_model(bedrock_client, "source-id")
+
+
+def test_find_existing_bedrock_model_still_fails_open_on_non_access_error(
+    boto_session, bedrock_client
+):
+    # A non-authorization ClientError (e.g. throttling) still degrades to None.
+    bedrock_client.list_custom_models.side_effect = ClientError(
+        {"Error": {"Code": "ThrottlingException", "Message": "Rate exceeded"}},
+        "ListCustomModels",
+    )
+
+    result = find_existing_bedrock_model(bedrock_client, "source-id")
+
+    assert result is None
+
+
+def test_find_active_bedrock_deployment_raises_on_access_denied(boto_session, bedrock_client):
+    bedrock_client.list_custom_model_deployments.side_effect = _access_denied_error(
+        "ListCustomModelDeployments", "bedrock:ListCustomModelDeployments"
+    )
+
+    with pytest.raises(PermissionError, match="bedrock:ListCustomModelDeployments"):
+        find_active_bedrock_deployment_for_model(bedrock_client, SAMPLE_ARN)
+
+
 def test_find_existing_sagemaker_endpoint_returns_arn_when_in_service(boto_session, sagemaker_client):
     _sagemaker_with_tagged_endpoint(sagemaker_client, ENDPOINT_ARN, "source-id")
     sagemaker_client.describe_endpoint.return_value = {"EndpointStatus": "InService"}
@@ -246,6 +299,16 @@ def test_find_existing_sagemaker_endpoint_returns_none_when_no_endpoints(boto_se
     assert result is None
 
 
+def test_find_existing_sagemaker_endpoint_raises_on_access_denied(boto_session, sagemaker_client):
+    sagemaker_client.list_endpoints.return_value = {"Endpoints": [{"EndpointArn": ENDPOINT_ARN}]}
+    sagemaker_client.list_tags.side_effect = _access_denied_error(
+        "ListTags", "sagemaker:ListTags"
+    )
+
+    with pytest.raises(PermissionError, match="sagemaker:ListTags"):
+        find_existing_sagemaker_endpoint(sagemaker_client, "source-id")
+
+
 def test_build_source_tag_returns_correct_dict():
     source_id = "s3://bucket/path/to/model"
     tag = build_source_tag(source_id)