feat: add interactive callback CLI

Author: bchampp

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_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,12 @@ 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
+        LOG.debug("Starting to poll execution status for: %s", execution_arn)
+        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 +164,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