fix(client): propagate resolved instance ID and version to schedule request builder (#345)

* fix(client): pass resolved instance ID and version to schedule request builder
* test(tracing): verify sync schedule_new_orchestration matches OTel span instance ID
* test(tracing): verify async schedule_new_orchestration matches OTel span instance ID
* docs(client): document instance ID and version resolution propagation in schedule_new_orchestration

---------

Signed-off-by: Hardik Kaurani <hardikkaurani1@gmail.com>

Author: hardikkaurani

durabletask/client.py

@@ -634,7 +634,11 @@ def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOu
                                    reuse_id_policy: pb.OrchestrationIdReusePolicy | None = None,
                                    tags: dict[str, str] | None = None,
                                    version: str | None = None) -> str:
+        """Schedule a new orchestration instance for execution.
 
+        If instance_id is not specified, a random UUID hex string is generated upfront
+        and shared across both the OpenTelemetry producer span and the CreateInstanceRequest.
+        """
         name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
         resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex
         resolved_version = version if version else self.default_version
@@ -644,9 +648,9 @@ def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOu
                 name, resolved_instance_id, version=resolved_version,
             ):
                 req = build_schedule_new_orchestration_req(
-                    orchestrator, input=input, instance_id=instance_id, start_at=start_at,
+                    orchestrator, input=input, instance_id=resolved_instance_id, start_at=start_at,
                     reuse_id_policy=reuse_id_policy, tags=tags,
-                    version=version if version else self.default_version,
+                    version=resolved_version,
                     data_converter=self._data_converter)
 
                 # Inject the active PRODUCER span context, or the ambient context
@@ -1182,7 +1186,11 @@ async def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInpu
                                          reuse_id_policy: pb.OrchestrationIdReusePolicy | None = None,
                                          tags: dict[str, str] | None = None,
                                          version: str | None = None) -> str:
+        """Schedule a new orchestration instance for execution asynchronously.
 
+        If instance_id is not specified, a random UUID hex string is generated upfront
+        and shared across both the OpenTelemetry producer span and the CreateInstanceRequest.
+        """
         name = orchestrator if isinstance(orchestrator, str) else task.get_name(orchestrator)
         resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex
         resolved_version = version if version else self.default_version
@@ -1192,9 +1200,9 @@ async def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInpu
                 name, resolved_instance_id, version=resolved_version,
             ):
                 req = build_schedule_new_orchestration_req(
-                    orchestrator, input=input, instance_id=instance_id, start_at=start_at,
+                    orchestrator, input=input, instance_id=resolved_instance_id, start_at=start_at,
                     reuse_id_policy=reuse_id_policy, tags=tags,
-                    version=version if version else self.default_version,
+                    version=resolved_version,
                     data_converter=self._data_converter)
 
                 parent_trace_ctx = tracing.get_current_trace_context()

tests/durabletask/test_tracing.py

@@ -591,6 +591,28 @@ def test_sync_client_propagates_ambient_context_without_lifecycle_spans(
         assert [span.name for span in otel_setup.get_finished_spans()] == [
             "host-request"]
 
+    def test_sync_client_schedule_new_orchestration_matches_span_instance_id_when_auto_generated(
+            self, otel_setup: InMemorySpanExporter):
+        """When instance_id is omitted, request.instanceId must match the producer span instance ID."""
+        stub = MagicMock()
+        stub.StartInstance.side_effect = lambda req: pb.CreateInstanceResponse(instanceId=req.instanceId)
+        with patch(
+                "durabletask.client.stubs.TaskHubSidecarServiceStub",
+                return_value=stub):
+            grpc_client = client.TaskHubGrpcClient(channel=MagicMock(), emit_trace_spans=True)
+
+        returned_instance_id = grpc_client.schedule_new_orchestration("MyOrchestrator")
+
+        start_request = stub.StartInstance.call_args.args[0]
+        assert start_request.instanceId == returned_instance_id
+        assert start_request.instanceId != ""
+
+        spans = otel_setup.get_finished_spans()
+        create_spans = [s for s in spans if s.name == "create_orchestration:MyOrchestrator"]
+        assert len(create_spans) == 1
+        assert create_spans[0].attributes[tracing.ATTR_TASK_INSTANCE_ID] == returned_instance_id
+        assert create_spans[0].attributes[tracing.ATTR_TASK_INSTANCE_ID] == start_request.instanceId
+
     @pytest.mark.asyncio
     async def test_async_client_propagates_ambient_context_without_lifecycle_spans(
             self, otel_setup: InMemorySpanExporter):
@@ -616,6 +638,29 @@ async def test_async_client_propagates_ambient_context_without_lifecycle_spans(
         assert [span.name for span in otel_setup.get_finished_spans()] == [
             "host-request"]
 
+    @pytest.mark.asyncio
+    async def test_async_client_schedule_new_orchestration_matches_span_instance_id_when_auto_generated(
+            self, otel_setup: InMemorySpanExporter):
+        """When instance_id is omitted, async request.instanceId must match the producer span instance ID."""
+        stub = MagicMock()
+        stub.StartInstance = AsyncMock(side_effect=lambda req: pb.CreateInstanceResponse(instanceId=req.instanceId))
+        with patch(
+                "durabletask.client.stubs.TaskHubSidecarServiceStub",
+                return_value=stub):
+            grpc_client = client.AsyncTaskHubGrpcClient(channel=MagicMock(), emit_trace_spans=True)
+
+        returned_instance_id = await grpc_client.schedule_new_orchestration("MyAsyncOrchestrator")
+
+        start_request = stub.StartInstance.call_args.args[0]
+        assert start_request.instanceId == returned_instance_id
+        assert start_request.instanceId != ""
+
+        spans = otel_setup.get_finished_spans()
+        create_spans = [s for s in spans if s.name == "create_orchestration:MyAsyncOrchestrator"]
+        assert len(create_spans) == 1
+        assert create_spans[0].attributes[tracing.ATTR_TASK_INSTANCE_ID] == returned_instance_id
+        assert create_spans[0].attributes[tracing.ATTR_TASK_INSTANCE_ID] == start_request.instanceId
+
 
 class TestOrchestrationServerSpan:
     """Tests for emit_orchestration_span."""