feat(bedrock): Add deployment status polling after CreateCustomModelDeployment

Previously create_deployment() only polled for the custom model
to reach Active status before calling CreateCustomModelDeployment,
but did not wait for the deployment itself to become Active. This
caused callers to receive a deployment ARN that was still in
Creating state, requiring manual polling in user code.

Add _wait_for_deployment_active() that polls
get_custom_model_deployment until status reaches Active, raises
RuntimeError on Failed, and times out after max_wait seconds
(default 3600s, poll interval 30s).

Wire it into create_deployment() so the full flow is now:
1. _wait_for_model_active (poll model creation)
2. create_custom_model_deployment (API call)
3. _wait_for_deployment_active (poll deployment creation)

Gracefully skips deployment polling if the API response does not
contain a customModelDeploymentArn.

Unit tests (48 passing):
- _wait_for_deployment_active: immediate Active, polling,
  Failed status, timeout
- create_deployment: full model+deployment polling chain,
  skip polling when no ARN in response
- deploy Nova chain: updated to verify deployment polling

Author: nargokul

sagemaker-serve/src/sagemaker/serve/bedrock_model_builder.py

@@ -193,20 +193,21 @@ def create_deployment(
         """Create a deployment for a Nova custom model.
 
         Polls the model status until it becomes Active before creating the deployment,
-        since Bedrock requires the custom model to be ready before deployment.
+        then polls the deployment status until it becomes Active.
 
         Args:
             model_arn: ARN of the custom model to deploy.
             deployment_name: Name for the deployment.
-            poll_interval: Seconds between status checks. Defaults to 60.
-            max_wait: Maximum seconds to wait for model to become Active. Defaults to 3600.
+            poll_interval: Seconds between status checks. Defaults to 60 for model,
+                30 for deployment.
+            max_wait: Maximum seconds to wait per polling phase. Defaults to 3600.
             **kwargs: Additional parameters for create_custom_model_deployment.
 
         Returns:
             Response from Bedrock create_custom_model_deployment API.
 
         Raises:
-            RuntimeError: If the model fails or times out waiting to become Active.
+            RuntimeError: If the model or deployment fails or times out.
             ValueError: If model_arn is not provided.
         """
         if not model_arn:
@@ -222,7 +223,15 @@ def create_deployment(
         params = {k: v for k, v in params.items() if v is not None}
 
         logger.info("Creating deployment %s for model %s", deployment_name, model_arn)
-        return self._get_bedrock_client().create_custom_model_deployment(**params)
+        response = self._get_bedrock_client().create_custom_model_deployment(**params)
+
+        deployment_arn = response.get("customModelDeploymentArn")
+        if deployment_arn:
+            self._wait_for_deployment_active(
+                deployment_arn, poll_interval=poll_interval, max_wait=max_wait
+            )
+
+        return response
 
     def _wait_for_model_active(
         self, model_arn: str, poll_interval: int = 60, max_wait: int = 3600
@@ -256,6 +265,40 @@ def _wait_for_model_active(
             f"Last status: {status}"
         )
 
+    def _wait_for_deployment_active(
+        self, deployment_arn: str, poll_interval: int = 30, max_wait: int = 3600
+    ):
+        """Poll Bedrock until the custom model deployment reaches Active status.
+
+        Args:
+            deployment_arn: ARN of the custom model deployment.
+            poll_interval: Seconds between status checks. Defaults to 30.
+            max_wait: Maximum seconds to wait. Defaults to 3600.
+
+        Raises:
+            RuntimeError: If the deployment status is Failed or the wait times out.
+        """
+        elapsed = 0
+        status = None
+        while elapsed < max_wait:
+            resp = self._get_bedrock_client().get_custom_model_deployment(
+                customModelDeploymentIdentifier=deployment_arn
+            )
+            status = resp.get("status")
+            logger.info("Deployment status: %s (elapsed %ds)", status, elapsed)
+            if status == "Active":
+                return
+            if status == "Failed":
+                raise RuntimeError(
+                    f"Deployment {deployment_arn} failed."
+                )
+            time.sleep(poll_interval)
+            elapsed += poll_interval
+        raise RuntimeError(
+            f"Timed out after {max_wait}s waiting for deployment {deployment_arn} to become Active. "
+            f"Last status: {status}"
+        )
+
     def _fetch_model_package(self) -> Optional[ModelPackage]:
         """Fetch the ModelPackage from the provided model.
 

sagemaker-serve/tests/unit/test_bedrock_model_builder.py

@@ -377,27 +377,44 @@ def test_timeout_raises(self):
 
 
 class TestCreateDeployment:
-    def test_polls_then_creates(self):
+    def test_polls_model_then_creates_then_polls_deployment(self):
         b = _builder()
         b._bedrock_client = Mock()
         b._bedrock_client.get_custom_model.return_value = {"modelStatus": "Active"}
         b._bedrock_client.create_custom_model_deployment.return_value = {
             "customModelDeploymentArn": "arn:dep"
         }
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Active"}
+
         result = b.create_deployment(model_arn="arn:model", deployment_name="dep")
+
         b._bedrock_client.get_custom_model.assert_called_once()
         b._bedrock_client.create_custom_model_deployment.assert_called_once()
+        b._bedrock_client.get_custom_model_deployment.assert_called_once()
         assert result["customModelDeploymentArn"] == "arn:dep"
 
     def test_passes_extra_kwargs(self):
         b = _builder()
         b._bedrock_client = Mock()
         b._bedrock_client.get_custom_model.return_value = {"modelStatus": "Active"}
-        b._bedrock_client.create_custom_model_deployment.return_value = {}
+        b._bedrock_client.create_custom_model_deployment.return_value = {
+            "customModelDeploymentArn": "arn:dep"
+        }
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Active"}
+
         b.create_deployment(model_arn="arn:model", deployment_name="d", commitmentDuration="ONE_MONTH")
         kw = b._bedrock_client.create_custom_model_deployment.call_args[1]
         assert kw["commitmentDuration"] == "ONE_MONTH"
 
+    def test_skips_deployment_polling_when_no_arn_in_response(self):
+        b = _builder()
+        b._bedrock_client = Mock()
+        b._bedrock_client.get_custom_model.return_value = {"modelStatus": "Active"}
+        b._bedrock_client.create_custom_model_deployment.return_value = {}
+
+        b.create_deployment(model_arn="arn:model", deployment_name="d")
+        b._bedrock_client.get_custom_model_deployment.assert_not_called()
+
     def test_empty_model_arn_raises(self):
         with pytest.raises(ValueError, match="model_arn is required"):
             _builder().create_deployment(model_arn="", deployment_name="d")
@@ -407,6 +424,47 @@ def test_none_model_arn_raises(self):
             _builder().create_deployment(model_arn=None, deployment_name="d")
 
 
+# ── _wait_for_deployment_active ─────────────────────────────────────────────
+
+
+class TestWaitForDeploymentActive:
+    def test_immediate_active(self):
+        b = _builder()
+        b._bedrock_client = Mock()
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Active"}
+        b._wait_for_deployment_active("arn:dep")
+        b._bedrock_client.get_custom_model_deployment.assert_called_once_with(
+            customModelDeploymentIdentifier="arn:dep"
+        )
+
+    def test_polls_then_active(self):
+        b = _builder()
+        b._bedrock_client = Mock()
+        b._bedrock_client.get_custom_model_deployment.side_effect = [
+            {"status": "Creating"},
+            {"status": "Creating"},
+            {"status": "Active"},
+        ]
+        with patch(f"{MODULE}.time.sleep"):
+            b._wait_for_deployment_active("arn:dep", poll_interval=1, max_wait=10)
+        assert b._bedrock_client.get_custom_model_deployment.call_count == 3
+
+    def test_failed_raises(self):
+        b = _builder()
+        b._bedrock_client = Mock()
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Failed"}
+        with pytest.raises(RuntimeError, match="failed"):
+            b._wait_for_deployment_active("arn:dep")
+
+    def test_timeout_raises(self):
+        b = _builder()
+        b._bedrock_client = Mock()
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Creating"}
+        with patch(f"{MODULE}.time.sleep"):
+            with pytest.raises(RuntimeError, match="Timed out"):
+                b._wait_for_deployment_active("arn:dep", poll_interval=1, max_wait=2)
+
+
 # ── deploy ──────────────────────────────────────────────────────────────────
 
 
@@ -432,10 +490,13 @@ def test_nova_full_chain(self):
         b._bedrock_client.create_custom_model_deployment.return_value = {
             "customModelDeploymentArn": "arn:dep"
         }
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Active"}
+
         result = b.deploy(custom_model_name="nova-m", role_arn="r")
         b._bedrock_client.create_custom_model.assert_called_once()
         b._bedrock_client.get_custom_model.assert_called_once()
         b._bedrock_client.create_custom_model_deployment.assert_called_once()
+        b._bedrock_client.get_custom_model_deployment.assert_called_once()
         assert result["customModelDeploymentArn"] == "arn:dep"
 
     def test_nova_via_hub_content_name(self):
@@ -449,6 +510,8 @@ def test_nova_via_hub_content_name(self):
         b._bedrock_client.create_custom_model_deployment.return_value = {
             "customModelDeploymentArn": "arn:dep"
         }
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Active"}
+
         result = b.deploy(custom_model_name="n", role_arn="r")
         assert result["customModelDeploymentArn"] == "arn:dep"
 
@@ -463,6 +526,8 @@ def test_nova_default_deployment_name(self):
         b._bedrock_client.create_custom_model_deployment.return_value = {
             "customModelDeploymentArn": "dep"
         }
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Active"}
+
         b.deploy(custom_model_name="my-model", role_arn="r")
         kw = b._bedrock_client.create_custom_model_deployment.call_args[1]
         assert kw["modelDeploymentName"] == "my-model-deployment"
@@ -478,6 +543,8 @@ def test_nova_with_tags(self):
         b._bedrock_client.create_custom_model_deployment.return_value = {
             "customModelDeploymentArn": "dep"
         }
+        b._bedrock_client.get_custom_model_deployment.return_value = {"status": "Active"}
+
         tags = [{"Key": "env", "Value": "test"}]
         b.deploy(custom_model_name="m", role_arn="r", model_tags=tags)
         kw = b._bedrock_client.create_custom_model.call_args[1]