fix(testing): match replay state to the service

.github/scripts/install_otel_test_wheels.py

@@ -16,6 +16,7 @@
 ROOT = Path(__file__).resolve().parents[2]
 CORE = "aws-durable-execution-sdk-python"
 OTEL = CORE + "-otel"
+TESTING = CORE + "-testing"
 
 
 def built_wheel(package: str) -> Path:
@@ -72,7 +73,9 @@ def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--legacy-plugin", action="store_true")
     args = parser.parse_args()
-    packages = [CORE] if args.legacy_plugin else [CORE, OTEL]
+    # Both lanes use the built local runner, including its real callback-update
+    # producer. The legacy lane keeps the actual published OTel plugin unchanged.
+    packages = [CORE, TESTING] if args.legacy_plugin else [CORE, OTEL, TESTING]
     wheels = [built_wheel(package) for package in packages]
     subprocess.run(
         [

.github/tests/otel_lifecycle_compatibility_test.py

@@ -18,6 +18,7 @@
     OperationStatus,
     OperationType,
 )
+from aws_durable_execution_sdk_python.plugin import InvocationStartInfo
 from aws_durable_execution_sdk_python.plugin import DurableInstrumentationPlugin
 from aws_durable_execution_sdk_python_otel.plugin_factory import (
     ExecutionOtelPluginFactory,
@@ -171,3 +172,258 @@ def handler(event, durable_context):
             provider.shutdown()
 
     contextvars.Context().run(run)
+
+
+@pytest.mark.parametrize(
+    "view", [InvocationOtelPluginFactory, ExecutionOtelPluginFactory]
+)
+def test_installed_pair_preserves_nested_callback_terminal_on_resume(view):
+    """The installed Core 3 / OTel 2 pair retains callback completion on resume."""
+    from aws_durable_execution_sdk_python_testing.runner import (
+        DurableFunctionTestRunner,
+    )
+    import aws_durable_execution_sdk_python_testing as installed_testing
+
+    import aws_durable_execution_sdk_python_otel as installed_otel
+
+    assert "site-packages" in Path(installed_testing.__file__).resolve().parts
+    assert "site-packages" in Path(core_execution.__file__).resolve().parts
+    assert "site-packages" in Path(installed_otel.__file__).resolve().parts
+    assert version("aws-durable-execution-sdk-python-otel") == "2.0.0"
+    provider = TracerProvider()
+    exporter = InMemorySpanExporter()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    plugin = view(
+        OtelPluginConfig(
+            tracer_provider=provider,
+            enrich_logger=False,
+            context_extractor=lambda _: None,
+        )
+    )
+    submitted = threading.Event()
+    suspended = threading.Event()
+    callback_ids = []
+    updated_ids = []
+    invocation_statuses = []
+    side_effects = []
+
+    class Observer(DurableInstrumentationPlugin):
+        def on_invocation_start(self, info):
+            updated_ids.append(list(info.updated_operations))
+
+        def on_invocation_end(self, info):
+            invocation_statuses.append(info.status.value)
+            if info.status.value == "PENDING":
+                suspended.set()
+
+    class ObserverFactory:
+        def create_plugin(
+            self, info: InvocationStartInfo
+        ) -> DurableInstrumentationPlugin:
+            return Observer()
+
+    def submit(callback_id, _context):
+        side_effects.append("submit")
+        callback_ids.append(callback_id)
+        submitted.set()
+
+    def saved(_context):
+        side_effects.append("saved")
+        return "saved"
+
+    def body(_event, context):
+        context.step(saved, name="saved-before")
+        return context.wait_for_callback(submit, name="await-external")
+
+    handler = durable_execution(body, plugins=[plugin, ObserverFactory()])
+    try:
+        with DurableFunctionTestRunner(
+            handler=handler, skip_time=False, poll_interval=0.01, execution_timeout=20
+        ) as runner:
+            arn = runner.run_async(input="{}")
+            assert submitted.wait(8) and suspended.wait(8)
+            runner.send_callback_success(callback_ids[0], result=b"done")
+            result = runner.wait_for_result(arn, timeout=10)
+        assert result.status.value == "SUCCEEDED"
+        assert result.result == '"done"'
+        assert invocation_statuses == ["PENDING", "SUCCEEDED"]
+        assert side_effects == ["saved", "submit"]
+        assert updated_ids[0] == [] and len(updated_ids[1]) == 1
+        spans = exporter.get_finished_spans()
+        callback_spans = [
+            span for span in spans if span.name == "await-external create callback id"
+        ]
+        terminal = [
+            span
+            for span in callback_spans
+            if span.attributes["durable.operation.status"] == "SUCCEEDED"
+        ]
+        assert len(terminal) == 1
+        assert terminal[0].attributes["durable.operation.id"] == updated_ids[1][0]
+        parent_spans = [span for span in spans if span.name == "await-external"]
+        assert terminal[0].parent in [span.context for span in parent_spans]
+        assert all(
+            span.context.trace_id == terminal[0].context.trace_id
+            for span in callback_spans
+        )
+    finally:
+        provider.shutdown()
+
+
+@pytest.mark.parametrize(
+    "view", [InvocationOtelPluginFactory, ExecutionOtelPluginFactory]
+)
+def test_installed_pair_waits_for_completed_replaychildren_parent(view):
+    """A real large child result retains and replays its callback's parent."""
+    import json
+    import time
+
+    from aws_durable_execution_sdk_python.config import CallbackConfig
+    from aws_durable_execution_sdk_python.serdes import JsonSerDes
+    from aws_durable_execution_sdk_python_testing.runner import (
+        DurableFunctionTestRunner,
+    )
+    import aws_durable_execution_sdk_python_otel as installed_otel
+    import aws_durable_execution_sdk_python_testing as installed_testing
+
+    for module in (core_execution, installed_otel, installed_testing):
+        assert "site-packages" in Path(module.__file__).resolve().parts
+    assert version("aws-durable-execution-sdk-python-otel") == "2.0.0"
+    provider = TracerProvider()
+    exporter = InMemorySpanExporter()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    plugin = view(
+        OtelPluginConfig(
+            tracer_provider=provider,
+            enrich_logger=False,
+            context_extractor=lambda _: None,
+        )
+    )
+    starts, ends, input_parents = [], [], []
+    child_calls, saved_calls, marker_calls = [], [], []
+    exports_at_marker = []
+
+    class Observer(DurableInstrumentationPlugin):
+        def on_invocation_start(self, info):
+            starts.append(info)
+
+        def on_invocation_end(self, info):
+            ends.append(info.status.value)
+
+    class ObserverFactory:
+        def create_plugin(
+            self, info: InvocationStartInfo
+        ) -> DurableInstrumentationPlugin:
+            return Observer()
+
+    def callback_spans():
+        return [
+            span
+            for span in exporter.get_finished_spans()
+            if span.name == "b"
+            and span.attributes.get("durable.operation.status") == "SUCCEEDED"
+        ]
+
+    def saved(_context):
+        saved_calls.append(1)
+        return "saved"
+
+    def child(context):
+        child_calls.append(1)
+        callback = context.create_callback(
+            name="b", config=CallbackConfig(serdes=JsonSerDes())
+        )
+        assert context.step(saved, name="saved-child") == "saved"
+        return {"id": callback.callback_id, "large": "x" * (300 * 1024)}
+
+    def marker(_context):
+        marker_calls.append(1)
+        exports_at_marker.extend(callback_spans())
+        return "marked"
+
+    def body(_event, context):
+        a = context.create_callback(
+            name="a", config=CallbackConfig(serdes=JsonSerDes())
+        )
+        b_id = context.run_in_child_context(child, name="large-parent")["id"]
+        result = a.result()
+        assert context.step(marker, name="after-a") == "marked"
+        return {"a": result, "b_id": b_id}
+
+    wrapped = durable_execution(body, plugins=[ObserverFactory(), plugin])
+
+    def entrypoint(event, context):
+        input_parents.append(
+            [
+                (op.operation_id, op.status.value, op.context_details.replay_children)
+                for op in event.initial_execution_state.operations
+                if op.name == "large-parent" and op.context_details is not None
+            ]
+        )
+        return wrapped(event, context)
+
+    def completed(runner, arn, count):
+        deadline = time.monotonic() + 10
+        while time.monotonic() < deadline:
+            events = runner.get_execution_history(
+                arn, include_execution_data=True
+            ).events
+            if (
+                len(ends) >= count
+                and sum(e.event_type == "InvocationCompleted" for e in events) >= count
+            ):
+                return
+            time.sleep(0.01)
+        raise AssertionError("Invocation did not finish")
+
+    try:
+        with DurableFunctionTestRunner(
+            handler=entrypoint,
+            skip_time=False,
+            poll_interval=0.01,
+            execution_timeout=25,
+        ) as runner:
+            arn = runner.run_async(input="{}")
+            a_id = runner.wait_for_callback(arn, name="a", timeout=5)
+            b_id = runner.wait_for_callback(arn, name="b", timeout=5)
+            completed(runner, arn, 1)
+            runner.send_callback_success(b_id, result=b'"B"')
+            completed(runner, arn, 2)
+            middle_exports = callback_spans()
+            runner.send_callback_success(a_id, result=b'"A"')
+            result = runner.wait_for_result(arn, timeout=10)
+        assert result.status.value == "SUCCEEDED"
+        assert json.loads(result.result) == {"a": "A", "b_id": b_id}
+        assert ends == ["PENDING", "PENDING", "SUCCEEDED"]
+        assert child_calls == [1, 1, 1] and saved_calls == [1] and marker_calls == [1]
+        b_info = next(
+            info for info in starts[1].updated_operations.values() if info.name == "b"
+        )
+        assert input_parents[1] == [(b_info.parent_id, "SUCCEEDED", True)]
+        terminal = callback_spans()
+        assert len(middle_exports) == len(terminal) == 1
+        assert exports_at_marker == middle_exports == terminal
+        assert terminal[0].attributes["durable.operation.id"] == b_info.operation_id
+        parents = [
+            span
+            for span in exporter.get_finished_spans()
+            if span.attributes.get("durable.operation.id") == b_info.parent_id
+            and span.attributes.get("durable.operation.status") == "SUCCEEDED"
+        ]
+        assert terminal[0].parent in [span.context for span in parents]
+        assert all(
+            span.context.trace_id == terminal[0].context.trace_id for span in parents
+        )
+        if view is InvocationOtelPluginFactory:
+            invocations = [
+                span
+                for span in exporter.get_finished_spans()
+                if span.name == "Invocation"
+            ]
+            assert len(parents) == len(invocations) == 3
+            assert terminal[0].parent == parents[1].context
+            assert parents[1].parent == invocations[1].context
+        else:
+            assert len(parents) == 1
+    finally:
+        provider.shutdown()

.github/workflows/ci.yml

@@ -10,7 +10,10 @@ on:
     branches: [ main ]
 
   pull_request:
-    branches: [ main ]
+    branches:
+      - main
+      - test/otel-conformance-feasibility
+      - fix/python-testing-replay-state-20261021
 
 jobs:
   lint-commits:

.github/workflows/cloud-tests.yml

@@ -2,7 +2,10 @@ name: Cloud tests
 
 on:
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - test/otel-conformance-feasibility
+      - fix/python-testing-replay-state-20261021
   workflow_dispatch:
     inputs:
       lambda_endpoint:

.github/workflows/conformance-tests.yml

@@ -8,7 +8,10 @@ on:
   push:
     branches: ["main"]
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - test/otel-conformance-feasibility
+      - fix/python-testing-replay-state-20261021
     paths:
       - "packages/aws-durable-execution-sdk-python/**"
       - "packages/aws-durable-execution-sdk-python-conformance-tests/**"

.github/workflows/js-examples.yml

@@ -20,7 +20,10 @@ on:
       - ".github/workflows/js-examples.yml"
       - "pyproject.toml"
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - test/otel-conformance-feasibility
+      - fix/python-testing-replay-state-20261021
     paths:
       - "packages/aws-durable-execution-sdk-python-testing/**"
       - ".github/scripts/js_examples/**"

.github/workflows/opentelemetry-conformance-tests.yml

@@ -8,7 +8,10 @@ name: OpenTelemetry Conformance Tests
 
 on:
   pull_request:
-    branches: ["main"]
+    branches:
+      - main
+      - test/otel-conformance-feasibility
+      - fix/python-testing-replay-state-20261021
     paths:
       - "packages/aws-durable-execution-sdk-python/**"
       - "packages/aws-durable-execution-sdk-python-otel/**"

CONTRIBUTING.md

@@ -72,7 +72,7 @@ hatch run dev-examples:test    # run examples tests only
 
 ### Installed package compatibility testing
 
-Build the core and OTel distributions with `hatch build` in each package, then run
+Build the core, OTel and testing-library distributions with `hatch build` in each package, then run
 these commands from the repository root:
 
 ```bash

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

@@ -489,7 +489,9 @@ durable checkpoint observations still do not close or re-export that segment.
 The service omits descendants of completed contexts unless their ancestors retain
 children through `ReplayChildren`. An `UpdatedOperationIds` entry alone does not
 provide missing operation metadata, and the plugin does not synthesize an
-operation span from that missing record.
+operation span from that missing record. Public input-delivery tests use a real
+large child result to retain metadata, while still asserting one saved-step
+execution and terminal export with the correct parent in each view.
 
 Known completed context metadata is retained from the full visible input snapshot
 for descendants delivered by later checkpoint responses. Both views retain actual