feat: add interactive callback CLI

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
 
@@ -146,8 +147,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 +160,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

tests/unit/lib/utils/test_durable_callback_handler.py

@@ -0,0 +1,197 @@
+"""
+Unit tests for DurableCallbackHandler
+"""
+
+import threading
+from unittest import TestCase
+from unittest.mock import Mock, patch
+
+from parameterized import parameterized
+
+from samcli.lib.utils.durable_callback_handler import (
+    DurableCallbackHandler,
+    CHOICE_SUCCESS,
+    CHOICE_FAILURE,
+    CHOICE_HEARTBEAT,
+    CHOICE_STOP,
+)
+
+
+class TestDurableCallbackHandler(TestCase):
+    def setUp(self):
+        self.mock_client = Mock()
+        self.handler = DurableCallbackHandler(self.mock_client)
+
+    def test_init(self):
+        """Test handler initializes with client and empty prompted callbacks set"""
+        self.assertEqual(self.handler.client, self.mock_client)
+        self.assertEqual(self.handler._prompted_callbacks, set())
+
+    @parameterized.expand(
+        [
+            (
+                "with_pending_callback",
+                {
+                    "Events": [
+                        {
+                            "Id": 1,
+                            "EventType": "CallbackStarted",
+                            "CallbackStartedDetails": {"CallbackId": "callback-123"},
+                        },
+                        {"Id": 2, "EventType": "StepStarted"},
+                    ]
+                },
+                "callback-123",
+            ),
+            (
+                "with_completed_callback",
+                {
+                    "Events": [
+                        {
+                            "Id": 1,
+                            "EventType": "CallbackStarted",
+                            "CallbackStartedDetails": {"CallbackId": "callback-123"},
+                        },
+                        {"Id": 1, "EventType": "CallbackCompleted"},
+                    ]
+                },
+                None,
+            ),
+            (
+                "no_callbacks",
+                {"Events": [{"Id": 1, "EventType": "StepStarted"}]},
+                None,
+            ),
+        ]
+    )
+    def test_check_for_pending_callbacks(self, name, history_response, expected_callback_id):
+        """Test checking for pending callbacks"""
+        self.mock_client.get_durable_execution_history.return_value = history_response
+
+        callback_id = self.handler.check_for_pending_callbacks("test-arn")
+
+        self.assertEqual(callback_id, expected_callback_id)
+        self.mock_client.get_durable_execution_history.assert_called_once_with("test-arn")
+
+    def test_check_for_pending_callbacks_handles_exception(self):
+        """Test that exceptions during callback check are handled gracefully"""
+        self.mock_client.get_durable_execution_history.side_effect = Exception("API error")
+
+        callback_id = self.handler.check_for_pending_callbacks("test-arn")
+
+        self.assertIsNone(callback_id)
+
+    @parameterized.expand(
+        [
+            (
+                "success",
+                CHOICE_SUCCESS,
+                ["test result"],
+                "send_callback_success",
+                {"callback_id": "callback-123", "result": "test result"},
+                True,
+            ),
+            (
+                "failure",
+                CHOICE_FAILURE,
+                ["Error occurred", "CustomError"],
+                "send_callback_failure",
+                {"callback_id": "callback-123", "error_message": "Error occurred", "error_type": "CustomError"},
+                True,
+            ),
+            (
+                "heartbeat",
+                CHOICE_HEARTBEAT,
+                [],
+                "send_callback_heartbeat",
+                {"callback_id": "callback-123"},
+                False,
+            ),
+            (
+                "stop_execution",
+                CHOICE_STOP,
+                ["Execution stopped by user", "StopError"],
+                "stop_durable_execution",
+                {
+                    "durable_execution_arn": "test-arn",
+                    "error_message": "Execution stopped by user",
+                    "error_type": "StopError",
+                },
+                True,
+            ),
+        ]
+    )
+    @patch("samcli.lib.utils.durable_callback_handler.click.prompt")
+    @patch("samcli.lib.utils.durable_callback_handler.click.echo")
+    def test_prompt_callback_response(
+        self,
+        name,
+        choice,
+        prompt_responses,
+        method_name,
+        expected_call_args,
+        expected_result,
+        mock_echo,
+        mock_prompt,
+    ):
+        """Test prompting for different callback response types"""
+        mock_prompt.side_effect = [choice] + prompt_responses
+
+        result = self.handler.prompt_callback_response("test-arn", "callback-123")
+
+        self.assertEqual(result, expected_result)
+        lambda_client_api_call = getattr(self.mock_client, method_name)
+        lambda_client_api_call.assert_called_once_with(**expected_call_args)
+        self.assertIn("callback-123", self.handler._prompted_callbacks)
+
+    def test_prompt_callback_response_only_prompts_once(self):
+        """Test that callback is only prompted once per ID"""
+        self.handler._prompted_callbacks.add("callback-123")
+
+        result = self.handler.prompt_callback_response("test-arn", "callback-123")
+
+        self.assertFalse(result)
+        self.mock_client.send_callback_success.assert_not_called()
+
+    @parameterized.expand(
+        [
+            ("before_prompt", True, False),
+            ("after_selection", False, True),
+        ]
+    )
+    @patch("samcli.lib.utils.durable_callback_handler.click.prompt")
+    @patch("samcli.lib.utils.durable_callback_handler.click.echo")
+    def test_prompt_callback_response_checks_execution_complete(
+        self, name, set_before_prompt, set_during_prompt, mock_echo, mock_prompt
+    ):
+        """Test that prompt respects execution_complete event"""
+        execution_complete = threading.Event()
+
+        if set_before_prompt:
+            execution_complete.set()
+        elif set_during_prompt:
+
+            def prompt_side_effect(*args, **kwargs):
+                execution_complete.set()
+                return CHOICE_SUCCESS
+
+            mock_prompt.side_effect = prompt_side_effect
+
+        result = self.handler.prompt_callback_response("test-arn", "callback-123", execution_complete)
+
+        self.assertFalse(result)
+        self.mock_client.send_callback_success.assert_not_called()
+        if set_before_prompt:
+            mock_prompt.assert_not_called()
+
+    @patch("samcli.lib.utils.durable_callback_handler.click.prompt")
+    @patch("samcli.lib.utils.durable_callback_handler.click.echo")
+    def test_prompt_callback_response_handles_exception(self, mock_echo, mock_prompt):
+        """Test that exceptions during callback send are handled gracefully"""
+        mock_prompt.return_value = CHOICE_HEARTBEAT
+
+        self.mock_client.send_callback_heartbeat.side_effect = Exception("API error")
+
+        result = self.handler.prompt_callback_response("test-arn", "callback-123")
+
+        self.assertFalse(result)