feat(sdk): run plugin hooks off checkpoint thread

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py

@@ -805,9 +805,9 @@ def _release_invocation_scope(self) -> None:
           on such a thread is still correlated to this finished invocation's
           spans.
         * ``_tracing_enabled`` is cleared so a hook that arrives after the
-          invocation end -- one dispatched off the checkpointing path, for
-          instance -- cannot start a span after the invocation span was ended and
-          the provider flushed.
+          invocation end cannot start a span after the invocation span was ended
+          and the provider flushed. The SDK dispatches no operation hook after
+          the end hook; this guard covers a caller that does.
         """
         self._detach_remaining_contexts()
         unbind_invocation(self)

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py

@@ -893,9 +893,9 @@ def _release_invocation_scope(self) -> None:
           on such a thread is still correlated to this finished invocation's
           spans.
         * ``_tracing_enabled`` is cleared so a hook that arrives after the
-          invocation end -- one dispatched off the checkpointing path, for
-          instance -- cannot start a span after the invocation span was ended and
-          the provider flushed.
+          invocation end cannot start a span after the invocation span was ended
+          and the provider flushed. The SDK dispatches no operation hook after
+          the end hook; this guard covers a caller that does.
         """
         self._detach_remaining_contexts()
         unbind_invocation(self)

packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/log_filter.py

@@ -41,10 +41,11 @@
       coordinator's Context, carrying the invocation claim. Plugins also
       reaffirm the claim from hooks that run before user code on that thread.
     - A record emitted on a thread that carries no live claim is left unstamped,
-      whatever the number of open invocations. Two threads carry no claim: the
-      SDK's background checkpointing thread, and a thread customer code starts
-      itself, since Python does not copy context into a new thread. Correlation
-      is lost for those records.
+      whatever the number of open invocations. Three kinds of thread carry no
+      claim: the SDK's background checkpointing thread, the SDK's plugin change
+      thread, which runs ``on_operation_change``, and a thread customer code
+      starts itself, since Python does not copy context into a new thread.
+      Correlation is lost for those records.
 
 Resolving an unclaimed record against the single open invocation, when exactly
 one is open, would be wrong in two orderings. A thread claimed by invocation A

packages/aws-durable-execution-sdk-python-telemetry-exporter/src/aws_durable_execution_sdk_python_telemetry_exporter/plugin.py

@@ -496,9 +496,10 @@ def __init__(
         # operation_id -> OperationInfo, adopted verbatim from the SDK's
         # authoritative snapshot (invocation start/end and operation-change).
         self._operations: dict[str, OperationInfo] = {}
-        # Set once this invocation has ended. A hook that arrives afterwards (an
-        # operation-change for a checkpoint that completed just before the end)
-        # must emit nothing (mirrors the Java ExecutionState.closed flag).
+        # Set once this invocation has ended. A hook that arrives afterwards
+        # must emit nothing (mirrors the Java ExecutionState.closed flag). The
+        # SDK delivers every pending operation-change before the end hook; the
+        # guard covers a caller that does not.
         self._closed = False
         # A duplicate End must wait for the admitted closing build before its
         # drain captures scheduler state. Nested callbacks defer that wait to

packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py

@@ -509,8 +509,15 @@ def _peek_next_operation_id(self) -> str:
         )
 
     def _peek_next_checkpoint(self) -> CheckpointedResult:
-        """Return the checkpoint for this context's next operation, without consuming it."""
-        return self.state.get_checkpoint_result(self._peek_next_operation_id())
+        """Return the checkpoint for this context's next operation, without consuming it.
+
+        The read takes no plugin end hook: the hook of an operation a
+        checkpoint response completed during this invocation belongs to the
+        read that consumes the result, the operation's own.
+        """
+        return self.state.get_checkpoint_result(
+            self._peek_next_operation_id(), claim_end_hook=False
+        )
 
     def _next_operation_exists(self) -> bool:
         """True if a checkpoint exists for this context's next operation."""
@@ -548,9 +555,14 @@ def _replay_aware(
         # Only look up a checkpoint when replaying. Operation call sites pass
         # their atomically allocated identifier so concurrent callers inspect
         # the checkpoint for their own ID rather than peeking at shared state.
+        # Both reads inspect: neither consumes the result, so neither takes
+        # the plugin end hook a checkpoint response may have recorded for the
+        # operation. The operation's own read, which consumes the result, does.
         next_checkpoint: CheckpointedResult | None = (
             (
-                self.state.get_checkpoint_result(operation_identifier.operation_id)
+                self.state.get_checkpoint_result(
+                    operation_identifier.operation_id, claim_end_hook=False
+                )
                 if operation_identifier is not None
                 else self._peek_next_checkpoint()
             )