@@ -357,15 +357,19 @@ def pad(self, file, size=6):
         Returns:
             generator of records
         """
-        buffer = ""
+        buffer = None
         for element in self.splitter.split(file):
+            if buffer is None:
+                # Match the buffer type to the record type so binary inputs
+                # concatenate correctly instead of raising a TypeError.
+                buffer = b"" if isinstance(element, bytes) else ""
             if _payload_size_within_limit(buffer + element, size):
                 buffer += element
             else:
                 tmp = buffer
                 buffer = element
                 yield tmp
-        if _validate_payload_size(buffer, size):
+        if buffer is not None and _validate_payload_size(buffer, size):
             yield buffer
 
 
 
@@ -39,6 +39,7 @@
 from sagemaker.core.config.config_schema import CONTAINER_CONFIG, LOCAL
 import sagemaker.core
 from sagemaker.core.common_utils import custom_extractall_tarfile
+from sagemaker.core.constants import DIR_PARAM_NAME, SAGEMAKER_OUTPUT_LOCATION
 
 CONTAINER_PREFIX = "algo"
 STUDIO_HOST_NAME = "sagemaker-local"
@@ -277,9 +278,7 @@ def train(self, input_data_config, output_data_config, hyperparameters, environm
             data_dir, input_data_config, output_data_config, hyperparameters
         )
         # If local, source directory needs to be updated to mounted /opt/ml/code path
-        hyperparameters = self._update_local_src_path(
-            hyperparameters, key=sagemaker.serve.model_builder.DIR_PARAM_NAME
-        )
+        hyperparameters = self._update_local_src_path(hyperparameters, key=DIR_PARAM_NAME)
 
         # Create the configuration files for each container that we will create
         # Each container will map the additional local volumes (if any).
@@ -344,15 +343,15 @@ def serve(self, model_dir, environment):
         volumes = self._prepare_serving_volumes(model_dir)
 
         # If the user script was passed as a file:// mount it to the container.
-        if sagemaker.serve.model_builder.DIR_PARAM_NAME.upper() in environment:
-            script_dir = environment[sagemaker.serve.model_builder.DIR_PARAM_NAME.upper()]
+        if DIR_PARAM_NAME.upper() in environment:
+            script_dir = environment[DIR_PARAM_NAME.upper()]
             parsed_uri = urlparse(script_dir)
             if parsed_uri.scheme == "file":
                 host_dir = os.path.abspath(parsed_uri.netloc + parsed_uri.path)
                 volumes.append(_Volume(host_dir, "/opt/ml/code"))
                 # Update path to mount location
                 environment = environment.copy()
-                environment[sagemaker.serve.model_builder.DIR_PARAM_NAME.upper()] = "/opt/ml/code"
+                environment[DIR_PARAM_NAME.upper()] = "/opt/ml/code"
 
         if _ecr_login_if_needed(self.sagemaker_session.boto_session, self.image):
             _pull_image(self.image)
@@ -583,8 +582,8 @@ def _prepare_training_volumes(
 
         # If there is a training script directory and it is a local directory,
         # mount it to the container.
-        if sagemaker.serve.model_builder.DIR_PARAM_NAME in hyperparameters:
-            training_dir = json.loads(hyperparameters[sagemaker.serve.model_builder.DIR_PARAM_NAME])
+        if DIR_PARAM_NAME in hyperparameters:
+            training_dir = json.loads(hyperparameters[DIR_PARAM_NAME])
             parsed_uri = urlparse(training_dir)
             if parsed_uri.scheme == "file":
                 host_dir = os.path.abspath(parsed_uri.netloc + parsed_uri.path)
@@ -593,10 +592,7 @@ def _prepare_training_volumes(
                 volumes.append(_Volume(shared_dir, "/opt/ml/shared"))
 
         parsed_uri = urlparse(output_data_config["S3OutputPath"])
-        if (
-            parsed_uri.scheme == "file"
-            and sagemaker.serve.model_builder.SAGEMAKER_OUTPUT_LOCATION in hyperparameters
-        ):
+        if parsed_uri.scheme == "file" and SAGEMAKER_OUTPUT_LOCATION in hyperparameters:
             dir_path = os.path.abspath(parsed_uri.netloc + parsed_uri.path)
             intermediate_dir = os.path.join(dir_path, "output", "intermediate")
             if not os.path.exists(intermediate_dir):
 
@@ -86,6 +86,25 @@ def __init__(self, sagemaker_session=None):
         """
         self.sagemaker_session = sagemaker_session or LocalSession()
 
+    def describe_user_profile(self, DomainId, UserProfileName, **kwargs):
+        """Passes ``describe_user_profile`` through to the real SageMaker client.
+
+        Local mode does not model user profiles, but Studio role resolution calls
+        this during session setup. Delegating to a real boto client keeps that path
+        working instead of raising ``AttributeError``.
+
+        Args:
+          DomainId (str): The domain ID the user profile belongs to.
+          UserProfileName (str): The name of the user profile to describe.
+          **kwargs: Additional keyword arguments forwarded to the boto client.
+
+        Returns: (dict) DescribeUserProfile response.
+        """
+        boto_client = self.sagemaker_session.boto_session.client("sagemaker")
+        return boto_client.describe_user_profile(
+            DomainId=DomainId, UserProfileName=UserProfileName, **kwargs
+        )
+
     @_telemetry_emitter(Feature.LOCAL_MODE, "local_session.create_processing_job")
     def create_processing_job(
         self,
@@ -533,7 +552,28 @@ def invoke_endpoint(
             Body = Body.encode("utf-8")
         r = self.http.request("POST", url, body=Body, preload_content=False, headers=headers)
 
-        return {"Body": r, "ContentType": Accept}
+        # Mirror the response shape of the real SageMaker runtime client so the same
+        # code works in local mode and against a hosted endpoint. HTTP header names
+        # are case-insensitive and containers send e.g. ``Content-Type`` or
+        # ``content-type``, so look them up without regard to case.
+        try:
+            response_headers = dict(r.headers)
+        except (TypeError, ValueError):
+            response_headers = {}
+        lowered = {str(k).lower(): v for k, v in response_headers.items()}
+        request_id = lowered.get("x-amzn-requestid", "local-request-id")
+        status_code = getattr(r, "status", None)
+        return {
+            "Body": r,
+            "ContentType": lowered.get("content-type", Accept),
+            "InvokedProductionVariant": TargetVariant or "AllTraffic",
+            "ResponseMetadata": {
+                "RequestId": request_id,
+                "HTTPStatusCode": status_code,
+                "HTTPHeaders": response_headers,
+                "RetryAttempts": 0,
+            },
+        }
 
 
 class LocalSession(Session):
 
@@ -278,11 +278,12 @@ def __init__(
         self.role = resolve_value_from_config(
             role, PROCESSING_JOB_ROLE_ARN_PATH, sagemaker_session=self.sagemaker_session
         )
-        if not self.role:
+        if not self.role and self.instance_type not in ("local", "local_gpu"):
             # Originally IAM role was a required parameter.
             # Now we marked that as Optional because we can fetch it from SageMakerConfig
             # Because of marking that parameter as optional, we should validate if it is None, even
-            # after fetching the config.
+            # after fetching the config. In Local Mode the role is never used
+            # (LocalSagemakerClient.create_processing_job discards it), so it is not required.
             raise ValueError("An AWS IAM role is required to create a Processing job.")
 
         self.env = resolve_value_from_config(
@@ -760,11 +761,12 @@ def _get_process_args(self, inputs, outputs, experiment_config):
             process_request_args["network_config"] = self.network_config._to_request_dict()
         else:
             process_request_args["network_config"] = None
-        process_request_args["role_arn"] = (
-            self.role
-            if is_pipeline_variable(self.role)
-            else self.sagemaker_session.expand_role(self.role)
-        )
+        if self.role is None or is_pipeline_variable(self.role):
+            # No role in Local Mode (see __init__): the local client discards RoleArn,
+            # and expand_role(None) would raise.
+            process_request_args["role_arn"] = self.role
+        else:
+            process_request_args["role_arn"] = self.sagemaker_session.expand_role(self.role)
         process_request_args["tags"] = self.tags
         return process_request_args
 
 
@@ -0,0 +1,15 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+#     http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""Integ tests for sagemaker-core Local Mode session/runtime fixes."""
+
+from __future__ import absolute_import