fix: address PR review feedback for concurrency support

- Fix inline @endpoint() max_concurrency loss: persist into
  __remote_config__ so manifest builder can read it when the
  Endpoint facade is not reachable as a module-level variable.
- Fix scanner is_async for class-based workers: detect async
  public methods so the correct handler template is selected.
- Harden _inject_concurrency_modifier: raise ValueError if the
  expected start call string is absent instead of silently
  producing a handler without concurrency_modifier.
- Add rationale comment for the >100 concurrency warning threshold.
- Add tests for all new behaviors.

Author: deanq

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

@@ -378,6 +378,8 @@ def _generate_deployed_handler_code(
             else "# No function to import"
         )
 
+        # 100 is a soft upper bound: most GPU workloads saturate VRAM well
+        # below this. The warning nudges users to verify resource capacity.
         if max_concurrency > 100:
             logger.warning(
                 "High max_concurrency=%d for resource '%s'. Ensure your handler "
@@ -447,8 +449,14 @@ def _generate_deployed_handler_code(
     @staticmethod
     def _inject_concurrency_modifier(code: str, max_concurrency: int) -> str:
         """Replace the default runpod.serverless.start call with one including concurrency_modifier."""
+        start_call = 'runpod.serverless.start({"handler": handler})'
+        if start_call not in code:
+            raise ValueError(
+                "Unable to inject concurrency_modifier: expected "
+                f"{start_call!r} in generated handler code."
+            )
         return code.replace(
-            'runpod.serverless.start({"handler": handler})',
+            start_call,
             "runpod.serverless.start({\n"
             '        "handler": handler,\n'
             f'        "concurrency_modifier": lambda current: {max_concurrency},\n'

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

@@ -166,6 +166,7 @@ def _extract_deployment_config(
 
             try:
                 resource_config = None
+                remote_cfg = None
 
                 if config_variable and hasattr(module, config_variable):
                     resource_config = getattr(module, config_variable)
@@ -181,12 +182,18 @@ def _extract_deployment_config(
                     return config
 
                 # Extract max_concurrency from Endpoint facade before unwrapping.
-                # max_concurrency is a Flash concept that lives on Endpoint,
-                # not on the internal resource config (LiveServerless, etc.).
+                # For module-level Endpoint variables, read from the instance.
+                # For inline @Endpoint() decorators, the facade is gone by the
+                # time we reach here -- read from __remote_config__ instead.
                 if hasattr(resource_config, "_max_concurrency"):
                     mc = resource_config._max_concurrency
                     if mc > 1:
                         config["max_concurrency"] = mc
+                elif (
+                    isinstance(remote_cfg, dict)
+                    and remote_cfg.get("max_concurrency", 1) > 1
+                ):
+                    config["max_concurrency"] = remote_cfg["max_concurrency"]
 
                 # unwrap Endpoint facade to the internal resource config
                 if hasattr(resource_config, "_build_resource_config"):

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

@@ -274,6 +274,15 @@ def _metadata_from_remote_config(
         is_async = inspect.iscoroutinefunction(original) or inspect.iscoroutinefunction(
             obj
         )
+    elif is_class and target_class is not None:
+        # A class is async if any of its public methods are coroutines.
+        # This drives handler template selection for concurrent endpoints.
+        is_async = any(
+            inspect.iscoroutinefunction(m)
+            for name, m in vars(target_class).items()
+            if not name.startswith("_")
+            and (inspect.isfunction(m) or inspect.iscoroutinefunction(m))
+        )
 
     # function/class name: for classes, use the original class name.
     # for functions, use __name__ from the unwrapped function.

src/runpod_flash/endpoint.py

@@ -611,14 +611,24 @@ async def process(data: dict) -> dict: ...
 
         from .client import remote as remote_decorator
 
-        return remote_decorator(
+        decorated = remote_decorator(
             resource_config=resource_config,
             dependencies=self.dependencies,
             system_dependencies=self.system_dependencies,
             accelerate_downloads=self.accelerate_downloads,
             _internal=True,
         )(func_or_class)
 
+        # Persist max_concurrency into __remote_config__ so the manifest
+        # builder can read it for inline @Endpoint() decorators where the
+        # Endpoint facade is not reachable via a module-level variable.
+        if self._max_concurrency > 1:
+            remote_cfg = getattr(decorated, "__remote_config__", None)
+            if isinstance(remote_cfg, dict):
+                remote_cfg["max_concurrency"] = self._max_concurrency
+
+        return decorated
+
     # -- route decorators (lb mode) --
 
     def _route(self, method: str, path: str):

tests/unit/cli/commands/build_utils/test_handler_generator.py

@@ -814,3 +814,9 @@ def test_high_concurrency_logs_warning(caplog):
             generator = HandlerGenerator(manifest, build_dir)
             generator.generate_handlers()
         assert any("max_concurrency=150" in r.message for r in caplog.records)
+
+
+def test_inject_concurrency_modifier_raises_on_missing_start_call():
+    """_inject_concurrency_modifier raises if the start call string is absent."""
+    with pytest.raises(ValueError, match="Unable to inject concurrency_modifier"):
+        HandlerGenerator._inject_concurrency_modifier("some random code", 5)

tests/unit/cli/commands/build_utils/test_runtime_scanner.py

@@ -739,3 +739,71 @@ def test_discovers_in_subdirectory(self, tmp_path):
         functions = scanner.discover_remote_functions()
         assert len(functions) == 1
         assert functions[0].module_path == "workers.gpu_worker"
+
+
+class TestClassAsyncDetection:
+    """scanner detects is_async for class-based workers."""
+
+    def test_sync_class_not_async(self, tmp_path):
+        _write_worker(
+            tmp_path,
+            "sync_cls.py",
+            """\
+            from runpod_flash import remote, LiveServerless
+            cfg = LiveServerless(name="sync-cls")
+
+            @remote(cfg)
+            class SyncWorker:
+                def predict(self, x):
+                    return x
+            """,
+        )
+        scanner = RuntimeScanner(tmp_path)
+        functions = scanner.discover_remote_functions()
+        assert len(functions) == 1
+        assert functions[0].is_class is True
+        assert functions[0].is_async is False
+
+    def test_async_class_detected(self, tmp_path):
+        _write_worker(
+            tmp_path,
+            "async_cls.py",
+            """\
+            from runpod_flash import remote, LiveServerless
+            cfg = LiveServerless(name="async-cls")
+
+            @remote(cfg)
+            class AsyncWorker:
+                async def predict(self, x):
+                    return x
+            """,
+        )
+        scanner = RuntimeScanner(tmp_path)
+        functions = scanner.discover_remote_functions()
+        assert len(functions) == 1
+        assert functions[0].is_class is True
+        assert functions[0].is_async is True
+
+    def test_mixed_class_detected_as_async(self, tmp_path):
+        """Class with both sync and async methods is detected as async."""
+        _write_worker(
+            tmp_path,
+            "mixed_cls.py",
+            """\
+            from runpod_flash import remote, LiveServerless
+            cfg = LiveServerless(name="mixed-cls")
+
+            @remote(cfg)
+            class MixedWorker:
+                def setup(self):
+                    pass
+
+                async def predict(self, x):
+                    return x
+            """,
+        )
+        scanner = RuntimeScanner(tmp_path)
+        functions = scanner.discover_remote_functions()
+        assert len(functions) == 1
+        assert functions[0].is_class is True
+        assert functions[0].is_async is True

tests/unit/test_endpoint.py

@@ -1036,3 +1036,27 @@ def test_rejects_zero(self):
     def test_rejects_negative(self):
         with pytest.raises(ValueError, match="max_concurrency must be >= 1"):
             Endpoint(name="test", max_concurrency=-1)
+
+    def test_inline_decorator_persists_max_concurrency(self):
+        """Inline @Endpoint() persists max_concurrency into __remote_config__."""
+        ep = Endpoint(name="test", gpu=GpuGroup.ANY, max_concurrency=5)
+
+        async def generate(prompt: str) -> str:
+            return prompt
+
+        decorated = ep(generate)
+        remote_cfg = getattr(decorated, "__remote_config__", None)
+        assert remote_cfg is not None
+        assert remote_cfg["max_concurrency"] == 5
+
+    def test_inline_decorator_omits_default_max_concurrency(self):
+        """Inline @Endpoint() with default max_concurrency=1 does not add the key."""
+        ep = Endpoint(name="test", gpu=GpuGroup.ANY)
+
+        async def generate(prompt: str) -> str:
+            return prompt
+
+        decorated = ep(generate)
+        remote_cfg = getattr(decorated, "__remote_config__", None)
+        assert remote_cfg is not None
+        assert "max_concurrency" not in remote_cfg