fix(extras): record the session resume mode

packages/aws-durable-execution-sdk-python-extras/README.md

@@ -236,16 +236,18 @@ The handler's return value is checkpointed as the session's result. So it must b
 
 A running MicroVM pays compute charges. A suspended MicroVM pays only for snapshot storage, and it keeps its memory and files. A session can suspend its MicroVM between jobs in two ways:
 
-1. `auto_suspend_on_idle=True` asks the worker in the MicroVM to suspend its own MicroVM when no job has run for `auto_suspend_idle_time`. The default idle time is 60 seconds, or the session `timeout` if that is shorter. The idle time must be at least 10 seconds.
+1. `auto_suspend_on_idle=True` asks the worker in the MicroVM to suspend its own MicroVM when no job has run for `auto_suspend_idle_time`. The default idle time is 60 seconds, or the session `timeout` if that is shorter. The idle time must be at least 10 seconds. So a session whose `timeout` is under 10 seconds does not suspend by default, even with `auto_suspend_on_idle=True`.
 2. `idle_policy` sets the `RunMicrovm` idle policy. The service then suspends the MicroVM when its endpoint receives no traffic. A running job receives no traffic after its delivery. So `maxIdleDurationSeconds` must be longer than the longest job.
 
-With either setting, each request step calls `GetMicrovm` before it sends the job. It resumes a suspended MicroVM, and it sends the job to the endpoint that `GetMicrovm` reports. The MicroVM can suspend itself between that check and the request. The endpoint then answers 502, or the worker answers 503. So after a 502, 503, or 504, or a request without an answer, the step checks the state again before it retries. The wait for a resume counts toward the request's `retry_window`.
+For a session that launched with either setting, each request step calls `GetMicrovm` before it sends the job. It resumes a suspended MicroVM, and it sends the job to the endpoint that `GetMicrovm` reports. The MicroVM can suspend itself between that check and the request. The endpoint then answers 502, or the worker answers 503. So after a 502, 503, or 504, or a request without an answer, the step checks the state again before it retries. The wait for a resume counts toward the request's `retry_window`.
 
-A MicroVM that is terminating, terminated, or removed fails the job at once with `MicrovmNotRunningError`, a subclass of `MicrovmDeliveryError`. A retry finds the same state, so the job is not retried. Without either setting, the session makes no `GetMicrovm` call. A terminated MicroVM's endpoint answers 502, and the request step retries a 502 as a MicroVM that is still starting. So the job fails with a plain `MicrovmDeliveryError`, after the retries of both tiers.
+The worker reads its idle time once, from the run hook payload of the launch. `RunMicrovm` sets the idle policy once, at the launch. So a deploy that changes `auto_suspend_on_idle` or `idle_policy` does not change what a running MicroVM does. The resume mode says whether each job checks the MicroVM state, and resumes a suspended MicroVM, before delivery. The launch checkpoint records the resume mode in its `resumeBeforeDelivery` field. Replay follows the recorded resume mode, not the current config. So a deploy changes the resume mode only for sessions that launch after it.
+
+A MicroVM that is terminating, terminated, or removed fails the job at once with `MicrovmNotRunningError`, a subclass of `MicrovmDeliveryError`. A retry finds the same state, so the job is not retried. A session that launched with neither setting makes no `GetMicrovm` call. A terminated MicroVM's endpoint answers 502, and the request step retries a 502 as a MicroVM that is still starting. So the job fails with a plain `MicrovmDeliveryError`, after the retries of both tiers.
 
 Suspending is off by default, because it needs more permissions and changes what the MicroVM does between jobs:
 
-- The function's role needs `lambda:GetMicrovm` and `lambda:ResumeMicrovm`. Without them, every job fails.
+- The function's role needs `lambda:GetMicrovm` and `lambda:ResumeMicrovm`. Without them, every job fails. A later deploy that turns suspending off does not change the resume mode of a session that launched with it. So keep `lambda:GetMicrovm` and `lambda:ResumeMicrovm` until every session that launched with `auto_suspend_on_idle=True` or an `idle_policy` has ended.
 - With `auto_suspend_on_idle`, the MicroVM's role needs `lambda:SuspendMicrovm` on the image. That permission lets code in any MicroVM of the image suspend any other. So use one image per trust boundary.
 - Background processes that an earlier job started stop while the MicroVM is suspended. The worker counts only its running jobs.
 

packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py

@@ -134,8 +134,9 @@
 class LaunchResult:
     """What the launch step checkpoints.
 
-    Wire names: ``microvmId``, ``endpoint``, and ``delivery``. The JavaScript
-    operation checkpoints the same document.
+    Wire names: ``microvmId``, ``endpoint``, ``delivery``, and
+    ``resumeBeforeDelivery``. The JavaScript operation checkpoints the same
+    document.
     """
 
     microvm_id: str
@@ -144,22 +145,40 @@ class LaunchResult:
     so every replay follows the delivery that the launch prepared for."""
     endpoint: str | None = None
     """The MicroVM's endpoint. Present only for HTTP delivery."""
+    resume_before_delivery: bool | None = None
+    """Whether each session job checks the MicroVM state, and resumes a
+    suspended MicroVM, before delivery. Only a session launch records it.
+    Replay reads it from the checkpoint. So every replay follows the mode that
+    the launch prepared the MicroVM for. A launch checkpoint from code that
+    predates this field does not have it."""
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> LaunchResult:
         # A checkpoint without a delivery gets an empty one. The operation
         # then terminates the MicroVM as for any delivery it does not know.
+        # Only a bool is a resume mode. In Python True == 1, so a check by
+        # equality would accept 1. Any other value reads as no resume mode.
+        resume_before_delivery = data.get("resumeBeforeDelivery")
         return cls(
             microvm_id=data["microvmId"],
             delivery=data.get("delivery", ""),
             endpoint=data.get("endpoint"),
+            resume_before_delivery=(
+                resume_before_delivery
+                if isinstance(resume_before_delivery, bool)
+                else None
+            ),
         )
 
     def to_dict(self) -> dict[str, Any]:
         data: dict[str, Any] = {"microvmId": self.microvm_id}
         if self.endpoint is not None:
             data["endpoint"] = self.endpoint
         data["delivery"] = self.delivery
+        # A microvm() launch records no resume mode. So its checkpoint has no
+        # such key, as in the JavaScript operation.
+        if self.resume_before_delivery is not None:
+            data["resumeBeforeDelivery"] = self.resume_before_delivery
         return data
 
 
@@ -363,20 +382,28 @@ def launch(
         payload: str,
         delivery: str,
         idle_policy: Mapping[str, Any] | None = None,
+        resume_before_delivery: bool | None = None,
     ) -> LaunchResult:
         """Run the launch step, ``<name>.launch``.
 
         The client token is the SHA-256 hex digest of ``token_source``. It
         must be unique per launch and stable across retries and replays.
 
+        ``resume_before_delivery`` is the resume mode to record. Only a
+        session launch sets it.
+
         Raises:
             MicrovmLaunchError: When ``RunMicrovm`` fails after all retries.
         """
         try:
             return LaunchResult.from_dict(
                 child.step(
                     lambda _: self._launch_call(
-                        token_source, payload, delivery, idle_policy
+                        token_source,
+                        payload,
+                        delivery,
+                        idle_policy,
+                        resume_before_delivery,
                     ).to_dict(),
                     name=f"{self.name}.launch",
                     config=StepConfig(
@@ -540,6 +567,7 @@ def _launch_call(
         payload: str,
         delivery: str,
         idle_policy: Mapping[str, Any] | None,
+        resume_before_delivery: bool | None = None,
     ) -> LaunchResult:
         """Call ``RunMicrovm``, and return what the launch step checkpoints."""
         over_http = delivery == HTTP_DELIVERY
@@ -589,6 +617,7 @@ def _launch_call(
             microvm_id=microvm_id,
             delivery=delivery,
             endpoint=endpoint if over_http else None,
+            resume_before_delivery=resume_before_delivery,
         )
 
     def _terminate_call(self, microvm_id: str) -> None:

packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py

@@ -106,7 +106,9 @@ def microvm_session(
     its own MicroVM when no job has run for ``auto_suspend_idle_time``. With
     that setting or an ``idle_policy``, the request step of each job calls
     ``GetMicrovm`` first. It resumes a suspended MicroVM, and it returns at
-    once for a new MicroVM that is still booting.
+    once for a new MicroVM that is still booting. The launch checkpoint
+    records whether the MicroVM can be suspended. Replay follows that record,
+    not the current config.
 
     A failed :meth:`MicrovmSession.invoke` raises inside the handler. The
     handler can catch the error and continue with the same MicroVM. An
@@ -198,9 +200,16 @@ class MicrovmSession:
     ``map`` and ``parallel`` items.
     """
 
-    def __init__(self, scope: _SessionScope, launched: LaunchResult) -> None:
+    def __init__(
+        self,
+        scope: _SessionScope,
+        launched: LaunchResult,
+        *,
+        resume_before_delivery: bool,
+    ) -> None:
         self._scope = scope
         self._launched = launched
+        self._resume_before_delivery = resume_before_delivery
 
     @property
     def microvm_id(self) -> str:
@@ -283,6 +292,7 @@ def invoke(
         heartbeat_timeout_seconds = _validate_heartbeat(name, options.heartbeat_timeout)
         _validate_input(name, job_input)
         launched = self._launched
+        resume = self._resume_before_delivery
 
         def run(job_context: DurableContext) -> Any:
             callback = create_job_callback(
@@ -297,7 +307,7 @@ def run(job_context: DurableContext) -> Any:
                 heartbeat_timeout_seconds=heartbeat_timeout_seconds,
                 request=options,
                 retry_window_seconds=retry_window_seconds,
-                resume=scope.resume_before_delivery,
+                resume=resume,
             )
             try:
                 return callback.result()
@@ -327,11 +337,13 @@ class _SessionScope(_MicrovmScope):
     in it would be outside the session's child context."""
 
     @property
-    def resume_before_delivery(self) -> bool:
-        """Whether each job must check the MicroVM state, and resume it.
+    def resume_from_config(self) -> bool:
+        """Whether the current config asks each job to check the MicroVM
+        state, and resume it.
 
         The worker suspends an idle MicroVM, and an idle policy can suspend
-        it too. Either way, each job must resume it first.
+        it too. Either way, each job must resume it first. A session launch
+        records this value as its resume mode.
         """
         return (
             self.auto_suspend_idle_seconds is not None or self.idle_policy is not None
@@ -381,14 +393,38 @@ def run_session(
             ),
             delivery=HTTP_DELIVERY,
             idle_policy=self.idle_policy,
+            resume_before_delivery=self.resume_from_config,
         )
         if launched.delivery != HTTP_DELIVERY:
             # A launch checkpoint from a release with another delivery. Every
             # session job goes over HTTP, so this release cannot run one.
             self.terminate(child, launched.microvm_id)
             raise unknown_delivery_error(self.name, launched)
+        # 1. The worker reads its idle time once, from the run hook payload of
+        #    the launch. RunMicrovm sets the idle policy once, at the launch.
+        # 2. The launch step's result comes from the checkpoint on replay.
+        # 3. So a deploy that changes auto_suspend_on_idle or idle_policy does
+        #    not change what the running MicroVM does.
+        # 4. A job that skips the state check of a suspended MicroVM gets a
+        #    502. The request step retries a 502 until both retry tiers end.
+        # 5. So each job follows the resume mode that the launch recorded,
+        #    not the current config.
+        # A launch checkpoint from code that predates the resumeBeforeDelivery
+        # field has no such field. That code computed the mode from the
+        # current config on every replay. So the session does the same for
+        # such a checkpoint.
+        resume_before_delivery = (
+            launched.resume_before_delivery
+            if launched.resume_before_delivery is not None
+            else self.resume_from_config
+        )
         try:
-            result = handler(MicrovmSession(self, launched), child)
+            result = handler(
+                MicrovmSession(
+                    self, launched, resume_before_delivery=resume_before_delivery
+                ),
+                child,
+            )
         except ExecutionError:
             # A replay that does not match the history raises
             # NonDeterministicExecutionError, and the execution fails without

packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py

@@ -463,6 +463,48 @@ def test_sends_every_job_over_http_to_a_configured_route(
     assert client.requests("create_microvm_auth_token")[0]["allowedPorts"] == [
         {"port": 3000}
     ]
+    # A single job never resumes its MicroVM. So its launch checkpoint has no
+    # resume mode.
+    launch = result.get_context("build").get_step("build.launch")
+    assert launch.result is not None
+    assert json.loads(launch.result) == {
+        "microvmId": "mvm-1",
+        "endpoint": endpoint.url,
+        "delivery": "http",
+    }
+
+
+@pytest.mark.parametrize("resume_before_delivery", [True, False])
+def test_launch_result_writes_and_reads_a_resume_mode(
+    resume_before_delivery: bool,  # noqa: FBT001 - a pytest parameter
+) -> None:
+    launched = operation.LaunchResult(
+        microvm_id="mvm-1",
+        delivery="http",
+        endpoint="mvm-1.lambda-microvm.us-east-1.on.aws",
+        resume_before_delivery=resume_before_delivery,
+    )
+
+    data = launched.to_dict()
+
+    assert data == {
+        "microvmId": "mvm-1",
+        "endpoint": "mvm-1.lambda-microvm.us-east-1.on.aws",
+        "delivery": "http",
+        "resumeBeforeDelivery": resume_before_delivery,
+    }
+    assert operation.LaunchResult.from_dict(json.loads(json.dumps(data))) == launched
+
+
+@pytest.mark.parametrize("value", ["false", "true", 1, 0, None, [], {}])
+def test_launch_result_reads_only_a_bool_as_a_resume_mode(value: Any) -> None:
+    # In Python True == 1. So a check by equality would read 1 as True.
+    launched = operation.LaunchResult.from_dict(
+        {"microvmId": "mvm-1", "delivery": "http", "resumeBeforeDelivery": value}
+    )
+
+    assert launched.resume_before_delivery is None
+    assert "resumeBeforeDelivery" not in launched.to_dict()
 
 
 def test_keeps_the_configured_ingress_for_http(