feat(RL): add RL worker admin routes

components/src/dynamo/common/rl/__init__.py

@@ -0,0 +1,26 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared RL admin utilities."""
+
+from .admin import (
+    RLAdminValidationError,
+    RLRouteHandler,
+    RLRouteRegistry,
+    env_bool,
+    first_endpoint_response,
+    register_rl_routes,
+    require_lora_load_request,
+    require_lora_unload_request,
+)
+
+__all__ = [
+    "RLAdminValidationError",
+    "RLRouteHandler",
+    "RLRouteRegistry",
+    "env_bool",
+    "first_endpoint_response",
+    "register_rl_routes",
+    "require_lora_load_request",
+    "require_lora_unload_request",
+]

components/src/dynamo/common/rl/admin.py

@@ -0,0 +1,166 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared helpers for RL admin request-plane endpoints."""
+
+from __future__ import annotations
+
+import logging
+import os
+from collections.abc import AsyncIterator, Awaitable, Callable, Mapping
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+TRUE_ENV_VALUES = {"1", "true", "yes", "on"}
+
+RLRouteHandler = Callable[[dict[str, Any]], Awaitable[dict[str, Any] | None]]
+EndpointGenerator = Callable[[dict[str, Any]], AsyncIterator[dict[str, Any] | None]]
+
+
+class RLAdminValidationError(ValueError):
+    """Validation error whose message can be returned directly to RL clients."""
+
+
+def env_bool(name: str, default: bool = False) -> bool:
+    """Parse a boolean environment variable using Dynamo's common true values."""
+    value = os.environ.get(name)
+    if value is None:
+        return default
+    return value.strip().lower() in TRUE_ENV_VALUES
+
+
+async def first_endpoint_response(
+    endpoint_handler: EndpointGenerator,
+    body: dict[str, Any],
+) -> dict[str, Any]:
+    """Return the first response from an async-generator endpoint handler.
+
+    The generator is explicitly closed before returning so handlers that hold
+    resources across their yield (e.g. load_lora/unload_lora holding a per-LoRA
+    lock) release them promptly rather than waiting for garbage collection.
+    """
+    gen = endpoint_handler(body)
+    try:
+        async for response in gen:
+            return response or {"status": "ok"}
+        return {"status": "ok"}
+    finally:
+        aclose = getattr(gen, "aclose", None)
+        if aclose is not None:
+            await aclose()
+
+
+def require_lora_load_request(request: Mapping[str, Any] | None) -> tuple[str, str]:
+    """Validate the shared URI-based LoRA load request shape."""
+    if request is None or not isinstance(request, Mapping):
+        raise RLAdminValidationError(
+            "Request is required with 'lora_name' and 'source.uri'"
+        )
+
+    lora_name = request.get("lora_name")
+    if not isinstance(lora_name, str) or not lora_name:
+        raise RLAdminValidationError("'lora_name' is required and must be a string")
+
+    source = request.get("source")
+    if not source or not isinstance(source, Mapping):
+        raise RLAdminValidationError("'source' object is required in request")
+
+    lora_uri = source.get("uri")
+    if not isinstance(lora_uri, str) or not lora_uri:
+        raise RLAdminValidationError("'source.uri' is required and must be a string")
+
+    return lora_name, lora_uri
+
+
+def require_lora_unload_request(request: Mapping[str, Any] | None) -> str:
+    """Validate the shared LoRA unload request shape."""
+    if request is None or not isinstance(request, Mapping):
+        raise RLAdminValidationError("Request is required with 'lora_name' field")
+
+    lora_name = request.get("lora_name")
+    if not isinstance(lora_name, str) or not lora_name:
+        raise RLAdminValidationError("'lora_name' is required and must be a string")
+
+    return lora_name
+
+
+class RLRouteRegistry:
+    """Registry for worker RL admin route descriptors."""
+
+    def __init__(
+        self,
+        runtime: Any,
+        *,
+        logger_: logging.Logger | None = None,
+    ) -> None:
+        self._runtime = runtime
+        self._logger = logger_ or logger
+        self.routes: dict[str, RLRouteHandler] = {}
+
+    def add_route(self, name: str, handler: RLRouteHandler) -> None:
+        self.routes[name] = handler
+
+    def add_routes(self, routes: Mapping[str, RLRouteHandler]) -> None:
+        for name, handler in routes.items():
+            self.add_route(name, handler)
+
+    def describe(self) -> dict[str, Any]:
+        response: dict[str, Any] = {
+            "status": "ok",
+            "routes": sorted(self.routes),
+        }
+
+        system_url_fn = getattr(self._runtime, "system_status_server_url", None)
+        if callable(system_url_fn):
+            system_url = system_url_fn()
+            if system_url:
+                response["system_url"] = system_url
+
+        return response
+
+    async def dispatch(
+        self, request: Mapping[str, Any] | None = None
+    ) -> dict[str, Any]:
+        if request is None or not isinstance(request, Mapping):
+            return {"status": "error", "message": "rl_dispatch: request required"}
+
+        method = request.get("method")
+
+        if not isinstance(method, str) or not method:
+            return {"status": "error", "message": "rl_dispatch: missing 'method' (str)"}
+
+        if method != "routes":
+            return {
+                "status": "error",
+                "method": method,
+                "message": "rl request-plane endpoint only supports method='routes'",
+            }
+
+        if "kwargs" in request and not isinstance(request.get("kwargs"), Mapping):
+            return {
+                "status": "error",
+                "method": method,
+                "message": "rl_dispatch: 'kwargs' must be an object",
+            }
+
+        return self.describe()
+
+    async def dispatch_stream(
+        self, request: Mapping[str, Any] | None = None
+    ) -> AsyncIterator[dict[str, Any]]:
+        yield await self.dispatch(request)
+
+
+def register_rl_routes(
+    runtime: Any,
+    registry: RLRouteRegistry,
+    routes: Mapping[str, RLRouteHandler],
+    *,
+    enable_dispatch: bool,
+) -> None:
+    """Register worker system routes and optionally expose route descriptors."""
+    for name, handler in routes.items():
+        runtime.register_engine_route(name, handler)
+        if enable_dispatch:
+            registry.add_route(name, handler)

components/src/dynamo/common/tests/test_rl_admin.py

@@ -0,0 +1,152 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from dynamo.common.rl import (
+    RLAdminValidationError,
+    RLRouteRegistry,
+    first_endpoint_response,
+    register_rl_routes,
+    require_lora_load_request,
+    require_lora_unload_request,
+)
+
+pytestmark = [pytest.mark.pre_merge, pytest.mark.unit, pytest.mark.gpu_0]
+
+
+class _Runtime:
+    def __init__(self, system_url: str | None = None) -> None:
+        self.system_url = system_url
+        self.registered: list[tuple[str, object]] = []
+
+    def system_status_server_url(self) -> str | None:
+        return self.system_url
+
+    def register_engine_route(self, name: str, handler: object) -> None:
+        self.registered.append((name, handler))
+
+
+def test_route_registry_describes_routes() -> None:
+    runtime = _Runtime("http://worker:8081")
+    registry = RLRouteRegistry(runtime)
+
+    async def ping(body: dict) -> dict:
+        return {"status": "ok", "body": body}
+
+    registry.add_route("ping", ping)
+
+    routes = asyncio.run(registry.dispatch({"method": "routes"}))
+    assert routes == {
+        "status": "ok",
+        "routes": ["ping"],
+        "system_url": "http://worker:8081",
+    }
+
+    routes_with_kwargs = asyncio.run(
+        registry.dispatch({"method": "routes", "kwargs": {}})
+    )
+    assert routes_with_kwargs == routes
+
+
+def test_route_registry_rejects_request_plane_admin_execution() -> None:
+    registry = RLRouteRegistry(_Runtime())
+
+    response = asyncio.run(registry.dispatch({"method": "missing"}))
+
+    assert response["status"] == "error"
+    assert response["method"] == "missing"
+    assert (
+        response["message"] == "rl request-plane endpoint only supports method='routes'"
+    )
+
+
+def test_route_registry_rejects_non_object_kwargs_for_routes() -> None:
+    registry = RLRouteRegistry(_Runtime())
+
+    response = asyncio.run(registry.dispatch({"method": "routes", "kwargs": []}))
+
+    assert response == {
+        "status": "error",
+        "method": "routes",
+        "message": "rl_dispatch: 'kwargs' must be an object",
+    }
+
+
+def test_register_rl_routes_always_registers_engine_route() -> None:
+    runtime = _Runtime()
+    registry = RLRouteRegistry(runtime)
+
+    async def ping(body: dict) -> dict:
+        return {"status": "ok", "body": body}
+
+    register_rl_routes(runtime, registry, {"ping": ping}, enable_dispatch=False)
+
+    assert runtime.registered == [("ping", ping)]
+    assert registry.routes == {}
+
+    register_rl_routes(runtime, registry, {"ping": ping}, enable_dispatch=True)
+
+    assert registry.routes == {"ping": ping}
+
+
+def test_first_endpoint_response_returns_first_chunk() -> None:
+    async def endpoint(_body: dict):
+        yield {"status": "ok", "value": 1}
+        yield {"status": "ok", "value": 2}
+
+    response = asyncio.run(first_endpoint_response(endpoint, {}))
+
+    assert response == {"status": "ok", "value": 1}
+
+
+def test_lora_load_request_validation() -> None:
+    assert require_lora_load_request(
+        {"lora_name": "adapter", "source": {"uri": "file:///tmp/adapter"}}
+    ) == ("adapter", "file:///tmp/adapter")
+
+    try:
+        require_lora_load_request({"lora_name": "adapter"})
+    except RLAdminValidationError as exc:
+        assert str(exc) == "'source' object is required in request"
+    else:
+        raise AssertionError("expected validation error")
+
+
+def test_lora_unload_request_validation() -> None:
+    assert require_lora_unload_request({"lora_name": "adapter"}) == "adapter"
+
+    try:
+        require_lora_unload_request({})
+    except RLAdminValidationError as exc:
+        assert str(exc) == "'lora_name' is required and must be a string"
+    else:
+        raise AssertionError("expected validation error")
+
+    # Non-string scalars must be rejected, not str()-coerced.
+    for bad in ([], {}, 123, ["adapter"]):
+        try:
+            require_lora_unload_request({"lora_name": bad})
+        except RLAdminValidationError:
+            pass
+        else:
+            raise AssertionError(f"expected validation error for lora_name={bad!r}")
+
+
+def test_lora_load_request_rejects_non_string_fields() -> None:
+    # lora_name / source.uri must be strings (no str() coercion of lists/dicts).
+    for req in (
+        {"lora_name": ["a"], "source": {"uri": "file:///x"}},
+        {"lora_name": "a", "source": {"uri": {}}},
+        {"lora_name": "a", "source": {"uri": ["file:///x"]}},
+    ):
+        try:
+            require_lora_load_request(req)
+        except RLAdminValidationError:
+            pass
+        else:
+            raise AssertionError(f"expected validation error for {req!r}")