Merge from aws/aws-sam-cli/develop

Author: aws-sam-cli-bot

samcli/__init__.py

@@ -2,4 +2,4 @@
 SAM CLI version
 """
 
-__version__ = "1.150.0"
+__version__ = "1.150.1"

samcli/lib/clients/lambda_client.py

@@ -53,11 +53,15 @@ def create(cls, host: str = "localhost", port: int = 5000, region: str = "us-wes
             # Create a fresh botocore session
             session = botocore.session.Session()
 
-            # Create the boto3 client with custom service model using the fresh session
+            # Create the boto3 client using the fresh session
             client = session.create_client(
                 "lambda",
                 endpoint_url=endpoint_url,
                 region_name=region,
+                # the emulator doesnt access any AWS resources,
+                # but we need _some_ credentials to create a boto client
+                aws_access_key_id="foo",
+                aws_secret_access_key="bar",
             )
 
             return cls(client)

samcli/lib/utils/durable_callback_handler.py

@@ -0,0 +1,137 @@
+"""
+Interactive callback handler for durable function executions.
+"""
+
+import logging
+from typing import Optional
+
+import click
+
+from samcli.lib.clients.lambda_client import DurableFunctionsClient
+
+LOG = logging.getLogger(__name__)
+
+# Menu choice constants
+CHOICE_SUCCESS = 1
+CHOICE_FAILURE = 2
+CHOICE_HEARTBEAT = 3
+CHOICE_STOP = 4
+
+
+class DurableCallbackHandler:
+    """
+    Handles interactive callback detection and response for durable executions.
+    """
+
+    def __init__(self, client: DurableFunctionsClient):
+        self.client = client
+        self._prompted_callbacks: set[str] = set()  # Track which callbacks we've already prompted for
+
+    def check_for_pending_callbacks(self, execution_arn: str) -> Optional[str]:
+        """
+        Check execution history for pending callbacks.
+
+        Returns:
+            callback_id if found, None otherwise
+        """
+        try:
+            LOG.debug("Checking for pending callbacks in execution: %s", execution_arn)
+            history = self.client.get_durable_execution_history(execution_arn)
+            events = history.get("Events", [])
+
+            if events:
+                callback_states = {}
+
+                for event in events:
+                    event_type = event.get("EventType")
+                    event_id = event.get("Id")
+
+                    if event_type == "CallbackStarted":
+                        callback_id = event.get("CallbackStartedDetails", {}).get("CallbackId")
+                        callback_states[event_id] = {"callback_id": callback_id, "status": "STARTED", "event": event}
+                    elif event_type in ["CallbackCompleted", "CallbackFailed", "CallbackSucceeded"]:
+                        if event_id in callback_states:
+                            callback_states[event_id]["status"] = "COMPLETED"
+
+                # Find callbacks that are started but not completed
+                for callback_id, state in callback_states.items():
+                    if state["status"] == "STARTED" and state["callback_id"]:
+                        return str(state["callback_id"])
+
+        except Exception as e:
+            LOG.error("Failed to check callback history: %s", e)
+
+        return None
+
+    def prompt_callback_response(self, execution_arn: str, callback_id: str, execution_complete=None) -> bool:
+        """
+        Prompt user for callback response and send it.
+
+        Args:
+            execution_arn: The execution ARN for stop execution operation
+            callback_id: The callback ID to respond to
+            execution_complete: Optional threading.Event to check if execution finished
+
+        Returns:
+            True if callback was sent, False if user chose to continue waiting
+        """
+        # Only prompt once per callback ID to avoid blocking on timed-out callbacks
+        if callback_id in self._prompted_callbacks:
+            return False
+
+        self._prompted_callbacks.add(callback_id)
+
+        # Check if execution already completed before prompting
+        if execution_complete and execution_complete.is_set():
+            return False
+
+        click.echo(f"\n🔄 Execution is waiting for callback: {callback_id}")
+        click.echo("Choose an action:")
+        click.echo("  1. Send callback success")
+        click.echo("  2. Send callback failure")
+        click.echo("  3. Send callback heartbeat")
+        click.echo("  4. Stop execution")
+
+        choice = click.prompt("Enter choice", type=click.IntRange(1, 4), default=CHOICE_SUCCESS)
+
+        # Check again after user makes selection in case execution completed
+        if execution_complete and execution_complete.is_set():
+            click.echo("⚠️  Execution already completed, callback no longer needed")
+            return False
+
+        try:
+            if choice == CHOICE_SUCCESS:
+                result = click.prompt("Enter success result (optional)", default="", show_default=False)
+                self.client.send_callback_success(callback_id=callback_id, result=result)
+                click.echo("✅ Callback success sent")
+                return True
+
+            elif choice == CHOICE_FAILURE:
+                error_message = click.prompt("Enter error message", default="User cancelled")
+                error_type = click.prompt("Enter error type (optional)", default="", show_default=False) or None
+
+                self.client.send_callback_failure(
+                    callback_id=callback_id, error_message=error_message, error_type=error_type
+                )
+                click.echo("❌ Callback failure sent")
+                return True
+
+            elif choice == CHOICE_HEARTBEAT:
+                self.client.send_callback_heartbeat(callback_id=callback_id)
+                click.echo("💓 Callback heartbeat sent")
+                return False  # Continue waiting after heartbeat
+
+            else:  # CHOICE_STOP
+                error_message = click.prompt("Enter error message", default="Execution stopped by user")
+                error_type = click.prompt("Enter error type (optional)", default="", show_default=False) or None
+
+                self.client.stop_durable_execution(
+                    durable_execution_arn=execution_arn, error_message=error_message, error_type=error_type
+                )
+                click.echo("🛑 Execution stopped")
+                return True
+
+        except Exception as e:
+            LOG.error("Failed to send callback: %s", e)
+            click.echo(f"❌ Failed to send callback: {e}")
+            return False

samcli/local/docker/durable_functions_emulator_container.py

@@ -7,6 +7,7 @@
 import time
 from http import HTTPStatus
 from pathlib import Path
+from tempfile import NamedTemporaryFile
 from typing import Optional
 
 import docker
@@ -15,7 +16,8 @@
 
 from samcli.lib.build.utils import _get_host_architecture
 from samcli.lib.clients.lambda_client import DurableFunctionsClient
-from samcli.local.docker.utils import get_validated_container_client, is_image_current
+from samcli.lib.utils.tar import create_tarball
+from samcli.local.docker.utils import get_tar_filter_for_windows, get_validated_container_client, is_image_current
 
 LOG = logging.getLogger(__name__)
 
@@ -27,6 +29,7 @@ class DurableFunctionsEmulatorContainer:
 
     _RAPID_SOURCE_PATH = Path(__file__).parent.joinpath("..", "rapid").resolve()
     _EMULATOR_IMAGE = "public.ecr.aws/ubuntu/ubuntu:24.04"
+    _EMULATOR_IMAGE_PREFIX = "samcli/durable-execution-emulator"
     _CONTAINER_NAME = "sam-durable-execution-emulator"
     _EMULATOR_DATA_DIR_NAME = ".durable-executions-local"
     _EMULATOR_DEFAULT_STORE_TYPE = "sqlite"
@@ -190,6 +193,62 @@ def _get_emulator_binary_name(self):
         arch = _get_host_architecture()
         return f"aws-durable-execution-emulator-{arch}"
 
+    def _generate_emulator_dockerfile(self, emulator_binary_name: str) -> str:
+        """Generate Dockerfile content for emulator image."""
+        return (
+            f"FROM {self._EMULATOR_IMAGE}\n"
+            f"COPY {emulator_binary_name} /usr/local/bin/{emulator_binary_name}\n"
+            f"RUN chmod +x /usr/local/bin/{emulator_binary_name}\n"
+        )
+
+    def _get_emulator_image_tag(self, emulator_binary_name: str) -> str:
+        """Get the Docker image tag for the emulator."""
+        return f"{self._EMULATOR_IMAGE_PREFIX}:{emulator_binary_name}"
+
+    def _build_emulator_image(self):
+        """Build Docker image with emulator binary."""
+        emulator_binary_name = self._get_emulator_binary_name()
+        binary_path = self._RAPID_SOURCE_PATH / emulator_binary_name
+
+        if not binary_path.exists():
+            raise RuntimeError(f"Durable Functions Emulator binary not found at {binary_path}")
+
+        image_tag = self._get_emulator_image_tag(emulator_binary_name)
+
+        # Check if image already exists
+        try:
+            self._docker_client.images.get(image_tag)
+            LOG.debug(f"Emulator image {image_tag} already exists")
+            return image_tag
+        except docker.errors.ImageNotFound:
+            LOG.debug(f"Building emulator image {image_tag}")
+
+        # Generate Dockerfile content
+        dockerfile_content = self._generate_emulator_dockerfile(emulator_binary_name)
+
+        # Write Dockerfile to temp location and build image
+        with NamedTemporaryFile(mode="w", suffix="_Dockerfile") as dockerfile:
+            dockerfile.write(dockerfile_content)
+            dockerfile.flush()
+
+            # Prepare tar paths for build context
+            tar_paths = {
+                dockerfile.name: "Dockerfile",
+                str(binary_path): emulator_binary_name,
+            }
+
+            # Use shared tar filter for Windows compatibility
+            tar_filter = get_tar_filter_for_windows()
+
+            # Build image using create_tarball utility
+            with create_tarball(tar_paths, tar_filter=tar_filter, dereference=True) as tarballfile:
+                try:
+                    self._docker_client.images.build(fileobj=tarballfile, custom_context=True, tag=image_tag, rm=True)
+                    LOG.info(f"Built emulator image {image_tag}")
+                    return image_tag
+                except Exception as e:
+                    raise ClickException(f"Failed to build emulator image: {e}")
+
     def _pull_image_if_needed(self):
         """Pull the emulator image if it doesn't exist locally or is out of date."""
         try:
@@ -218,9 +277,6 @@ def start(self):
             return
 
         emulator_binary_name = self._get_emulator_binary_name()
-        binary_path = self._RAPID_SOURCE_PATH / emulator_binary_name
-        if not binary_path.exists():
-            raise RuntimeError(f"Durable Functions Emulator binary not found at {binary_path}")
 
         """
         Create persistent volume for execution data to be stored in.
@@ -231,16 +287,15 @@ def start(self):
         os.makedirs(emulator_data_dir, exist_ok=True)
 
         volumes = {
-            str(self._RAPID_SOURCE_PATH): {"bind": "/usr/local/bin", "mode": "ro"},
             emulator_data_dir: {"bind": "/tmp/.durable-executions-local", "mode": "rw"},
         }
 
-        # Pull the image if needed
-        self._pull_image_if_needed()
+        # Build image with emulator binary
+        image_tag = self._build_emulator_image()
 
         LOG.debug(f"Creating container with name={self._container_name}, port={self.port}")
         self.container = self._docker_client.containers.create(
-            image=self._EMULATOR_IMAGE,
+            image=image_tag,
             command=[f"/usr/local/bin/{emulator_binary_name}", "--host", "0.0.0.0", "--port", str(self.port)],
             name=self._container_name,
             ports={f"{self.port}/tcp": self.port},

samcli/local/docker/durable_lambda_container.py

@@ -9,6 +9,7 @@
 import click
 from flask import has_request_context
 
+from samcli.lib.utils.durable_callback_handler import DurableCallbackHandler
 from samcli.lib.utils.durable_formatters import format_execution_details, format_next_commands_after_invoke
 from samcli.local.docker.lambda_container import LambdaContainer
 
@@ -58,6 +59,9 @@ def _update_lambda_environment_with_emulator_endpoint(self, kwargs):
         extra_hosts["host.docker.internal"] = "host-gateway"
         kwargs["extra_hosts"] = extra_hosts
 
+        # Bind to 0.0.0.0 so emulator can reach Lambda via host.docker.internal
+        kwargs["container_host_interface"] = "0.0.0.0"
+
     def _get_lambda_container_endpoint(self):
         """
         Get the Lambda container endpoint URL for the emulator to invoke.
@@ -146,8 +150,11 @@ def _write_execution_result_to_stdout(self, execution_details: dict, stdout):
     def _wait_for_execution(self, execution_arn):
         """Poll the execution status until completion and return the final result."""
 
-        # TODO - poll until the execution timeout is hit
+        callback_handler = DurableCallbackHandler(self.emulator_container.lambda_client)
         execution_details = None
+        callback_thread = None
+        stop_callback_prompts = threading.Event()
+
         try:
             while True:
                 try:
@@ -156,13 +163,36 @@ def _wait_for_execution(self, execution_arn):
                     status = execution_details.get("Status")
 
                     if status != "RUNNING":
+                        stop_callback_prompts.set()  # Signal callback thread to stop
+                        if callback_thread and callback_thread.is_alive():
+                            callback_thread.join(timeout=0.5)  # Brief wait for thread cleanup
                         return execution_details
 
+                    # Check for pending callbacks (only in CLI context)
+                    if self._is_cli_context():
+                        callback_id = callback_handler.check_for_pending_callbacks(execution_arn)
+                        if callback_id:
+
+                            def _prompt_in_thread():
+                                if not stop_callback_prompts.is_set():
+                                    # give the function logs time to settle after the invocation is suspended
+                                    time.sleep(0.5)
+                                    callback_sent = callback_handler.prompt_callback_response(
+                                        execution_arn, callback_id, stop_callback_prompts
+                                    )
+                                    if callback_sent:
+                                        click.echo("\n" + "─" * 80)
+
+                            # Start callback prompt in separate thread so it doesn't block polling
+                            callback_thread = threading.Thread(target=_prompt_in_thread, daemon=True)
+                            callback_thread.start()
+
                     time.sleep(1)  # Poll every second
                 except Exception as e:
                     LOG.error("Error polling execution status: %s", e)
                     break
         finally:
+            stop_callback_prompts.set()  # Ensure callback thread knows to stop
             self._cleanup_if_needed()
 
         return execution_details

samcli/local/docker/lambda_image.py

@@ -26,7 +26,12 @@
 from samcli.lib.utils.stream_writer import StreamWriter
 from samcli.lib.utils.tar import create_tarball
 from samcli.local.common.file_lock import FileLock, cleanup_stale_locks
-from samcli.local.docker.utils import get_docker_platform, get_rapid_name, get_validated_container_client
+from samcli.local.docker.utils import (
+    get_docker_platform,
+    get_rapid_name,
+    get_tar_filter_for_windows,
+    get_validated_container_client,
+)
 
 LOG = logging.getLogger(__name__)
 
@@ -409,15 +414,8 @@ def _build_image(self, base_image, docker_tag, layers, architecture, stream=None
             for layer in layers:
                 tar_paths[layer.codeuri] = "/" + layer.name
 
-            # Set permission for all the files in the tarball to 500(Read and Execute Only)
-            # This is need for systems without unix like permission bits(Windows) while creating a unix image
-            # Without setting this explicitly, tar will default the permission to 666 which gives no execute permission
-            def set_item_permission(tar_info):
-                tar_info.mode = 0o500
-                return tar_info
-
-            # Set only on Windows, unix systems will preserve the host permission into the tarball
-            tar_filter = set_item_permission if platform.system().lower() == "windows" else None
+            # Use shared tar filter for Windows compatibility
+            tar_filter = get_tar_filter_for_windows()
 
             with create_tarball(tar_paths, tar_filter=tar_filter, dereference=True) as tarballfile:
                 try: