feat(deploy): provision client-mode Endpoint(image=...) resources during flash deploy

src/runpod_flash/cli/commands/build_utils/handler_generator.py

@@ -315,6 +315,16 @@ def generate_handlers(self) -> List[Path]:
         )
 
         for resource_name, resource_data in resources.items():
+            # Skip client-mode resources (image-based, external service):
+            # they have no functions, so there is no handler to generate.
+            client_mode = (
+                resource_data.client_mode
+                if hasattr(resource_data, "client_mode")
+                else resource_data.get("client_mode", False)
+            )
+            if client_mode:
+                continue
+
             # Skip load-balanced resources (handled by LBHandlerGenerator)
             # Use flag determined by isinstance() at scan time
             is_load_balanced = (

src/runpod_flash/cli/commands/build_utils/manifest.py

@@ -11,6 +11,7 @@
     DEFAULT_PYTHON_VERSION,
     validate_python_version,
 )
+from runpod_flash.runtime.resource_provisioner import PROVISIONABLE_RESOURCE_TYPES
 
 from .scanner import (
     RemoteFunctionMetadata,
@@ -281,6 +282,145 @@ def _extract_config_properties(config: Dict[str, Any], resource_config) -> None:
 
         return config
 
+    @staticmethod
+    def _derive_path_fields(
+        file_path: Optional[Path], project_root: Path, module_path: str
+    ) -> tuple[str, str, str]:
+        """Derive (file_path_str, local_path_prefix, resource_module_path).
+
+        ``file_path`` is the source file a resource was declared in. When it
+        lives under ``project_root`` the fields describe the project-relative
+        location; otherwise (outside the root, or the path does not exist,
+        e.g. relative test paths) the raw path string is kept and the URL
+        prefix is derived from the module path instead.
+        """
+        file_path_str = ""
+        local_path_prefix = ""
+        resource_module_path = module_path
+        if file_path and file_path.exists():
+            try:
+                file_path_str = str(file_path.relative_to(project_root))
+                local_path_prefix = file_to_url_prefix(file_path, project_root)
+                resource_module_path = file_to_module_path(file_path, project_root)
+            except ValueError:
+                # File is outside project root — fall back to module_path
+                file_path_str = str(file_path)
+                local_path_prefix = "/" + module_path.replace(".", "/")
+        elif file_path:
+            # File path may be relative (in test scenarios)
+            file_path_str = str(file_path)
+            local_path_prefix = "/" + module_path.replace(".", "/")
+        return file_path_str, local_path_prefix, resource_module_path
+
+    def _add_client_endpoints(
+        self, resources_dict: Dict[str, Dict[str, Any]], project_root: Path
+    ) -> None:
+        """Register client-mode Endpoints (image=..., no decorated functions).
+
+        The scanner collects top-level ``Endpoint(image=...)`` objects. They
+        have no decorated functions, but ``flash deploy`` should still
+        provision them, so each gets a function-less manifest entry that the
+        deploy provisioning path (create_resource_from_manifest) consumes.
+        The entry is marked ``client_mode: True`` so consumers that generate
+        worker code (handler generators, local preview) skip it.
+        """
+        client_endpoints = (
+            getattr(self.scanner, "client_endpoints", None) if self.scanner else None
+        )
+        if not client_endpoints:
+            return
+
+        for name in sorted(client_endpoints):
+            if name in resources_dict:
+                raise ValueError(
+                    f"endpoint '{name}' is defined as both a client-mode "
+                    f"Endpoint(image=...) and a decorated resource. "
+                    f"each endpoint name must be unique across the project."
+                )
+
+            info = client_endpoints[name]
+
+            file_path_str, local_path_prefix, resource_module_path = (
+                self._derive_path_fields(
+                    info.get("file_path"), project_root, info.get("module_path", "")
+                )
+            )
+
+            deployment_config, resource_type = self._extract_client_deployment_config(
+                name, info
+            )
+
+            resources_dict[name] = {
+                "resource_type": resource_type,
+                "file_path": file_path_str,
+                "local_path_prefix": local_path_prefix,
+                "module_path": resource_module_path,
+                "functions": [],
+                "is_load_balanced": False,
+                "is_live_resource": True,
+                "config_variable": info.get("var_name"),
+                "makes_remote_calls": False,
+                "client_mode": True,
+                **deployment_config,
+            }
+
+    def _extract_client_deployment_config(
+        self, resource_name: str, info: Dict[str, Any]
+    ) -> tuple[Dict[str, Any], str]:
+        """Extract deployment config from a client-mode Endpoint instance.
+
+        Returns (config, resource_type). resource_type is the concrete
+        resource class name (e.g. "LiveServerless") when
+        create_resource_from_manifest can rebuild it; otherwise it falls back
+        to "Endpoint", which the provisioner resolves via gpuIds.
+        """
+        config: Dict[str, Any] = {}
+        resource_type = "Endpoint"
+
+        file_path = info.get("file_path")
+        var_name = info.get("var_name")
+
+        try:
+            module, cleanup = (
+                self._import_module(file_path)
+                if file_path and file_path.exists()
+                else (None, None)
+            )
+            if module is not None:
+                try:
+                    ep = getattr(module, var_name, None) if var_name else None
+                    if ep is not None:
+                        mc = getattr(ep, "_max_concurrency", 1)
+                        if mc > 1:
+                            config["max_concurrency"] = mc
+
+                        resource_config = ep._build_resource_config()
+                        self._extract_config_properties(config, resource_config)
+                        type_name = type(resource_config).__name__
+                        # Record the concrete class only when the deploy-time
+                        # provisioner can rebuild it (e.g. "LiveServerless").
+                        # Otherwise keep "Endpoint" and let the provisioner
+                        # resolve GPU vs CPU from gpuIds instead of failing on
+                        # an unsupported class name.
+                        if type_name in PROVISIONABLE_RESOURCE_TYPES:
+                            resource_type = type_name
+                finally:
+                    cleanup()
+        except Exception as e:
+            logger.debug(
+                f"Failed to extract deployment config for client endpoint "
+                f"{resource_name}: {e}"
+            )
+
+        if "imageName" not in config and "templateId" not in config:
+            logger.warning(
+                "Failed to extract image/template for client endpoint "
+                f"'{resource_name}'; the deployed endpoint may not use the "
+                "expected image."
+            )
+
+        return config, resource_type
+
     def _reconcile_python_version(
         self, resources_dict: Dict[str, Dict[str, Any]]
     ) -> str:
@@ -387,24 +527,13 @@ def build(self) -> Dict[str, Any]:
 
             # Derive path fields from the first function's source file.
             # All functions in a resource share the same source file per convention.
-            first_file = functions[0].file_path if functions else None
-            file_path_str = ""
-            local_path_prefix = ""
-            resource_module_path = functions[0].module_path if functions else ""
-
-            if first_file and first_file.exists():
-                try:
-                    file_path_str = str(first_file.relative_to(project_root))
-                    local_path_prefix = file_to_url_prefix(first_file, project_root)
-                    resource_module_path = file_to_module_path(first_file, project_root)
-                except ValueError:
-                    # File is outside project root — fall back to module_path
-                    file_path_str = str(first_file)
-                    local_path_prefix = "/" + functions[0].module_path.replace(".", "/")
-            elif first_file:
-                # File path may be relative (in test scenarios)
-                file_path_str = str(first_file)
-                local_path_prefix = "/" + functions[0].module_path.replace(".", "/")
+            file_path_str, local_path_prefix, resource_module_path = (
+                self._derive_path_fields(
+                    functions[0].file_path if functions else None,
+                    project_root,
+                    functions[0].module_path if functions else "",
+                )
+            )
 
             # Validate and collect routing for LB endpoints
             resource_routes = {}
@@ -509,6 +638,8 @@ def build(self) -> Dict[str, Any]:
                     )
                 function_registry[f.function_name] = resource_name
 
+        self._add_client_endpoints(resources_dict, project_root)
+
         # Reconcile app-level python_version across resources. One tarball serves
         # every resource in an app, so all resources must agree on one version.
         self.python_version = self._reconcile_python_version(resources_dict)

src/runpod_flash/cli/commands/build_utils/scanner.py

@@ -457,6 +457,9 @@ def __init__(self, project_dir: Path):
         self.resource_variables: Dict[str, str] = {}
         # populated after discover_remote_functions() runs
         self.import_errors: Dict[str, str] = {}
+        # client-mode Endpoints (image=..., id=None) keyed by endpoint name;
+        # each value carries var_name/file_path/module_path for the manifest.
+        self.client_endpoints: Dict[str, Dict[str, Any]] = {}
 
     def discover_remote_functions(self) -> List[RemoteFunctionMetadata]:
         """discover all @remote decorated functions and classes by importing modules."""
@@ -556,6 +559,8 @@ def _extract_from_module(
         """extract RemoteFunctionMetadata from a single imported module."""
         results: List[RemoteFunctionMetadata] = []
 
+        self._collect_client_endpoints(module, module_path, file_path)
+
         remote_objects = _find_remote_decorated(module)
         endpoint_instances = _find_endpoint_instances(module)
 
@@ -608,6 +613,61 @@ def _extract_from_module(
 
         return results
 
+    def _collect_client_endpoints(
+        self, module: Any, module_path: str, file_path: Path
+    ) -> None:
+        """record client-mode Endpoints (image=..., no id) for the manifest.
+
+        client-mode endpoints declare an external service image; they have no
+        decorated functions, but `flash deploy` should still provision them.
+        id= endpoints are excluded: they attach to an already-existing
+        endpoint, so there is nothing to provision.
+
+        Re-exports (``from defs import ep`` in a second file) surface as a
+        distinct object with the same name and image because each scanned file
+        is imported fresh; those keep the first record. The same name with a
+        *different* image is a genuine collision and fails the build.
+        """
+        for member_name in dir(module):
+            try:
+                member = getattr(module, member_name)
+            except Exception:
+                continue
+            if (
+                not isinstance(member, Endpoint)
+                or member.image is None
+                or member.id is not None
+            ):
+                continue
+            # image-based endpoints always carry a name: Endpoint.__init__
+            # raises when name is None and image is set.
+            name = member.name
+            existing = self.client_endpoints.get(name)
+            record = {
+                "var_name": member_name,
+                "file_path": file_path,
+                "module_path": module_path,
+                "object_id": id(member),
+                "image": member.image,
+            }
+            if existing is None:
+                self.client_endpoints[name] = record
+                continue
+            if existing["object_id"] == id(member):
+                continue  # aliased under a second variable name
+            if existing["file_path"] == file_path:
+                # two distinct objects with the same name in one file: last wins
+                self.client_endpoints[name] = record
+                continue
+            if existing["image"] == member.image:
+                # re-exported from another file; keep the first record
+                continue
+            raise ValueError(
+                f"endpoint '{name}' is defined in multiple files: "
+                f"{existing['file_path']}, {file_path}. "
+                f"each endpoint name must be unique across the project."
+            )
+
     def _populate_resource_dicts(self, functions: List[RemoteFunctionMetadata]) -> None:
         """populate resource tracking dicts for ManifestBuilder compatibility."""
         for f in functions:

src/runpod_flash/cli/commands/deploy.py

@@ -217,10 +217,20 @@ async def _resolve_and_deploy(
 ) -> None:
     import time as _time
 
-    app, resolved_env_name = await _resolve_environment(app_name, env_name)
-
     local_manifest = validate_local_manifest()
 
+    # An empty resources map means the project declares no deployable
+    # endpoints at all (no @remote functions, no client-mode
+    # Endpoint(image=...) declarations -- the build registers those in the
+    # manifest). Nothing is provisioned here, so don't claim a deployment.
+    if not local_manifest.get("resources"):
+        console.print(
+            "[yellow]No deployable resources found; nothing was deployed[/yellow]"
+        )
+        return
+
+    app, resolved_env_name = await _resolve_environment(app_name, env_name)
+
     from rich.progress import (
         BarColumn,
         DownloadColumn,

src/runpod_flash/cli/commands/preview.py

@@ -173,6 +173,10 @@ def _parse_resources_from_manifest(manifest: dict) -> dict:
     # Parse resources from manifest
     manifest_resources = manifest.get("resources", {})
     for resource_name, resource_data in manifest_resources.items():
+        # Client-mode endpoints run a user-supplied image and are provisioned
+        # remotely by flash deploy; they are not launched as local containers.
+        if resource_data.get("client_mode", False):
+            continue
         resources[resource_name] = {
             "is_load_balanced": resource_data.get("is_load_balanced", False),
             "imageName": resource_data.get("imageName", FLASH_CPU_LB_IMAGE),

src/runpod_flash/cli/utils/deployment.py

@@ -122,9 +122,14 @@ async def reconcile_and_provision_resources(
     # changes (no resource config diff) still trigger a rolling release.
     # The fingerprint is computed during flash build from user source files.
     # Mutation intentional: persisted to state manifest via update_build_manifest below.
+    # Client-mode endpoints run a user-supplied image; local source changes do
+    # not affect them, so skip the fingerprint to avoid redeploying them on
+    # every code-only deploy.
     source_fingerprint = local_manifest.get("source_fingerprint")
     if source_fingerprint:
         for resource_config in local_manifest.get("resources", {}).values():
+            if resource_config.get("client_mode"):
+                continue
             env = resource_config.setdefault("env", {})
             env["_FLASH_SOURCE_FINGERPRINT"] = source_fingerprint
 

src/runpod_flash/runtime/resource_provisioner.py

@@ -6,6 +6,19 @@
 
 logger = logging.getLogger(__name__)
 
+# Resource classes that create_resource_from_manifest can construct from a
+# manifest entry. "Endpoint" is accepted separately (resolved to one of these
+# via gpuIds). ManifestBuilder gates client-mode resource_type names against
+# this list so the manifest never names a class this module cannot rebuild.
+PROVISIONABLE_RESOURCE_TYPES = (
+    "ServerlessResource",
+    "LiveServerless",
+    "CpuLiveServerless",
+    "LoadBalancerSlsResource",
+    "LiveLoadBalancer",
+    "CpuLiveLoadBalancer",
+)
+
 
 def create_resource_from_manifest(
     resource_name: str,
@@ -59,14 +72,7 @@ def create_resource_from_manifest(
             resource_type = "CpuLiveServerless"
 
     # Support both Serverless and LoadBalancer resource types
-    if resource_type not in [
-        "ServerlessResource",
-        "LiveServerless",
-        "CpuLiveServerless",
-        "LoadBalancerSlsResource",
-        "LiveLoadBalancer",
-        "CpuLiveLoadBalancer",
-    ]:
+    if resource_type not in PROVISIONABLE_RESOURCE_TYPES:
         raise ValueError(
             f"Unsupported resource type for auto-provisioning: {resource_type}"
         )