diff --git a/samcli/commands/local/lib/local_lambda.py b/samcli/commands/local/lib/local_lambda.py
index c4e0c0f6ae..a9405d564d 100644
--- a/samcli/commands/local/lib/local_lambda.py
+++ b/samcli/commands/local/lib/local_lambda.py
@@ -96,47 +96,35 @@ def __init__(
         self.container_host_interface = container_host_interface
         self.extra_hosts = extra_hosts
 
-    def invoke(
-        self,
-        function_identifier: str,
-        event: str,
-        tenant_id: Optional[str] = None,
-        stdout: Optional[StreamWriter] = None,
-        stderr: Optional[StreamWriter] = None,
-        override_runtime: Optional[str] = None,
-        invocation_type: str = "RequestResponse",
-        durable_execution_name: Optional[str] = None,
-    ) -> Optional[Dict[str, str]]:
+    def get_function(self, function_identifier: str, tenant_id: Optional[str] = None) -> Function:
         """
-        Find the Lambda function with given name and invoke it. Pass the given event to the function and return
-        response through the given streams.
-
-        This function will block until either the function completes or times out.
+        Get a Lambda function by identifier, raising FunctionNotFound if not found.
 
         Parameters
         ----------
-        function_identifier str
-            Identifier of the Lambda function to invoke, it can be logicalID, function name or full path
-        event str
-            Event data passed to the function. Must be a valid JSON String.
-        stdout samcli.lib.utils.stream_writer.StreamWriter
-            Stream writer to write the output of the Lambda function to.
-        stderr samcli.lib.utils.stream_writer.StreamWriter
-            Stream writer to write the Lambda runtime logs to.
-        Runtime: str
-            To use instead of the runtime specified in the function configuration
-        durable_execution_name: str
-            Optional name for the durable execution (for durable functions only)
-
+        function_identifier : str
+            Identifier of the Lambda function, it can be logicalID, function name or full path
+        tenant_id : Optional[str]
+            Optional tenant ID for multi-tenant functions
         Returns
         -------
-        Optional[Dict[str, str]]
-            HTTP headers dict if this was a durable function invocation, None otherwise
+        Function
+            The Lambda function configuration
 
         Raises
         ------
-        FunctionNotfound
-            When we cannot find a function with the given name
+        InvalidFunctionNameException
+            When the function identifier doesn't match AWS Lambda's validation pattern
+        FunctionNotFound
+            When we cannot find a function with the given identifier
+        TenantIdValidationError
+            When the tenant ID is not provided for multi-tenant functions
+        UnsupportedInlineCodeError
+            When the function has inline code and is being invoked locally
+        InvalidIntermediateImageError
+            When the function has an intermediate image and is being invoked locally
+        UnsupportedRuntimeArchitectureError
+            When the function runtime and architecture are not compatible
         """
         # Normalize function identifier from ARN if provided
         normalized_function_identifier = normalize_sam_function_identifier(function_identifier)
@@ -182,6 +170,55 @@ def invoke(
                 "Remove the tenant ID from your request and try again."
             )
 
+        return function
+
+    def invoke(
+        self,
+        function_identifier: str,
+        event: str,
+        tenant_id: Optional[str] = None,
+        stdout: Optional[StreamWriter] = None,
+        stderr: Optional[StreamWriter] = None,
+        override_runtime: Optional[str] = None,
+        invocation_type: str = "RequestResponse",
+        durable_execution_name: Optional[str] = None,
+        function: Optional[Function] = None,
+    ) -> Optional[Dict[str, str]]:
+        """
+        Find the Lambda function with given name and invoke it. Pass the given event to the function and return
+        response through the given streams.
+
+        This function will block until either the function completes or times out.
+
+        Parameters
+        ----------
+        function_identifier str
+            Identifier of the Lambda function to invoke, it can be logicalID, function name or full path
+        event str
+            Event data passed to the function. Must be a valid JSON String.
+        stdout samcli.lib.utils.stream_writer.StreamWriter
+            Stream writer to write the output of the Lambda function to.
+        stderr samcli.lib.utils.stream_writer.StreamWriter
+            Stream writer to write the Lambda runtime logs to.
+        Runtime: str
+            To use instead of the runtime specified in the function configuration
+        durable_execution_name: str
+            Optional name for the durable execution (for durable functions only)
+
+        Returns
+        -------
+        Optional[Dict[str, str]]
+            HTTP headers dict if this was a durable function invocation, None otherwise
+
+        Raises
+        ------
+        FunctionNotfound
+            When we cannot find a function with the given name
+        """
+        # Get the function configuration
+        if not function:
+            function = self.get_function(function_identifier, tenant_id)
+
         config = self.get_invoke_config(function, override_runtime)
 
         if (
diff --git a/samcli/local/lambda_service/lambda_error_responses.py b/samcli/local/lambda_service/lambda_error_responses.py
index 4ec0b08b2c..73feff1d04 100644
--- a/samcli/local/lambda_service/lambda_error_responses.py
+++ b/samcli/local/lambda_service/lambda_error_responses.py
@@ -2,6 +2,9 @@
 
 import json
 from collections import OrderedDict
+from typing import Any, Dict
+
+from flask import Response
 
 from samcli.local.services.base_local_service import BaseLocalService
 
@@ -40,7 +43,7 @@ class LambdaErrorResponses:
     CONTENT_TYPE_HEADER_KEY = "Content-Type"
 
     @staticmethod
-    def resource_not_found(function_name):
+    def resource_not_found(function_name: str) -> Response:
         """
         Creates a Lambda Service ResourceNotFound Response
 
@@ -66,7 +69,7 @@ def resource_not_found(function_name):
         )
 
     @staticmethod
-    def invalid_request_content(message):
+    def invalid_request_content(message: str) -> Response:
         """
         Creates a Lambda Service InvalidRequestContent Response
 
@@ -89,7 +92,7 @@ def invalid_request_content(message):
         )
 
     @staticmethod
-    def validation_exception(message):
+    def validation_exception(message: str) -> Response:
         """
         Creates a Lambda Service ValidationException Response
 
@@ -112,7 +115,7 @@ def validation_exception(message):
         )
 
     @staticmethod
-    def unsupported_media_type(content_type):
+    def unsupported_media_type(content_type: str) -> Response:
         """
         Creates a Lambda Service UnsupportedMediaType Response
 
@@ -137,7 +140,7 @@ def unsupported_media_type(content_type):
         )
 
     @staticmethod
-    def generic_service_exception(*args):
+    def generic_service_exception(*args: Any) -> Response:
         """
         Creates a Lambda Service Generic ServiceException Response
 
@@ -160,7 +163,7 @@ def generic_service_exception(*args):
         )
 
     @staticmethod
-    def not_implemented_locally(message):
+    def not_implemented_locally(message: str) -> Response:
         """
         Creates a Lambda Service NotImplementedLocally Response
 
@@ -183,7 +186,7 @@ def not_implemented_locally(message):
         )
 
     @staticmethod
-    def generic_path_not_found(*args):
+    def generic_path_not_found(*args: Any) -> Response:
         """
         Creates a Lambda Service Generic PathNotFound Response
 
@@ -208,7 +211,7 @@ def generic_path_not_found(*args):
         )
 
     @staticmethod
-    def generic_method_not_allowed(*args):
+    def generic_method_not_allowed(*args: Any) -> Response:
         """
         Creates a Lambda Service Generic MethodNotAllowed Response
 
@@ -233,13 +236,13 @@ def generic_method_not_allowed(*args):
         )
 
     @staticmethod
-    def container_creation_failed(message):
+    def container_creation_failed(message: str) -> Response:
         """
         Creates a Container Creation Failed response
         Parameters
         ----------
-        args list
-            List of arguments Flask passes to the method
+        message str
+            Message to be added to the body of the response
         Returns
         -------
         Flask.Response
@@ -256,7 +259,7 @@ def container_creation_failed(message):
         )
 
     @staticmethod
-    def _construct_error_response_body(error_type, error_message):
+    def _construct_error_response_body(error_type: str, error_message: str) -> str:
         """
         Constructs a string to be used in the body of the Response that conforms
         to the structure of the Lambda Service Responses
@@ -278,7 +281,7 @@ def _construct_error_response_body(error_type, error_message):
 
     # Durable Functions Error Responses
     @staticmethod
-    def durable_execution_not_found(execution_arn):
+    def durable_execution_not_found(execution_arn: str) -> Response:
         """Creates a ResourceNotFound response for durable executions"""
         exception_tuple = LambdaErrorResponses.ResourceNotFoundException
         return BaseLocalService.service_response(
@@ -290,7 +293,7 @@ def durable_execution_not_found(execution_arn):
         )
 
     @staticmethod
-    def _construct_headers(error_type):
+    def _construct_headers(error_type: str) -> Dict[str, str]:
         """
         Constructs Headers for the Local Lambda Error Response
 
diff --git a/samcli/local/lambda_service/local_lambda_http_service.py b/samcli/local/lambda_service/local_lambda_http_service.py
index c4fecd3b7c..05f2a6e598 100644
--- a/samcli/local/lambda_service/local_lambda_http_service.py
+++ b/samcli/local/lambda_service/local_lambda_http_service.py
@@ -3,14 +3,22 @@
 import io
 import json
 import logging
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
+from typing import Dict, Optional, Tuple
 from urllib.parse import unquote
 
 from flask import Flask, request
 from werkzeug.routing import BaseConverter
 
 from samcli.commands.local.cli_common.durable_context import DurableContext
-from samcli.commands.local.lib.exceptions import TenantIdValidationError, UnsupportedInlineCodeError
+from samcli.commands.local.lib.exceptions import (
+    InvalidIntermediateImageError,
+    TenantIdValidationError,
+    UnsupportedInlineCodeError,
+)
+from samcli.lib.providers.provider import Function
+from samcli.lib.utils.invocation_type import EVENT
 from samcli.lib.utils.name_utils import InvalidFunctionNameException, normalize_sam_function_identifier
 from samcli.lib.utils.stream_writer import StreamWriter
 from samcli.local.docker.exceptions import DockerContainerCreationFailedException
@@ -67,6 +75,7 @@ def __init__(self, lambda_runner, port, host, stderr=None, ssl_context=None):
         super().__init__(lambda_runner.is_debugging(), port=port, host=host, ssl_context=ssl_context)
         self.lambda_runner = lambda_runner
         self.stderr = stderr
+        self.executor = ThreadPoolExecutor()
 
     def create(self):
         """
@@ -240,36 +249,41 @@ def _invoke_request_handler(self, function_name):
         # Extract durable execution name from headers
         durable_execution_name = flask_request.headers.get("X-Amz-Durable-Execution-Name")
 
-        stdout_stream_string = io.StringIO()
-        stdout_stream_bytes = io.BytesIO()
-        stdout_stream_writer = StreamWriter(stdout_stream_string, stdout_stream_bytes, auto_flush=True)
+        headers = {"Content-Type": "application/json"}
 
         try:
-            # Normalize function name from ARN if provided
-            normalized_function_name = normalize_sam_function_identifier(function_name)
-
-            invoke_headers = self.lambda_runner.invoke(
-                normalized_function_name,
-                request_data,
-                invocation_type=invocation_type,
-                durable_execution_name=durable_execution_name,
-                tenant_id=tenant_id,
-                stdout=stdout_stream_writer,
-                stderr=self.stderr,
-            )
-        except (InvalidFunctionNameException, TenantIdValidationError) as e:
+            function = self.lambda_runner.get_function(function_name, tenant_id)
+        except (InvalidFunctionNameException, TenantIdValidationError, InvalidIntermediateImageError) as e:
             LOG.error("Validation error: %s", str(e))
             return LambdaErrorResponses.validation_exception(str(e))
-        except UnsupportedInvocationType as e:
-            LOG.warning("invocation-type: %s is not supported. RequestResponse is only supported.", invocation_type)
-            return LambdaErrorResponses.not_implemented_locally(str(e))
         except FunctionNotFound:
+            normalized_function_name = normalize_sam_function_identifier(function_name)
             LOG.debug("%s was not found to invoke.", normalized_function_name)
             return LambdaErrorResponses.resource_not_found(normalized_function_name)
         except UnsupportedInlineCodeError:
             return LambdaErrorResponses.not_implemented_locally(
                 "Inline code is not supported for sam local commands. Please write your code in a separate file."
             )
+
+        arguments = {
+            "function_name": function_name,
+            "request_data": request_data,
+            "invocation_type": invocation_type,
+            "durable_execution_name": durable_execution_name,
+            "tenant_id": tenant_id,
+            "function": function,
+        }
+
+        if invocation_type == EVENT:
+            self.executor.submit(self._invoke_async_lambda, **arguments)
+            return self.service_response("", headers, 202)
+        try:
+            invoke_headers, stdout_stream_string, stdout_stream_bytes = self._invoke_lambda(**arguments)
+        except UnsupportedInvocationType as e:
+            LOG.warning(
+                "invocation-type: %s is not supported. Only Event and RequestResponse are supported.", invocation_type
+            )
+            return LambdaErrorResponses.not_implemented_locally(str(e))
         except DockerContainerCreationFailedException as ex:
             return LambdaErrorResponses.container_creation_failed(ex.message)
 
@@ -278,20 +292,55 @@ def _invoke_request_handler(self, function_name):
         )
 
         # Prepare headers
-        headers = {"Content-Type": "application/json"}
         if invoke_headers and isinstance(invoke_headers, dict):
             headers.update(invoke_headers)
 
         if is_lambda_user_error_response:
             headers["x-amz-function-error"] = "Unhandled"
-            return self.service_response(lambda_response, headers, 200)
-
-        # For async invocations (Event type), return 202
-        if invocation_type == "Event":
-            return self.service_response("", headers, 202)
 
         return self.service_response(lambda_response, headers, 200)
 
+    def _invoke_async_lambda(self, function_name: str, **kwargs) -> None:
+        """
+        Wrapper for _invoke_lambda that runs in an async context (Event invocation type)
+        """
+        try:
+            self._invoke_lambda(function_name=function_name, **kwargs)
+        except Exception as e:
+            LOG.error("Async invocation failed for function %s: %s", function_name, str(e), exc_info=True)
+
+    def _invoke_lambda(
+        self,
+        function_name: str,
+        request_data: str,
+        invocation_type: str,
+        durable_execution_name: Optional[str],
+        tenant_id: Optional[str],
+        function: Optional[Function],
+    ) -> Tuple[Optional[Dict[str, str]], io.StringIO, io.BytesIO]:
+        """
+        Invokes a Lambda function and returns the result
+        """
+
+        stdout_stream_string = io.StringIO()
+        stdout_stream_bytes = io.BytesIO()
+        stdout_stream_writer = StreamWriter(stdout_stream_string, stdout_stream_bytes, auto_flush=True)
+
+        normalized_function_name = normalize_sam_function_identifier(function_name)
+
+        invoke_headers = self.lambda_runner.invoke(
+            normalized_function_name,
+            request_data,
+            invocation_type=invocation_type,
+            durable_execution_name=durable_execution_name,
+            tenant_id=tenant_id,
+            function=function,
+            stdout=stdout_stream_writer,
+            stderr=self.stderr,
+        )
+
+        return invoke_headers, stdout_stream_string, stdout_stream_bytes
+
     def _get_durable_execution_handler(self, durable_execution_arn):
         """
         Handler for GET /2025-12-01/durable-executions/{DurableExecutionArn}
diff --git a/samcli/local/lambdafn/runtime.py b/samcli/local/lambdafn/runtime.py
index 0f1b5611bc..40e66d55e0 100644
--- a/samcli/local/lambdafn/runtime.py
+++ b/samcli/local/lambdafn/runtime.py
@@ -13,6 +13,7 @@
 
 from samcli.lib.telemetry.metric import capture_parameter
 from samcli.lib.utils.file_observer import LambdaFunctionObserver
+from samcli.lib.utils.invocation_type import EVENT, REQUEST_RESPONSE
 from samcli.lib.utils.packagetype import ZIP
 from samcli.local.docker.container import Container, ContainerContext
 from samcli.local.docker.container_analyzer import ContainerAnalyzer
@@ -305,9 +306,10 @@ def invoke(
                 )
             else:
                 # Only RequestResponse supported for regular Lambda functions
-                if invocation_type != "RequestResponse":
+                if invocation_type not in [EVENT, REQUEST_RESPONSE]:
                     raise UnsupportedInvocationType(
-                        f"invocation-type: {invocation_type} is not supported. RequestResponse is only supported."
+                        f"invocation-type: {invocation_type} is not supported. "
+                        "Only Event and RequestResponse are supported."
                     )
 
                 # The container handles concurrency control internally via its semaphore.
diff --git a/samcli/local/services/base_local_service.py b/samcli/local/services/base_local_service.py
index b2033c91f4..d1acfd82e1 100644
--- a/samcli/local/services/base_local_service.py
+++ b/samcli/local/services/base_local_service.py
@@ -4,7 +4,7 @@
 import json
 import logging
 import signal
-from typing import Optional, Tuple, Union
+from typing import Dict, Optional, Tuple, Union
 
 from flask import Response
 
@@ -79,7 +79,7 @@ def interrupt_handler(sig, frame):
         self._app.run(threaded=multi_threaded, host=self.host, port=self.port, ssl_context=self.ssl_context)
 
     @staticmethod
-    def service_response(body, headers, status_code):
+    def service_response(body: str, headers: Dict[str, str], status_code: int) -> Response:
         """
         Constructs a Flask Response from the body, headers, and status_code.
 
@@ -89,7 +89,7 @@ def service_response(body, headers, status_code):
         :return: Flask Response
         """
         response = Response(body)
-        response.headers = headers
+        response.headers.update(headers)
         response.status_code = status_code
         return response
 
diff --git a/tests/integration/local/start_lambda/test_start_lambda.py b/tests/integration/local/start_lambda/test_start_lambda.py
index c853eb3c21..c2765887b5 100644
--- a/tests/integration/local/start_lambda/test_start_lambda.py
+++ b/tests/integration/local/start_lambda/test_start_lambda.py
@@ -69,6 +69,15 @@ def test_invoke_with_non_json_data(self):
 
         self.assertEqual(str(error.exception), expected_error_message)
 
+        with self.assertRaises(ClientError) as error:
+            self.lambda_client.invoke(
+                FunctionName="EchoEventFunction",
+                Payload="notat:asdfasdf",
+                InvocationType="Event",
+            )
+
+        self.assertEqual(str(error.exception), expected_error_message)
+
     @pytest.mark.flaky(reruns=3)
     @pytest.mark.timeout(timeout=300, method="thread")
     def test_invoke_with_log_type_not_None(self):
@@ -87,7 +96,7 @@ def test_invoke_with_log_type_not_None(self):
     def test_invoke_with_invocation_type_not_RequestResponse(self):
         expected_error_message = (
             "An error occurred (NotImplemented) when calling the Invoke operation: "
-            "invocation-type: DryRun is not supported. RequestResponse is only supported."
+            "invocation-type: DryRun is not supported. Only Event and RequestResponse are supported."
         )
 
         with self.assertRaises(ClientError) as error:
@@ -95,6 +104,73 @@ def test_invoke_with_invocation_type_not_RequestResponse(self):
 
         self.assertEqual(str(error.exception), expected_error_message)
 
+    @pytest.mark.flaky(reruns=3)
+    @pytest.mark.timeout(timeout=300, method="thread")
+    def test_invoke_function_with_image_uri_missing(self):
+        expected_error_message = (
+            "An error occurred (ValidationException) when calling the Invoke operation:"
+            " ImageUri not provided for Function: HelloWorldFunctionMissingImageUri of PackageType: Image"
+        )
+
+        with self.assertRaises(ClientError) as error:
+            self.lambda_client.invoke(FunctionName="HelloWorldFunctionMissingImageUri", Payload='"This is json data"')
+
+        self.assertEqual(str(error.exception), expected_error_message)
+
+        with self.assertRaises(ClientError) as error:
+            self.lambda_client.invoke(
+                FunctionName="HelloWorldFunctionMissingImageUri",
+                Payload='"This is json data"',
+                InvocationType="Event",
+            )
+
+        self.assertEqual(str(error.exception), expected_error_message)
+
+    @pytest.mark.flaky(reruns=3)
+    @pytest.mark.timeout(timeout=300, method="thread")
+    def test_invoke_function_with_missing_tenant_id(self):
+        expected_error_message = (
+            "An error occurred (ValidationException) when calling the Invoke operation:"
+            " The invoked function is enabled with tenancy configuration. Add a valid tenant ID in your request and try again."
+        )
+
+        with self.assertRaises(ClientError) as error:
+            self.lambda_client.invoke(FunctionName="MultiTenantFunction", Payload='"This is json data"')
+
+        self.assertEqual(str(error.exception), expected_error_message)
+
+        with self.assertRaises(ClientError) as error:
+            self.lambda_client.invoke(
+                FunctionName="MultiTenantFunction", InvocationType="Event", Payload='"This is json data"'
+            )
+
+        self.assertEqual(str(error.exception), expected_error_message)
+
+    @pytest.mark.flaky(reruns=3)
+    @pytest.mark.timeout(timeout=300, method="thread")
+    def test_invoke_function_with_tenant_id(self):
+        expected_error_message = (
+            "An error occurred (ValidationException) when calling the Invoke operation:"
+            " The invoked function is not enabled with tenancy configuration. Remove the tenant ID from your request and try again."
+        )
+
+        with self.assertRaises(ClientError) as error:
+            self.lambda_client.invoke(
+                FunctionName="EchoEventFunction", TenantId="tenant-123", Payload='"This is json data"'
+            )
+
+        self.assertEqual(str(error.exception), expected_error_message)
+
+        with self.assertRaises(ClientError) as error:
+            self.lambda_client.invoke(
+                FunctionName="EchoEventFunction",
+                TenantId="tenant-123",
+                Payload='"This is json data"',
+                InvocationType="Event",
+            )
+
+        self.assertEqual(str(error.exception), expected_error_message)
+
 
 class TestLambdaServiceWithInlineCode(StartLambdaIntegBaseClass):
     template_path = "/testdata/invoke/template-inlinecode.yaml"
@@ -292,6 +368,34 @@ def test_invoke_with_function_timeout_using_lookup_value(self, use_full_path):
         self.assertIsNone(response.get("FunctionError"))
         self.assertEqual(response.get("StatusCode"), 200)
 
+    @pytest.mark.flaky(reruns=3)
+    @pytest.mark.timeout(timeout=300, method="thread")
+    def test_invoke_with_event_invocation_type(self):
+        response = self.lambda_client.invoke(
+            FunctionName="EchoEventFunction",
+            Payload='"This is json data"',
+            InvocationType="Event",
+        )
+        self.assertEqual(response.get("Payload").read().decode("utf-8"), "")
+        self.assertIsNone(response.get("FunctionError"))
+        self.assertEqual(response.get("StatusCode"), 202)
+
+    @pytest.mark.flaky(reruns=3)
+    @pytest.mark.timeout(timeout=300, method="thread")
+    def test_invoke_with_event_invocation_type_calls_function(self):
+        response = self.lambda_client.invoke(
+            FunctionName="WriteToStderrFunction",
+            InvocationType="Event",
+        )
+        self.assertIsNone(response.get("FunctionError"))
+        self.assertEqual(response.get("StatusCode"), 202)
+
+        # Wait for function to be called
+        sleep(2)
+
+        # Check stderr that function was called
+        self.assertIn("Docker Lambda is writing to stderr", self.start_lambda_process_output)
+
 
 class TestWarmContainersBaseClass(StartLambdaIntegBaseClass):
     def setUp(self):
diff --git a/tests/integration/local/start_lambda/test_start_lambda_cdk.py b/tests/integration/local/start_lambda/test_start_lambda_cdk.py
index a8261afddf..3744e6a823 100644
--- a/tests/integration/local/start_lambda/test_start_lambda_cdk.py
+++ b/tests/integration/local/start_lambda/test_start_lambda_cdk.py
@@ -112,7 +112,7 @@ def test_invoke_with_log_type_not_None(self):
     def test_invoke_with_invocation_type_not_RequestResponse(self):
         expected_error_message = (
             "An error occurred (NotImplemented) when calling the Invoke operation: "
-            "invocation-type: DryRun is not supported. RequestResponse is only supported."
+            "invocation-type: DryRun is not supported. Only Event and RequestResponse are supported."
         )
 
         with self.assertRaises(ClientError) as error:
diff --git a/tests/integration/testdata/invoke/template.yml b/tests/integration/testdata/invoke/template.yml
index fc7a6b39e5..61efc176eb 100644
--- a/tests/integration/testdata/invoke/template.yml
+++ b/tests/integration/testdata/invoke/template.yml
@@ -244,5 +244,13 @@ Resources:
       Layers:
         - !Sub "arn:aws:lambda:us-east-1:553035198032:layer:git-lambda2:${LayerVersion}"
 
+  HelloWorldFunctionMissingImageUri:
+    Type: AWS::Serverless::Function
+    Properties:
+      PackageType: Image
+      ImageConfig:
+        Command:
+          - main.handler
+      Timeout: 600
 
 # UTF-8 Test 😐
diff --git a/tests/unit/local/lambda_service/test_lambda_error_responses.py b/tests/unit/local/lambda_service/test_lambda_error_responses.py
index 532fa66167..6fffc0566f 100644
--- a/tests/unit/local/lambda_service/test_lambda_error_responses.py
+++ b/tests/unit/local/lambda_service/test_lambda_error_responses.py
@@ -97,6 +97,19 @@ def test_generic_method_not_allowed(self, service_response_mock):
             405,
         )
 
+    @patch("samcli.local.services.base_local_service.BaseLocalService.service_response")
+    def test_validation_exception(self, service_response_mock):
+        service_response_mock.return_value = "ValidationException"
+
+        response = LambdaErrorResponses.validation_exception("ValidationException")
+
+        self.assertEqual(response, "ValidationException")
+        service_response_mock.assert_called_once_with(
+            '{"Type": "User", "Message": "ValidationException"}',
+            {"x-amzn-errortype": "ValidationException", "Content-Type": "application/json"},
+            400,
+        )
+
     @patch("samcli.local.services.base_local_service.BaseLocalService.service_response")
     def test_durable_execution_not_found(self, service_response_mock):
         service_response_mock.return_value = "DurableExecutionNotFound"
@@ -109,3 +122,16 @@ def test_durable_execution_not_found(self, service_response_mock):
             {"x-amzn-errortype": "ResourceNotFound", "Content-Type": "application/json"},
             404,
         )
+
+    @patch("samcli.local.services.base_local_service.BaseLocalService.service_response")
+    def test_container_creation_failed(self, service_response_mock):
+        service_response_mock.return_value = "ContainerCreationFailed"
+
+        response = LambdaErrorResponses.container_creation_failed("Container creation failed: test message")
+
+        self.assertEqual(response, "ContainerCreationFailed")
+        service_response_mock.assert_called_once_with(
+            '{"Type": "LocalService", "Message": "Container creation failed: test message"}',
+            {"x-amzn-errortype": "ContainerCreationFailed", "Content-Type": "application/json"},
+            501,
+        )
diff --git a/tests/unit/local/lambda_service/test_local_lambda_http_service.py b/tests/unit/local/lambda_service/test_local_lambda_http_service.py
index b105e2d02c..359b0c2065 100644
--- a/tests/unit/local/lambda_service/test_local_lambda_http_service.py
+++ b/tests/unit/local/lambda_service/test_local_lambda_http_service.py
@@ -1,3 +1,6 @@
+import threading
+import json
+from datetime import datetime
 from unittest import TestCase
 from unittest.mock import ANY, Mock, call, patch
 
@@ -7,7 +10,11 @@
 from samcli.lib.utils.name_utils import InvalidFunctionNameException
 from samcli.local.docker.exceptions import DockerContainerCreationFailedException
 from samcli.local.lambda_service import local_lambda_http_service
-from samcli.local.lambda_service.local_lambda_http_service import FunctionNamePathConverter, LocalLambdaHttpService
+from samcli.local.lambda_service.local_lambda_http_service import (
+    DateTimeEncoder,
+    FunctionNamePathConverter,
+    LocalLambdaHttpService,
+)
 from samcli.local.lambdafn.exceptions import DurableExecutionNotFound, FunctionNotFound
 
 
@@ -128,6 +135,7 @@ def test_invoke_request_handler(self, lambda_output_parser_mock, service_respons
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         service_response_mock.assert_called_once_with("hello world", {"Content-Type": "application/json"}, 200)
 
@@ -161,6 +169,7 @@ def test_invoke_request_handler_with_durable_execution_name_header(
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         service_response_mock.assert_called_once_with("hello world", {"Content-Type": "application/json"}, 200)
 
@@ -196,6 +205,7 @@ def test_invoke_request_handler_with_durable_execution_arn(self, lambda_output_p
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         service_response_mock.assert_called_once_with(
             "hello world", {"Content-Type": "application/json", "X-Amz-Durable-Execution-Arn": expected_arn}, 200
@@ -210,7 +220,7 @@ def test_invoke_request_handler_on_incorrect_path(self, lambda_error_responses_m
         local_lambda_http_service.request = request_mock
 
         lambda_runner_mock = Mock()
-        lambda_runner_mock.invoke.side_effect = FunctionNotFound
+        lambda_runner_mock.get_function.side_effect = FunctionNotFound
 
         lambda_error_responses_mock.resource_not_found.return_value = "Couldn't find Lambda"
 
@@ -220,15 +230,9 @@ def test_invoke_request_handler_on_incorrect_path(self, lambda_error_responses_m
 
         self.assertEqual(response, "Couldn't find Lambda")
 
-        lambda_runner_mock.invoke.assert_called_once_with(
-            "NotFound",
-            "{}",
-            invocation_type="RequestResponse",
-            durable_execution_name=None,
-            tenant_id=None,
-            stdout=ANY,
-            stderr=None,
-        )
+        # get_function is called first; invoke is never called when it raises
+        lambda_runner_mock.get_function.assert_called_once_with("NotFound", None)
+        lambda_runner_mock.invoke.assert_not_called()
 
         lambda_error_responses_mock.resource_not_found.assert_called_once_with("NotFound")
 
@@ -241,7 +245,7 @@ def test_invoke_request_function_contains_inline_code(self, lambda_error_respons
         local_lambda_http_service.request = request_mock
 
         lambda_runner_mock = Mock()
-        lambda_runner_mock.invoke.side_effect = UnsupportedInlineCodeError(message="Inline code is not supported")
+        lambda_runner_mock.get_function.side_effect = UnsupportedInlineCodeError(message="Inline code is not supported")
 
         lambda_error_responses_mock.not_implemented_locally.return_value = "Inline code is not supported"
 
@@ -251,16 +255,8 @@ def test_invoke_request_function_contains_inline_code(self, lambda_error_respons
 
         self.assertEqual(response, "Inline code is not supported")
 
-        lambda_runner_mock.invoke.assert_called_once_with(
-            "FunctionWithInlineCode",
-            "{}",
-            durable_execution_name=None,
-            tenant_id=None,
-            stdout=ANY,
-            stderr=None,
-            invocation_type="RequestResponse",
-        )
-
+        lambda_runner_mock.get_function.assert_called_once_with("FunctionWithInlineCode", None)
+        lambda_runner_mock.invoke.assert_not_called()
         lambda_error_responses_mock.not_implemented_locally.assert_called()
 
     @patch("samcli.local.lambda_service.local_lambda_http_service.LambdaErrorResponses")
@@ -285,11 +281,12 @@ def test_invoke_request_container_creation_failed(self, lambda_error_responses_m
         lambda_runner_mock.invoke.assert_called_once_with(
             "FunctionContainerCreationFailed",
             "{}",
+            invocation_type="RequestResponse",
             durable_execution_name=None,
             tenant_id=None,
             stdout=ANY,
             stderr=None,
-            invocation_type="RequestResponse",
+            function=ANY,
         )
 
         lambda_error_responses_mock.container_creation_failed.assert_called()
@@ -364,6 +361,7 @@ def test_invoke_request_handler_with_lambda_that_errors(self, lambda_output_pars
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         service_response_mock.assert_called_once_with(
             "hello world", {"Content-Type": "application/json", "x-amz-function-error": "Unhandled"}, 200
@@ -396,6 +394,7 @@ def test_invoke_request_handler_with_no_data(self, lambda_output_parser_mock, se
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         service_response_mock.assert_called_once_with("hello world", {"Content-Type": "application/json"}, 200)
 
@@ -431,13 +430,13 @@ def test_invoke_request_handler_async_invocation_returns_202(
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         # For async invocation, should return empty body with 202 status and execution ARN header
         service_response_mock.assert_called_once_with(
             "",
             {
                 "Content-Type": "application/json",
-                "X-Amz-Durable-Execution-Arn": "arn:aws:lambda:us-west-2:123456789012:function:test-function:$LATEST/durable-execution/test-123",
             },
             202,
         )
@@ -450,15 +449,13 @@ def test_invoke_request_handler_async_invocation_unsupported_function_returns_er
         request_mock = Mock()
         request_mock.get_data.return_value = b"{}"
         request_mock.args = {}
-        request_mock.headers = {"X-Amz-Invocation-Type": "Event"}
+        request_mock.headers = {"X-Amz-Invocation-Type": "DryRun"}
         local_lambda_http_service.request = request_mock
 
         lambda_runner_mock = Mock()
         from samcli.local.lambdafn.exceptions import UnsupportedInvocationType
 
-        lambda_runner_mock.invoke.side_effect = UnsupportedInvocationType(
-            "Async invocation not supported for regular Lambda functions"
-        )
+        lambda_runner_mock.invoke.side_effect = UnsupportedInvocationType("Dry Run invocation not supported")
 
         service = LocalLambdaHttpService(lambda_runner=lambda_runner_mock, port=3000, host="localhost")
 
@@ -469,17 +466,192 @@ def test_invoke_request_handler_async_invocation_unsupported_function_returns_er
         lambda_runner_mock.invoke.assert_called_once_with(
             "HelloWorld",
             "{}",
-            invocation_type="Event",
+            invocation_type="DryRun",
             durable_execution_name=None,
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
-        lambda_error_responses_mock.not_implemented_locally.assert_called_once_with(
-            "Async invocation not supported for regular Lambda functions"
-        )
+        lambda_error_responses_mock.not_implemented_locally.assert_called_once_with("Dry Run invocation not supported")
         self.assertEqual(result, "error response")
 
+    def test_event_invocation_runs_async_task(self):
+        # Test that Event invocation type runs the function asynchronously
+        handler_returned = threading.Event()
+        finished = threading.Event()
+
+        def fake_invoke(*args, **kwargs):
+            if handler_returned.wait(timeout=5):
+                finished.set()
+
+        request_mock = Mock()
+        request_mock.get_data.return_value = b"{}"
+        request_mock.args = {}
+        request_mock.headers = {"X-Amz-Invocation-Type": "Event"}
+        local_lambda_http_service.request = request_mock
+
+        lambda_runner_mock = Mock()
+        service = LocalLambdaHttpService(lambda_runner=lambda_runner_mock, port=3000, host="localhost")
+        service._invoke_lambda = fake_invoke
+        service.create()
+
+        response = service._invoke_request_handler(function_name="HelloWorld")
+
+        # Assert that first a 202 response is returned
+        self.assertEqual(response.status_code, 202)
+
+        # Then assert that invoke has not finished
+        self.assertFalse(finished.is_set())
+        handler_returned.set()
+
+        # Finally assert that invoke has finished
+        self.assertTrue(finished.wait(timeout=5), "Task never finished")
+
+    @patch("samcli.local.lambda_service.local_lambda_http_service.LocalLambdaHttpService.service_response")
+    @patch("samcli.local.lambda_service.local_lambda_http_service.ThreadPoolExecutor")
+    def test_invoke_request_handler_async_invocation_submits_to_executor(
+        self, executor_class_mock, service_response_mock
+    ):
+        # Test that async invocation (Event type) submits _invoke_async_lambda to executor
+        service_response_mock.return_value = "request response"
+        executor_mock = Mock()
+        executor_class_mock.return_value = executor_mock
+        future_mock = Mock()
+        executor_mock.submit.return_value = future_mock
+
+        request_mock = Mock()
+        request_mock.get_data.return_value = b'{"test": "data"}'
+        request_mock.args = {}
+        request_mock.headers = {"X-Amz-Invocation-Type": "Event"}
+        local_lambda_http_service.request = request_mock
+
+        lambda_runner_mock = Mock()
+        service = LocalLambdaHttpService(lambda_runner=lambda_runner_mock, port=3000, host="localhost")
+
+        response = service._invoke_request_handler(function_name="HelloWorld")
+
+        self.assertEqual(response, "request response")
+        # Verify executor.submit was called with _invoke_async_lambda and correct arguments
+        executor_mock.submit.assert_called_once()
+        submit_call = executor_mock.submit.call_args
+        self.assertEqual(submit_call[0][0], service._invoke_async_lambda)
+        self.assertEqual(submit_call[1]["function_name"], "HelloWorld")
+        self.assertEqual(submit_call[1]["request_data"], '{"test": "data"}')
+        self.assertEqual(submit_call[1]["invocation_type"], "Event")
+        # Verify 202 response is returned
+        service_response_mock.assert_called_once_with("", {"Content-Type": "application/json"}, 202)
+
+    @patch("samcli.local.lambda_service.local_lambda_http_service.LOG")
+    @patch("samcli.local.lambda_service.local_lambda_http_service.ThreadPoolExecutor")
+    def test_invoke_async_lambda_logs_exceptions(self, executor_class_mock, log_mock):
+        # Test that exceptions in async lambda invocation are logged
+        executor_mock = Mock()
+        executor_class_mock.return_value = executor_mock
+
+        request_mock = Mock()
+        request_mock.get_data.return_value = b"{}"
+        request_mock.args = {}
+        request_mock.headers = {"X-Amz-Invocation-Type": "Event"}
+        local_lambda_http_service.request = request_mock
+
+        lambda_runner_mock = Mock()
+        test_exception = Exception("Test async exception")
+        lambda_runner_mock.invoke.side_effect = test_exception
+
+        service = LocalLambdaHttpService(lambda_runner=lambda_runner_mock, port=3000, host="localhost")
+
+        # Trigger async invocation
+        service._invoke_request_handler(function_name="HelloWorld")
+
+        # Get the submitted function and execute it to trigger exception
+        submit_call = executor_mock.submit.call_args
+        async_function = submit_call[0][0]
+        async_kwargs = submit_call[1]
+
+        # Execute the async function which should catch and log the exception
+        async_function(**async_kwargs)
+
+        # Verify exception was logged
+        log_mock.error.assert_called_once()
+        error_call = log_mock.error.call_args
+        # Check format string
+        self.assertEqual(error_call[0][0], "Async invocation failed for function %s: %s")
+        # Check function name argument
+        self.assertEqual(error_call[0][1], "HelloWorld")
+        # Check exception message argument
+        self.assertEqual(error_call[0][2], "Test async exception")
+        # Check exc_info keyword argument
+        self.assertTrue(error_call[1]["exc_info"])
+
+    @patch("samcli.local.lambda_service.local_lambda_http_service.ThreadPoolExecutor")
+    @patch("samcli.local.lambda_service.local_lambda_http_service.LambdaErrorResponses")
+    def test_invoke_request_handler_async_invocation_function_not_found_returns_error(
+        self, lambda_error_responses_mock, executor_class_mock
+    ):
+        # Test that async invocation (Event type) returns error when function doesn't exist
+        executor_mock = Mock()
+        executor_class_mock.return_value = executor_mock
+
+        request_mock = Mock()
+        request_mock.get_data.return_value = b"{}"
+        request_mock.args = {}
+        request_mock.headers = {"X-Amz-Invocation-Type": "Event"}
+        local_lambda_http_service.request = request_mock
+
+        lambda_runner_mock = Mock()
+        lambda_runner_mock.get_function.side_effect = FunctionNotFound
+        lambda_error_responses_mock.resource_not_found.return_value = "function not found response"
+
+        service = LocalLambdaHttpService(lambda_runner=lambda_runner_mock, port=3000, host="localhost")
+
+        response = service._invoke_request_handler(function_name="NonExistentFunction")
+
+        # Verify error response is returned
+        self.assertEqual(response, "function not found response")
+        # Verify get_function was called to validate
+        lambda_runner_mock.get_function.assert_called_once_with("NonExistentFunction", None)
+        # Verify executor was NOT called since validation failed
+        executor_mock.submit.assert_not_called()
+        # Verify error response uses normalized function name
+        lambda_error_responses_mock.resource_not_found.assert_called_once_with("NonExistentFunction")
+
+    @patch("samcli.local.lambda_service.local_lambda_http_service.LOG")
+    @patch("samcli.local.lambda_service.local_lambda_http_service.ThreadPoolExecutor")
+    @patch("samcli.local.lambda_service.local_lambda_http_service.LambdaErrorResponses")
+    def test_invoke_request_handler_async_invocation_invalid_function_name_returns_error(
+        self, lambda_error_responses_mock, executor_class_mock, log_mock
+    ):
+        # Test that async invocation (Event type) returns error when function name is invalid
+        executor_mock = Mock()
+        executor_class_mock.return_value = executor_mock
+
+        request_mock = Mock()
+        request_mock.get_data.return_value = b"{}"
+        request_mock.args = {}
+        request_mock.headers = {"X-Amz-Invocation-Type": "Event"}
+        local_lambda_http_service.request = request_mock
+
+        lambda_runner_mock = Mock()
+        test_exception = InvalidFunctionNameException("Invalid function name format")
+        lambda_runner_mock.get_function.side_effect = test_exception
+        lambda_error_responses_mock.validation_exception.return_value = "validation error response"
+
+        service = LocalLambdaHttpService(lambda_runner=lambda_runner_mock, port=3000, host="localhost")
+
+        response = service._invoke_request_handler(function_name="Invalid@Function#Name")
+
+        # Verify error response is returned
+        self.assertEqual(response, "validation error response")
+        # Verify get_function was called to validate
+        lambda_runner_mock.get_function.assert_called_once_with("Invalid@Function#Name", None)
+        # Verify executor was NOT called since validation failed
+        executor_mock.submit.assert_not_called()
+        # Verify error was logged
+        log_mock.error.assert_called_once_with("Validation error: %s", "Invalid function name format")
+        # Verify validation exception was called with the error message
+        lambda_error_responses_mock.validation_exception.assert_called_once_with("Invalid function name format")
+
 
 class TestValidateInvokeRequestHandling(TestCase):
     def setUp(self):
@@ -558,11 +730,9 @@ def test_non_invoke_endpoint_not_validated(self, request_mock):
 
         self.assertIsNone(response)
 
-    @patch("samcli.local.lambda_service.local_lambda_http_service.normalize_sam_function_identifier")
     @patch("samcli.local.lambda_service.local_lambda_http_service.LambdaErrorResponses")
-    def test_invoke_request_handler_invalid_function_name(self, error_responses_mock, normalize_mock):
-        # Setup mocks
-        normalize_mock.side_effect = InvalidFunctionNameException("Invalid function name")
+    def test_invoke_request_handler_invalid_function_name(self, error_responses_mock):
+        # Setup mocks - get_function is called first and raises InvalidFunctionNameException
         error_responses_mock.validation_exception.return_value = "validation exception response"
 
         request_mock = Mock()
@@ -572,11 +742,13 @@ def test_invoke_request_handler_invalid_function_name(self, error_responses_mock
         local_lambda_http_service.request = request_mock
 
         lambda_runner_mock = Mock()
+        lambda_runner_mock.get_function.side_effect = InvalidFunctionNameException("Invalid function name")
         service = LocalLambdaHttpService(lambda_runner=lambda_runner_mock, port=3000, host="localhost")
 
         response = service._invoke_request_handler("invalid-function-name")
 
         self.assertEqual(response, "validation exception response")
+        lambda_runner_mock.get_function.assert_called_once_with("invalid-function-name", None)
         error_responses_mock.validation_exception.assert_called_once_with("Invalid function name")
 
 
@@ -635,11 +807,13 @@ def test_invoke_request_handler_with_arn(self, lambda_output_parser_mock, servic
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         service_response_mock.assert_called_once_with("hello world", {"Content-Type": "application/json"}, 200)
 
+    @patch("samcli.local.lambda_service.local_lambda_http_service.normalize_sam_function_identifier")
     @patch("samcli.local.lambda_service.local_lambda_http_service.LambdaErrorResponses")
-    def test_invoke_request_handler_function_not_found_with_arn(self, lambda_error_responses_mock):
+    def test_invoke_request_handler_function_not_found_with_arn(self, lambda_error_responses_mock, normalize_mock):
         """Test that error handling uses normalized function name when ARN is provided"""
         request_mock = Mock()
         request_mock.get_data.return_value = b"{}"
@@ -648,7 +822,8 @@ def test_invoke_request_handler_function_not_found_with_arn(self, lambda_error_r
         local_lambda_http_service.request = request_mock
 
         lambda_runner_mock = Mock()
-        lambda_runner_mock.invoke.side_effect = FunctionNotFound
+        lambda_runner_mock.get_function.side_effect = FunctionNotFound
+        normalize_mock.return_value = "NotFound"
 
         lambda_error_responses_mock.resource_not_found.return_value = "Couldn't find Lambda"
 
@@ -660,16 +835,9 @@ def test_invoke_request_handler_function_not_found_with_arn(self, lambda_error_r
 
         self.assertEqual(response, "Couldn't find Lambda")
 
-        # Verify that the lambda runner was called with the normalized function name
-        lambda_runner_mock.invoke.assert_called_once_with(
-            "NotFound",
-            "{}",
-            invocation_type="RequestResponse",
-            durable_execution_name=None,
-            tenant_id=None,
-            stdout=ANY,
-            stderr=None,
-        )
+        # get_function is called first with the ARN; invoke is never called when it raises
+        lambda_runner_mock.get_function.assert_called_once_with(arn, None)
+        lambda_runner_mock.invoke.assert_not_called()
 
         # Verify that error response uses the normalized function name
         lambda_error_responses_mock.resource_not_found.assert_called_once_with("NotFound")
@@ -1132,6 +1300,7 @@ def test_invoke_request_handler_combines_headers_with_durable_execution_arn(
             tenant_id=None,
             stdout=ANY,
             stderr=None,
+            function=ANY,
         )
         expected_headers = {
             "Content-Type": "application/json",
diff --git a/tests/unit/local/lambdafn/test_runtime.py b/tests/unit/local/lambdafn/test_runtime.py
index 0e77b66505..c17a271301 100644
--- a/tests/unit/local/lambdafn/test_runtime.py
+++ b/tests/unit/local/lambdafn/test_runtime.py
@@ -20,6 +20,7 @@
 from samcli.local.docker.container import ContainerContext
 from samcli.commands.local.lib.debug_context import DebugContext
 from samcli.local.docker.durable_lambda_container import DurableLambdaContainer
+from samcli.local.lambdafn.exceptions import UnsupportedInvocationType
 
 
 class LambdaRuntime_create(TestCase):
@@ -664,6 +665,62 @@ def test_durable_execution_calls_wait_for_result_and_skips_cleanup(self):
         self.assertEqual(headers["X-Amz-Durable-Execution-Arn"], "test-arn")
         self.runtime._check_exit_state.assert_called_with(container)
 
+    @patch("samcli.local.lambdafn.runtime.LambdaContainer")
+    def test_unsupported_invocation_type_raises_exception(self, LambdaContainerMock):
+        """Test that unsupported invocation types raise UnsupportedInvocationType for regular Lambda functions"""
+        event = "event"
+        code_dir = "some code dir"
+        stdout = "stdout"
+        stderr = "stderr"
+        container = Mock()
+        start_timer = Mock()
+        debug_options = Mock()
+        lambda_image_mock = Mock()
+        unsupported_invocation_type = "DryRun"  # An unsupported invocation type
+
+        self.runtime = LambdaRuntime(self.manager_mock, lambda_image_mock)
+
+        # Using MagicMock to mock the context manager
+        self.runtime._get_code_dir = MagicMock()
+        self.runtime._get_code_dir.return_value = code_dir
+
+        self.runtime._clean_decompressed_paths = MagicMock()
+
+        # Configure interrupt handler
+        self.runtime._configure_interrupt = Mock()
+        self.runtime._configure_interrupt.return_value = start_timer
+
+        self.runtime._check_exit_state = Mock()
+
+        # Mock create and run to return the container
+        self.runtime.create = Mock(return_value=container)
+        self.runtime.run = Mock(return_value=container)
+
+        LambdaContainerMock.return_value = container
+        container.is_running.return_value = False
+
+        # Regular LambdaContainer (not DurableLambdaContainer) should raise exception for unsupported types
+        with self.assertRaises(UnsupportedInvocationType) as context:
+            self.runtime.invoke(
+                self.func_config,
+                event,
+                debug_context=debug_options,
+                stdout=stdout,
+                stderr=stderr,
+                invocation_type=unsupported_invocation_type,
+            )
+
+        # Verify the exception message
+        self.assertIn("invocation-type: DryRun is not supported", str(context.exception))
+        self.assertIn("Only Event and RequestResponse are supported", str(context.exception))
+
+        # Verify that wait_for_result was not called due to the exception
+        container.wait_for_result.assert_not_called()
+
+        # Verify cleanup was still called
+        self.manager_mock.stop.assert_called_with(container)
+        self.runtime._clean_decompressed_paths.assert_called_with()
+
 
 class TestLambdaRuntime_configure_interrupt(TestCase):
     def setUp(self):
@@ -2255,3 +2312,37 @@ def test_should_reload_when_both_config_and_debug_context_changed(self):
 
         result = _should_reload_container(self.base_config, self.different_config, container, debug_context2)
         self.assertTrue(result)
+
+
+class TestLambdaRuntime_clean_runtime_containers(TestCase):
+    @patch("samcli.local.lambdafn.runtime.LOG")
+    def test_clean_runtime_containers_stops_and_deletes_durable_container(self, log_mock):
+        """Test that clean_runtime_containers stops and deletes durable lambda container"""
+        manager_mock = Mock()
+        lambda_image_mock = Mock()
+        runtime = LambdaRuntime(manager_mock, lambda_image_mock)
+
+        container = Mock(spec=DurableLambdaContainer)
+        runtime._container = container
+
+        runtime.clean_runtime_containers()
+
+        container._stop.assert_called_once()
+        container._delete.assert_called_once()
+        self.assertIsNone(runtime._container)
+
+    @patch("samcli.local.lambdafn.runtime.LOG")
+    def test_clean_runtime_containers_stops_emulator_container(self, log_mock):
+        """Test that clean_runtime_containers stops and cleans up emulator container"""
+        manager_mock = Mock()
+        lambda_image_mock = Mock()
+        runtime = LambdaRuntime(manager_mock, lambda_image_mock)
+
+        emulator_container = Mock()
+        runtime._durable_execution_emulator_container = emulator_container
+
+        runtime.clean_runtime_containers()
+
+        emulator_container.stop.assert_called_once()
+        log_mock.debug.assert_called_with("Stopping durable functions emulator container")
+        self.assertIsNone(runtime._durable_execution_emulator_container)
diff --git a/tests/unit/local/services/test_base_local_service.py b/tests/unit/local/services/test_base_local_service.py
index b1513177c5..fc7ea58547 100644
--- a/tests/unit/local/services/test_base_local_service.py
+++ b/tests/unit/local/services/test_base_local_service.py
@@ -54,7 +54,7 @@ def test_service_response(self, flask_response_patch):
         flask_response_patch.assert_called_once_with("this is the body")
 
         self.assertEqual(actual_response.status_code, 200)
-        self.assertEqual(actual_response.headers, {"Content-Type": "application/json"})
+        flask_response_mock.headers.update.assert_called_once_with({"Content-Type": "application/json"})
 
     @patch("samcli.local.services.base_local_service.signal.signal")
     def test_service_registers_sigterm_interrupt(self, signal_mock):