fix: reject multiple @remote functions on the same queue-based resour…

src/runpod_flash/client.py

@@ -223,6 +223,27 @@ def decorator(func_or_class):
             func_or_class.__is_lb_route_handler__ = True
             return func_or_class
 
+        # queue-based endpoints support one function per resource config.
+        # a second @remote on the same config object would silently shadow the
+        # first function in the generated handler, so reject it early.
+        if not is_lb_resource:
+            func_name_for_check = (
+                func_or_class.__name__
+                if hasattr(func_or_class, "__name__")
+                else str(func_or_class)
+            )
+            existing_owner = getattr(resource_config, "_remote_function_name", None)
+            if existing_owner is not None:
+                raise ValueError(
+                    f"Queue-based resource '{resource_config.name}' is already used by "
+                    f"@remote function '{existing_owner}'. Each queue-based resource "
+                    f"config supports only one function. Create a separate resource "
+                    f"config for '{func_name_for_check}'."
+                )
+            object.__setattr__(
+                resource_config, "_remote_function_name", func_name_for_check
+            )
+
         # Local execution mode - execute without provisioning remote servers
         if local:
             func_or_class.__remote_config__ = routing_config

tests/unit/test_live_serverless_stub.py

@@ -5,8 +5,8 @@
 from runpod_flash import remote, LiveServerless
 
 
-# Create a dummy config for testing
-dummy_config = LiveServerless(name="test-endpoint")
+def _fresh_config():
+    return LiveServerless(name="test-endpoint")
 
 
 class TestGetFunctionSource:
@@ -44,7 +44,7 @@ def real_sync_function(x: int) -> int:
             """A real sync function."""
             return x * 3
 
-        decorated_real = remote(resource_config=dummy_config)(real_sync_function)
+        decorated_real = remote(resource_config=_fresh_config())(real_sync_function)
 
         source, src_hash = get_function_source(decorated_real)
 
@@ -55,7 +55,7 @@ def real_sync_function(x: int) -> int:
     def test_async_function_with_remote_decorator(self):
         """Test extraction of decorated async function source."""
 
-        @remote(resource_config=dummy_config)
+        @remote(resource_config=_fresh_config())
         async def decorated_async(x: int) -> int:
             """A decorated async function."""
             return x * 4
@@ -69,7 +69,7 @@ async def decorated_async(x: int) -> int:
     def test_function_source_excludes_decorator_line(self):
         """Test that source extraction correctly excludes decorator lines."""
 
-        @remote(resource_config=dummy_config)
+        @remote(resource_config=_fresh_config())
         async def function_with_decorator(x: int) -> int:
             """Function with decorator."""
             return x * 5
@@ -85,7 +85,7 @@ async def function_with_decorator(x: int) -> int:
     def test_ast_parsing_handles_async_function(self):
         """Test that AST parsing correctly identifies async functions."""
 
-        @remote(resource_config=dummy_config)
+        @remote(resource_config=_fresh_config())
         async def async_test_function(x: int) -> int:
             """Async function for AST test."""
             result = x * 6

tests/unit/test_queue_resource_one_function.py

@@ -0,0 +1,86 @@
+"""Tests that queue-based resources reject multiple @remote functions."""
+
+import os
+from unittest.mock import patch
+
+import pytest
+
+from runpod_flash.client import remote
+from runpod_flash.core.resources.serverless import ServerlessEndpoint
+from runpod_flash.core.resources.gpu import GpuGroup
+
+
+@patch.dict(os.environ, {}, clear=True)
+class TestQueueResourceOneFunction:
+    def test_second_remote_on_same_config_raises(self):
+        config = ServerlessEndpoint(
+            name="worker",
+            imageName="img:latest",
+            gpus=[GpuGroup.ANY],
+        )
+
+        @remote(resource_config=config)
+        async def first_fn():
+            pass
+
+        with pytest.raises(ValueError, match="already used by.*'first_fn'"):
+
+            @remote(resource_config=config)
+            async def second_fn():
+                pass
+
+    def test_separate_configs_are_independent(self):
+        config_a = ServerlessEndpoint(
+            name="worker-a",
+            imageName="img:latest",
+            gpus=[GpuGroup.ANY],
+        )
+        config_b = ServerlessEndpoint(
+            name="worker-b",
+            imageName="img:latest",
+            gpus=[GpuGroup.ANY],
+        )
+
+        @remote(resource_config=config_a)
+        async def fn_a():
+            pass
+
+        @remote(resource_config=config_b)
+        async def fn_b():
+            pass
+
+        # no error raised
+
+    def test_lb_resource_allows_multiple_functions(self):
+        from runpod_flash.core.resources.load_balancer_sls_resource import (
+            LoadBalancerSlsResource,
+        )
+
+        lb = LoadBalancerSlsResource(name="lb", imageName="img:latest")
+
+        @remote(resource_config=lb, method="POST", path="/a")
+        async def route_a():
+            pass
+
+        @remote(resource_config=lb, method="POST", path="/b")
+        async def route_b():
+            pass
+
+        # no error raised
+
+    def test_error_message_includes_both_function_names(self):
+        config = ServerlessEndpoint(
+            name="my-worker",
+            imageName="img:latest",
+            gpus=[GpuGroup.ANY],
+        )
+
+        @remote(resource_config=config)
+        async def greet():
+            pass
+
+        with pytest.raises(ValueError, match="'greet'.*'add'"):
+
+            @remote(resource_config=config)
+            async def add():
+                pass