perf: replace defensive deepcopy with deep_freeze immutability

expand_language_extensions was deep-copying the template 4-10 times per
invocation for defensive isolation (cache put, cache get, no-LE path,
LE path). For large templates this is measurable overhead.

Replace with a one-time deep_freeze that converts dicts to
MappingProxyType and lists to tuples. Frozen templates are truly
immutable — any mutation attempt raises TypeError immediately instead
of silently corrupting shared state. Callers that need mutable copies
use deep_thaw (not copy.deepcopy which cannot pickle MappingProxyType
on Python 3.13).

Savings:
- No-LE fast path: 4 deepcopies -> 1 deep_freeze (same O(n) cost)
- Cache hit: 2 deepcopies -> 0 (return cached frozen result directly)
- Cold LE path: 3 deepcopies -> 2 deep_freezes (expanded + original)
- Cache put: 2 deepcopies -> 0 (frozen values are safe to share)

Also:
- Fixed isinstance(x, dict) check in build_context to accept Mapping
  so frozen MappingProxyType templates pass the type check.
- Updated callers (build_context, package_context, artifact_exporter,
  language_extensions_packaging, wrapper) to use deep_thaw instead of
  copy.deepcopy when they need mutable copies.
- Updated tests to verify immutability contract (TypeError on mutation)
  instead of asserting independent mutable copies.

Author: bnusunny

samcli/commands/build/build_context.py

@@ -7,6 +7,7 @@
 import pathlib
 import shutil
 from collections import Counter
+from collections.abc import Mapping
 from typing import Any, Dict, List, Optional, Tuple
 
 import click
@@ -40,6 +41,7 @@
     sanitize_resource_key_for_mapping,
     substitute_loop_variable,
 )
+from samcli.lib.cfn_language_extensions.utils import deep_thaw
 from samcli.lib.intrinsic_resolver.intrinsics_symbol_table import IntrinsicsSymbolTable
 from samcli.lib.providers.provider import LayerVersion, ResourcesToBuildCollector, Stack
 from samcli.lib.providers.sam_api_provider import SamApiProvider
@@ -442,17 +444,16 @@ def _get_template_for_output(self, stack: Stack, modified_template: Dict, artifa
         Dict
             The template to write to disk
         """
-        import copy
 
         # If no original template, use the modified (expanded) template
         # Check if original_template_dict exists and is a dict (not a Mock or other type)
         original_template_dict = getattr(stack, "original_template_dict", None)
-        if not isinstance(original_template_dict, dict):
+        if not isinstance(original_template_dict, (dict, Mapping)):
             return modified_template
 
         # Use the original template (with Fn::ForEach intact)
         # We need to update artifact paths in the Fn::ForEach constructs
-        original_template = copy.deepcopy(original_template_dict)
+        original_template = deep_thaw(original_template_dict)
 
         # Update artifact paths in the original template
         self._update_original_template_paths(original_template, modified_template, stack)

samcli/commands/package/package_context.py

@@ -171,6 +171,7 @@ def run(self):
     def _export(self, template_path, use_json):
         from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException
         from samcli.lib.cfn_language_extensions.sam_integration import expand_language_extensions
+        from samcli.lib.cfn_language_extensions.utils import deep_thaw
 
         # Read the original template
         with open(template_path, "r", encoding="utf-8") as f:
@@ -192,7 +193,7 @@ def _export(self, template_path, use_json):
 
         uses_language_extensions = result.had_language_extensions
         dynamic_properties = result.dynamic_artifact_properties
-        template_dict_for_export = result.expanded_template
+        template_dict_for_export = deep_thaw(result.expanded_template)
 
         # Create Template with the (possibly expanded) template
         template = Template(

samcli/lib/cfn_language_extensions/sam_integration.py

@@ -27,7 +27,12 @@
     ResolutionMode,
     TemplateProcessingContext,
 )
-from samcli.lib.cfn_language_extensions.utils import FOREACH_PREFIX, FOREACH_REQUIRED_ELEMENTS, is_foreach_key
+from samcli.lib.cfn_language_extensions.utils import (
+    FOREACH_PREFIX,
+    FOREACH_REQUIRED_ELEMENTS,
+    deep_freeze,
+    is_foreach_key,
+)
 
 LOG = logging.getLogger(__name__)
 
@@ -77,6 +82,11 @@ class LanguageExtensionResult:
     This dataclass carries all Phase 1 outputs so that callers don't need
     to re-derive them.
 
+    All template fields are deeply frozen (dicts as ``MappingProxyType``,
+    lists as ``tuple``).  Callers that need to mutate must use
+    ``deep_thaw()`` from ``samcli.lib.cfn_language_extensions.utils``
+    (not ``copy.deepcopy`` which cannot pickle ``MappingProxyType``).
+
     Attributes
     ----------
     expanded_template : Dict[str, Any]
@@ -552,31 +562,18 @@ def expand_language_extensions(
         cached = _expansion_cache.get(cache_key)
         if cached is not None:
             LOG.debug("Cache hit for template expansion: %s", template_path)
-            return LanguageExtensionResult(
-                expanded_template=copy.deepcopy(cached.expanded_template),
-                original_template=copy.deepcopy(cached.original_template),
-                dynamic_artifact_properties=list(cached.dynamic_artifact_properties),
-                had_language_extensions=cached.had_language_extensions,
-            )
+            return cached
 
     if not check_using_language_extension(template):
+        frozen = deep_freeze(template)
         result = LanguageExtensionResult(
-            expanded_template=copy.deepcopy(template),
-            original_template=copy.deepcopy(template),
+            expanded_template=frozen,
+            original_template=frozen,
             dynamic_artifact_properties=[],
             had_language_extensions=False,
         )
         if cache_key is not None:
-            # Store a deep copy so callers can't poison the cache
-            _cache_put(
-                cache_key,
-                LanguageExtensionResult(
-                    expanded_template=copy.deepcopy(result.expanded_template),
-                    original_template=copy.deepcopy(result.original_template),
-                    dynamic_artifact_properties=list(result.dynamic_artifact_properties),
-                    had_language_extensions=result.had_language_extensions,
-                ),
-            )
+            _cache_put(cache_key, result)
         return result
 
     LOG.debug("Expanding CloudFormation Language Extensions (Phase 1)")
@@ -599,8 +596,8 @@ def expand_language_extensions(
         LOG.debug("Successfully expanded CloudFormation Language Extensions")
 
         result = LanguageExtensionResult(
-            expanded_template=expanded_template,
-            original_template=copy.deepcopy(template),
+            expanded_template=deep_freeze(expanded_template),
+            original_template=deep_freeze(template),
             dynamic_artifact_properties=dynamic_properties,
             had_language_extensions=True,
         )
@@ -611,16 +608,7 @@ def expand_language_extensions(
         EventTracker.track_event(EventName.USED_FEATURE.value, UsedFeature.CFN_LANGUAGE_EXTENSIONS.value)
 
         if cache_key is not None:
-            # Store a deep copy so callers can't poison the cache
-            _cache_put(
-                cache_key,
-                LanguageExtensionResult(
-                    expanded_template=copy.deepcopy(result.expanded_template),
-                    original_template=copy.deepcopy(result.original_template),
-                    dynamic_artifact_properties=list(result.dynamic_artifact_properties),
-                    had_language_extensions=result.had_language_extensions,
-                ),
-            )
+            _cache_put(cache_key, result)
 
         return result
 

samcli/lib/cfn_language_extensions/utils.py

@@ -5,7 +5,8 @@
 for working with templates that may contain Fn::ForEach blocks.
 """
 
-from typing import Dict, Iterator, Tuple
+from types import MappingProxyType
+from typing import Any, Dict, Iterator, Tuple
 
 FOREACH_PREFIX = "Fn::ForEach::"
 
@@ -110,3 +111,42 @@ def iter_regular_resources(template_dict: Dict) -> Iterator[Tuple[str, Dict]]:
     for key, value in template_dict.get("Resources", {}).items():
         if not is_foreach_key(key) and isinstance(value, dict):
             yield key, value
+
+
+def deep_freeze(obj: Any) -> Any:
+    """Recursively make a template structure immutable.
+
+    - dicts become ``MappingProxyType`` (read-only view)
+    - lists become ``tuple`` (immutable sequence)
+    - primitives (str, int, float, bool, None) pass through unchanged
+
+    Any caller that tries to mutate a frozen template gets an immediate
+    ``TypeError`` instead of silently corrupting shared state.  Callers
+    that need a mutable copy should use ``deep_thaw()`` (not
+    ``copy.deepcopy`` which cannot pickle ``MappingProxyType``).
+
+    Cost is O(n) — same as one ``copy.deepcopy`` — but you pay it once
+    at creation time and then never need defensive copies again.
+    """
+    if isinstance(obj, MappingProxyType):
+        return obj  # already frozen
+    if isinstance(obj, dict):
+        return MappingProxyType({k: deep_freeze(v) for k, v in obj.items()})
+    if isinstance(obj, list):
+        return tuple(deep_freeze(item) for item in obj)
+    return obj
+
+
+def deep_thaw(obj: Any) -> Any:
+    """Recursively convert a frozen template back to mutable dicts and lists.
+
+    Inverse of ``deep_freeze``.  Use this instead of ``copy.deepcopy``
+    on frozen templates (``MappingProxyType`` is not picklable).
+    """
+    if isinstance(obj, MappingProxyType):
+        return {k: deep_thaw(v) for k, v in obj.items()}
+    if isinstance(obj, dict):
+        return {k: deep_thaw(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [deep_thaw(item) for item in obj]
+    return obj

samcli/lib/package/artifact_exporter.py

@@ -23,7 +23,7 @@
 from samcli.commands._utils.experimental import ExperimentalFlag, is_experimental_enabled
 from samcli.commands.package import exceptions
 from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException
-from samcli.lib.cfn_language_extensions.utils import iter_regular_resources
+from samcli.lib.cfn_language_extensions.utils import deep_thaw, iter_regular_resources
 from samcli.lib.package.code_signer import CodeSigner
 from samcli.lib.package.local_files_utils import get_uploaded_s3_object_name, mktempfile
 from samcli.lib.package.packageable_resources import (
@@ -222,7 +222,7 @@ def do_export(self, resource_id, resource_dict, parent_dir):
                 normalize_template=True,
                 normalize_parameters=True,
                 parent_stack_id=resource_id,
-                template_str=yaml_dump(result.expanded_template),
+                template_str=yaml_dump(deep_thaw(result.expanded_template)),
                 parameter_values=parameter_values,
             )
             template.template_dir = child_template_dir

samcli/lib/package/language_extensions_packaging.py

@@ -6,7 +6,6 @@
 None of the functions use instance state — they are pure functions.
 """
 
-import copy
 import itertools
 import logging
 import re
@@ -24,7 +23,7 @@
     sanitize_resource_key_for_mapping,
     substitute_loop_variable,
 )
-from samcli.lib.cfn_language_extensions.utils import FOREACH_REQUIRED_ELEMENTS, is_foreach_key
+from samcli.lib.cfn_language_extensions.utils import FOREACH_REQUIRED_ELEMENTS, deep_thaw, is_foreach_key
 
 LOG = logging.getLogger(__name__)
 
@@ -66,7 +65,7 @@ def merge_language_extensions_s3_uris(
     dict
         The original template with updated S3 URIs
     """
-    result = copy.deepcopy(original_template)
+    result = deep_thaw(original_template)
 
     # Build a set of (foreach_key, property_name) tuples for dynamic properties
     dynamic_prop_keys: set = set()

samcli/lib/samlib/wrapper.py

@@ -27,6 +27,7 @@
 from samcli.lib.cfn_language_extensions.models import (
     DynamicArtifactProperty,
 )
+from samcli.lib.cfn_language_extensions.utils import deep_thaw
 from samcli.lib.samlib.local_uri_plugin import SupportLocalUriPlugin
 
 LOG = logging.getLogger(__name__)
@@ -122,7 +123,7 @@ def get_original_template(self) -> Dict[str, Any]:
         dict
             A deep copy of the original template with language extensions preserved
         """
-        return copy.deepcopy(self._original_template)
+        return deep_thaw(self._original_template)
 
     def get_dynamic_artifact_properties(self) -> List[DynamicArtifactProperty]:
         """

tests/unit/lib/cfn_language_extensions/test_phase_separation.py

@@ -844,10 +844,11 @@ def test_result_preserves_original_template_structure(
         assert foreach_key in result.original_template["Resources"]
 
         original_foreach = result.original_template["Resources"][foreach_key]
-        assert isinstance(original_foreach, list)
+        assert isinstance(original_foreach, (list, tuple))
         assert len(original_foreach) == 3
         assert original_foreach[0] == loop_var
-        assert original_foreach[1] == collection
+        # Collection may be a tuple (frozen) or list
+        assert list(original_foreach[1]) == list(collection)
 
     @pytest.mark.parametrize(
         "loop_name, loop_var, collection",

tests/unit/lib/cfn_language_extensions/test_sam_integration.py

@@ -624,16 +624,21 @@ def test_nonexistent_template_path_does_not_error(self):
         result = expand_language_extensions(template, template_path="/nonexistent/path/template.yaml")
         assert result.had_language_extensions is False
 
-    def test_non_language_extension_template_returns_independent_copies(self):
+    def test_non_language_extension_template_returns_frozen_result(self):
         template = {"Resources": {}}
         result = expand_language_extensions(template)
         assert result.had_language_extensions is False
-        # expanded_template should be an independent copy so mutations don't leak
-        assert result.expanded_template is not result.original_template
-        assert result.expanded_template == template
+        # Both fields point to the same frozen object (no-LE path optimization)
+        assert result.expanded_template is result.original_template
+        assert dict(result.expanded_template) == template
+        # Frozen — mutation raises TypeError
+        with pytest.raises(TypeError):
+            result.expanded_template["Resources"] = {}
 
     def test_mutation_does_not_affect_subsequent_calls(self):
-        """Mutating a returned result must not affect subsequent calls."""
+        """Frozen results prevent mutation; callers must deep_thaw first."""
+        from samcli.lib.cfn_language_extensions.utils import deep_thaw
+
         template = {
             "Resources": {
                 "MyStack": {
@@ -643,10 +648,11 @@ def test_mutation_does_not_affect_subsequent_calls(self):
             }
         }
         result1 = expand_language_extensions(template)
-        # Simulate what update_template does: mutate Location in-place
-        result1.expanded_template["Resources"]["MyStack"]["Properties"]["Location"] = "SomeOther/template.yaml"
+        # Must deep_thaw before mutating (matches the documented contract)
+        mutable = deep_thaw(result1.expanded_template)
+        mutable["Resources"]["MyStack"]["Properties"]["Location"] = "SomeOther/template.yaml"
 
-        # A fresh template dict should not be affected
+        # A fresh call should not be affected
         template2 = {
             "Resources": {
                 "MyStack": {
@@ -922,7 +928,7 @@ def test_no_template_path_skips_cache(self):
             assert mock_process.call_count == 2
 
     def test_cache_returns_independent_copies(self):
-        """Mutating a cached result must not affect subsequent cache hits."""
+        """Cached results are frozen — mutation raises TypeError."""
         with tempfile.TemporaryDirectory() as tmp:
             path = self._make_template_file(tmp)
             template = self._lang_ext_template()
@@ -939,11 +945,13 @@ def test_cache_returns_independent_copies(self):
                 return_value=expanded,
             ):
                 result1 = expand_language_extensions(template, template_path=path)
-                # Mutate the first result
-                result1.expanded_template["Resources"]["TopicA"]["Properties"]["DisplayName"] = "mutated"
+                # Frozen — mutation raises TypeError
+                with pytest.raises(TypeError):
+                    result1.expanded_template["Resources"]["TopicA"]["Properties"]["DisplayName"] = "mutated"
 
+                # Cache hit returns the same frozen object
                 result2 = expand_language_extensions(template, template_path=path)
-                # Second result should have the original value
+                assert result2 is result1
                 assert result2.expanded_template["Resources"]["TopicA"]["Properties"]["DisplayName"] == "original"
 
     def test_nonexistent_template_path_skips_cache(self):
@@ -972,8 +980,8 @@ def test_non_language_ext_template_cached(self):
             assert result1.had_language_extensions is False
             assert result2.had_language_extensions is False
             assert len(_expansion_cache) == 1
-            # Ensure independence
-            assert result1.expanded_template is not result2.expanded_template
+            # Cache returns the same frozen object
+            assert result1 is result2
 
     def test_cache_evicts_oldest_when_full(self):
         """Cache should evict the oldest entry when _MAX_CACHE_SIZE is reached."""

tests/unit/lib/samlib/test_wrapper.py

@@ -143,7 +143,9 @@ def test_expand_language_extensions_success(self, mock_process):
         result = expand_language_extensions(input_template, parameter_values={"AWS::Region": "us-east-1"})
 
         self.assertTrue(result.had_language_extensions)
-        self.assertEqual(result.expanded_template, expected_output)
+        from samcli.lib.cfn_language_extensions.utils import deep_thaw
+
+        self.assertEqual(deep_thaw(result.expanded_template), expected_output)
         mock_process.assert_called_once()
 
     @patch("samcli.lib.cfn_language_extensions.sam_integration.process_template_for_sam_cli")