Fix deployment-id attribution for az webapp deploy (OneDeploy)

src/azure-cli/azure/cli/command_modules/appservice/custom.py

@@ -9897,11 +9897,14 @@ def _check_runtimestatus_with_deploymentstatusapi(cmd, resource_group_name, name
         response_body = _check_zip_deployment_status(cmd, resource_group_name, name, deployment_status_url,
                                                      slot, timeout, deploy_params=deploy_params)
     else:
-        # get the deployment id
-        # once deploymentstatus/latest is available, we can use it to track the deployment
-        deployment_id = _get_latest_deployment_id(cmd, resource_group_name,
-                                                  name, deployment_status_url, slot,
-                                                  deploy_params=deploy_params)
+        # Prefer the request-specific id from the publish response; fall back to /latest.
+        deployment_id = None
+        if deploy_params is not None:
+            deployment_id = deploy_params._deployment_id  # pylint: disable=protected-access
+        if deployment_id is None:
+            deployment_id = _get_latest_deployment_id(cmd, resource_group_name,
+                                                      name, deployment_status_url, slot,
+                                                      deploy_params=deploy_params)
         if deployment_id is None:
             logger.warning("Failed to enable tracking runtime status for this deployment. "
                            "Resuming without tracking status.")
@@ -11149,6 +11152,7 @@ def __init__(self):
         # host_name_ssl_states on each access (trivial iteration).
         self._cached_scm_headers = None
         self._cached_site = None
+        self._deployment_id = None
 # pylint: enable=too-many-instance-attributes,too-few-public-methods
 
 
@@ -11561,6 +11565,9 @@ def _make_onedeploy_request(params):
             response = send_raw_request(params.cmd.cli_ctx, "PUT", deploy_url, body=body)
         poll_async_deployment_for_debugging = False
 
+    # Pin THIS request's deployment id from the publish response; avoids the racy /latest lookup.
+    params._deployment_id = response.headers.get('SCM-DEPLOYMENT-ID')  # pylint: disable=protected-access
+
     # check the status of deployment
     # pylint: disable=too-many-nested-blocks
     if response.status_code == 202 or response.status_code == 200:

src/azure-cli/azure/cli/command_modules/appservice/tests/latest/test_deployment_context_engine.py

@@ -25,6 +25,7 @@
     EnrichedDeploymentError,
     _determine_deployment_type,
 )
+from azure.cli.command_modules.appservice import custom
 
 
 def _make_mock_params(**overrides):
@@ -450,5 +451,49 @@ def test_200_not_extracted(self):
         self.assertIsNone(extract_status_code_from_message("Status Code: 200"))
 
 
+# ---------------------------------------------------------------------------
+# Deployment-id attribution: prefer the request-specific SCM-DEPLOYMENT-ID
+# (captured on OneDeployParams) over the global /api/deployments/latest lookup.
+# ---------------------------------------------------------------------------
+def _make_deploy_params(deployment_id):
+    params = MagicMock()
+    params._deployment_id = deployment_id
+    return params
+
+
+class TestDeploymentIdSelection(unittest.TestCase):
+
+    @patch.object(custom, '_poll_deployment_runtime_status', return_value={'status': 'RuntimeSuccessful'})
+    @patch.object(custom, '_build_deploymentstatus_url', return_value='https://scm/deploymentstatus/x')
+    @patch.object(custom, '_get_latest_deployment_id')
+    @patch.object(custom, '_get_or_fetch_is_linux_webapp', return_value=True)
+    def test_uses_request_specific_id_when_present(self, _linux, mock_latest, mock_build_url, _poll):
+        deploy_params = _make_deploy_params('scm-id-123')
+
+        result = custom._check_runtimestatus_with_deploymentstatusapi(
+            MagicMock(), 'rg', 'app', None,
+            'https://scm/api/deployments/latest', False, None,
+            deploy_params=deploy_params)
+
+        mock_latest.assert_not_called()
+        self.assertEqual(mock_build_url.call_args[0][4], 'scm-id-123')
+        self.assertEqual(result, {'status': 'RuntimeSuccessful'})
+
+    @patch.object(custom, '_poll_deployment_runtime_status', return_value={'status': 'RuntimeSuccessful'})
+    @patch.object(custom, '_build_deploymentstatus_url', return_value='https://scm/deploymentstatus/x')
+    @patch.object(custom, '_get_latest_deployment_id', return_value='latest-id-999')
+    @patch.object(custom, '_get_or_fetch_is_linux_webapp', return_value=True)
+    def test_falls_back_to_latest_when_id_absent(self, _linux, mock_latest, mock_build_url, _poll):
+        deploy_params = _make_deploy_params(None)
+
+        custom._check_runtimestatus_with_deploymentstatusapi(
+            MagicMock(), 'rg', 'app', None,
+            'https://scm/api/deployments/latest', False, None,
+            deploy_params=deploy_params)
+
+        mock_latest.assert_called_once()
+        self.assertEqual(mock_build_url.call_args[0][4], 'latest-id-999')
+
+
 if __name__ == '__main__':
     unittest.main()