feat(workflow): transfer remote tensors with NIXL

components/src/dynamo/experimental/workflow/__init__.py

@@ -4,19 +4,59 @@
 """Experimental inference workflow APIs."""
 
 from dynamo.experimental.workflow.builder import StageHandle, Workflow
+from dynamo.experimental.workflow.compiler import DeploymentSpec, compile_workflow
+from dynamo.experimental.workflow.endpoint import WorkflowEndpointHandler
 from dynamo.experimental.workflow.ir import StageIR, WorkflowIR
+from dynamo.experimental.workflow.nixl import (
+    NixlLeaseRegistry,
+    NixlTensorCarrier,
+    NixlTensorFanout,
+    NixlTensorRef,
+)
+from dynamo.experimental.workflow.orchestrator import WorkflowOrchestrator
+from dynamo.experimental.workflow.plan import (
+    ExecutionPlan,
+    GenerateEndpointBinding,
+    InlineBinding,
+    RemoteBinding,
+)
+from dynamo.experimental.workflow.remote import RemoteStageClient, RemoteStageServer
+from dynamo.experimental.workflow.runtime import (
+    StageContext,
+    StageRunner,
+    TensorCarrier,
+    WorkflowExecutionError,
+)
 from dynamo.experimental.workflow.types import (
     StageContract,
     ValueRef,
     WorkflowValidationError,
 )
 
 __all__ = [
+    "DeploymentSpec",
     "StageContract",
     "StageHandle",
     "StageIR",
+    "StageContext",
+    "StageRunner",
+    "TensorCarrier",
     "ValueRef",
     "Workflow",
     "WorkflowIR",
+    "WorkflowEndpointHandler",
+    "WorkflowExecutionError",
+    "WorkflowOrchestrator",
     "WorkflowValidationError",
+    "ExecutionPlan",
+    "GenerateEndpointBinding",
+    "InlineBinding",
+    "NixlLeaseRegistry",
+    "NixlTensorCarrier",
+    "NixlTensorFanout",
+    "NixlTensorRef",
+    "RemoteBinding",
+    "RemoteStageClient",
+    "RemoteStageServer",
+    "compile_workflow",
 ]

components/src/dynamo/experimental/workflow/builder.py

@@ -127,7 +127,7 @@ def build(self) -> WorkflowIR:
 
         return WorkflowIR(
             name=self._name,
-            inputs=self._inputs,
+            inputs=frozenset(self._inputs),
             stages=tuple(self._stages.values()),
             outputs=self._outputs,
         )

components/src/dynamo/experimental/workflow/compiler.py

@@ -0,0 +1,92 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Lower placement-neutral workflows into physical execution plans."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Mapping, Optional, Union
+
+from dynamo.experimental.workflow.builder import Workflow
+from dynamo.experimental.workflow.ir import WorkflowIR
+from dynamo.experimental.workflow.plan import (
+    Binding,
+    ExecutionPlan,
+    InlineBinding,
+    RemoteBinding,
+)
+from dynamo.experimental.workflow.types import WorkflowValidationError, validate_name
+
+
+@dataclass(frozen=True)
+class DeploymentSpec:
+    """Stage placement requested when compiling one WorkflowIR."""
+
+    bindings: Mapping[str, Binding]
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.bindings, Mapping):
+            raise WorkflowValidationError("deployment bindings must be a mapping")
+        bindings: dict[str, Binding] = {}
+        for stage_id, binding in sorted(self.bindings.items()):
+            validate_name(stage_id, "deployment stage id")
+            if not isinstance(binding, (InlineBinding, RemoteBinding)):
+                raise WorkflowValidationError(
+                    f"binding for stage {stage_id!r} uses an unsupported type"
+                )
+            bindings[stage_id] = binding
+        object.__setattr__(self, "bindings", MappingProxyType(bindings))
+
+    @staticmethod
+    def inline(**runner_keys: str) -> "DeploymentSpec":
+        """Build bindings to runners in the orchestrator process."""
+
+        return DeploymentSpec(
+            bindings={
+                stage_id: InlineBinding(runner_key)
+                for stage_id, runner_key in runner_keys.items()
+            }
+        )
+
+    @staticmethod
+    def remote(
+        *, tensor_carrier: str | None = None, **endpoint_ids: str
+    ) -> "DeploymentSpec":
+        """Build round-robin bindings to discovered Dynamo endpoints."""
+
+        return DeploymentSpec(
+            bindings={
+                stage_id: RemoteBinding(endpoint_id, tensor_carrier=tensor_carrier)
+                for stage_id, endpoint_id in endpoint_ids.items()
+            }
+        )
+
+
+def compile_workflow(
+    workflow: Union[Workflow, WorkflowIR],
+    deployment: Optional[DeploymentSpec] = None,
+) -> ExecutionPlan:
+    """Compile one logical workflow, defaulting every stage to inline placement."""
+
+    workflow_ir = workflow.build() if isinstance(workflow, Workflow) else workflow
+    if not isinstance(workflow_ir, WorkflowIR):
+        raise TypeError("workflow must be a Workflow or WorkflowIR")
+    stage_ids = tuple(stage.id for stage in workflow_ir.stages)
+    if deployment is None:
+        deployment = DeploymentSpec.inline(
+            **{stage_id: stage_id for stage_id in stage_ids}
+        )
+    if not isinstance(deployment, DeploymentSpec):
+        raise TypeError("deployment must use DeploymentSpec")
+
+    expected = set(stage_ids)
+    actual = set(deployment.bindings)
+    if actual != expected:
+        raise WorkflowValidationError(
+            "deployment bindings differ from workflow stages; "
+            f"missing={sorted(expected - actual)}, extra={sorted(actual - expected)}"
+        )
+
+    return ExecutionPlan(workflow_ir, deployment.bindings)