feat(service): add external model API with safety guards

Add support for routing chat completions to external OpenAI-compatible
APIs (e.g. GPT-4o) through the unified gateway/router/data-proxy stack,
with interaction caching, reward assignment, and trajectory export.

Key changes:
- External model registration flow: gateway -> router -> data proxy
- Proxy forwarding for streaming and non-streaming chat completions
- ExternalSessionData for interaction caching and trajectory export
- Controller external_mode: skip inference server launch, validate config
- Fail-fast on missing external_api_key to prevent admin key leakage
- Rollback router entry when data proxy registration fails
- Guard pause/continue against None inf_bridge in external mode
- Only cache successful external API responses
- Assert external mode requires workflow=None and group_size=1
- /v1/chat/completions and /v1/models OpenAI-compatible aliases
- HITL demo and online_rollout support for --external-url flags
- Comprehensive unit and integration tests (1150+ lines)

Author: nuzant

areal/experimental/inference_service/controller/config.py

@@ -54,3 +54,9 @@ class GatewayControllerConfig:
 
     # -- OpenAI proxy configuration (for agent-like workflows) ---------------
     openai: OpenAIProxyConfig = field(default_factory=lambda: OpenAIProxyConfig())
+
+    # -- External model API ------------------------------------------------
+    external_api_url: str | None = None
+    external_api_key: str | None = None
+    external_api_model: str | None = None
+    external_model_name: str = "ext-model"

areal/experimental/inference_service/controller/controller.py

@@ -82,13 +82,15 @@ def __init__(
         config: GatewayControllerConfig,
         scheduler: Scheduler,
     ) -> None:
-        from areal.api.alloc_mode import ModelAllocation
-
         self.config = config
         self.scheduler = scheduler
 
-        # Parse allocation from config.backend
-        self.rollout_alloc = ModelAllocation.from_str(config.backend)
+        if config.external_api_url is not None:
+            self.rollout_alloc = None
+        else:
+            from areal.api.alloc_mode import ModelAllocation
+
+            self.rollout_alloc = ModelAllocation.from_str(config.backend)
 
         # Worker management
         self.workers: list[Worker] = []
@@ -191,6 +193,15 @@ def initialize(
 
         logger.info("GatewayInferenceController initialized (role=%s)", role)
 
+        if self.config.external_api_url:
+            self._register_external_model()
+            logger.info(
+                "External model mode: url=%s, model=%s, name=%s",
+                self.config.external_api_url,
+                self.config.external_api_model,
+                self.config.external_model_name,
+            )
+
     async def _async_initialize(
         self,
         server_args: dict[str, Any] | None,
@@ -208,6 +219,8 @@ async def _async_initialize(
         * **server_infos is not None** — SGLang servers already exist so
           we only fork data proxy on every worker; fork router + gateway
           on worker 0.
+        * **external_mode** — skip inference servers entirely; data proxies
+          start with an empty ``--backend-addr``.
         """
         from dataclasses import asdict
 
@@ -216,30 +229,40 @@ async def _async_initialize(
         from areal.api.cli_args import SchedulingSpec, SchedulingStrategy
         from areal.api.scheduler_api import Job
 
-        alloc = self.rollout_alloc
-        dp_size = alloc.parallel.dp_size
         cfg = self.config
         admin_api_key = self.config.openai.admin_api_key
 
-        inf_backend = alloc.backend
+        if self.external_mode:
+            dp_size = 1
+            inf_backend = None
+        else:
+            alloc = self.rollout_alloc
+            dp_size = alloc.parallel.dp_size
+            inf_backend = alloc.backend
 
         # ==================================================================
         # Step 0: Always create dp_size RPCGuard workers
         # ==================================================================
-        inf_spec = SchedulingSpec(**asdict(cfg.scheduling_spec[0]))
-        instance_size = alloc.parallel.tp_size * alloc.parallel.pp_size
-        if server_infos is not None:
-            # Pre-existing inference servers — RPCGuard workers only host
-            # CPU services (data proxy, router, gateway), no GPUs needed.
-            inf_spec.gpu = 0
+        if self.external_mode:
+            inf_spec = SchedulingSpec(
+                task_type="worker",
+                port_count=2,
+                gpu=0,
+                mem=8,
+                cmd="python -m areal.experimental.inference_service.guard",
+            )
         else:
-            inf_spec.cpu *= instance_size
-            inf_spec.mem *= instance_size
-            if inf_spec.gpu > 0:
-                inf_spec.gpu = instance_size
-
-        # Override cmd to launch RPCGuard instead of RPC server
-        inf_spec.cmd = "python -m areal.experimental.inference_service.guard"
+            inf_spec = SchedulingSpec(**asdict(cfg.scheduling_spec[0]))
+            instance_size = alloc.parallel.tp_size * alloc.parallel.pp_size
+            if server_infos is not None:
+                inf_spec.gpu = 0
+            else:
+                inf_spec.cpu *= instance_size
+                inf_spec.mem *= instance_size
+                if inf_spec.gpu > 0:
+                    inf_spec.gpu = instance_size
+            # Override cmd to launch RPCGuard instead of RPC server
+            inf_spec.cmd = "python -m areal.experimental.inference_service.guard"
 
         inf_role = f"{self._worker_role}{self._INF_SUFFIX}"
         inf_job = Job(
@@ -256,9 +279,11 @@ async def _async_initialize(
         logger.info("RPCGuard workers ready: %s", [w.id for w in inf_workers])
 
         # ==================================================================
-        # Step 1: Launch inference servers (skip when pre-existing)
+        # Step 1: Launch inference servers (skip in external mode or when pre-existing)
         # ==================================================================
-        if server_infos is not None:
+        if self.external_mode:
+            logger.info("External mode — skipping inference server launch")
+        elif server_infos is not None:
             # Pre-existing servers — just record their addresses
             self.server_infos = server_infos
             self._inf_addrs = [
@@ -327,7 +352,6 @@ def _build_launch_cmd(host: str, port: int) -> list[str]:
             else:
                 raise ValueError(f"Unsupported inference backend: {inf_backend!r}")
 
-            # For each RPCGuard worker: alloc port, build cmd, fork server
             for rank, worker in enumerate(inf_workers):
                 guard_addr = (
                     f"http://{format_hostport(worker.ip, int(worker.worker_ports[0]))}"
@@ -447,12 +471,15 @@ def _build_launch_cmd(host: str, port: int) -> list[str]:
                 f"http://{format_hostport(worker.ip, int(worker.worker_ports[0]))}"
             )
             # Each data proxy connects to its corresponding inference server
-            data_proxy_cmd = data_proxy_base_cmd + [
-                "--backend-addr",
-                self._inf_addrs[rank],
-                "--backend-type",
-                inf_backend or "sglang",
-            ]
+            if self.external_mode:
+                data_proxy_cmd = data_proxy_base_cmd + ["--backend-addr", ""]
+            else:
+                data_proxy_cmd = data_proxy_base_cmd + [
+                    "--backend-addr",
+                    self._inf_addrs[rank],
+                    "--backend-type",
+                    inf_backend or "sglang",
+                ]
             data_proxy_host, data_proxy_port = self._fork_on_guard(
                 guard_addr=guard_addr,
                 role="data-proxy",
@@ -533,6 +560,40 @@ def _register_data_proxies_in_router(self) -> None:
                 worker_id,
             )
 
+    def _register_external_model(self) -> None:
+        import requests
+
+        cfg = self.config
+        if cfg.external_api_key is None:
+            raise ValueError(
+                "external_api_key must be set when using external model mode. "
+                "Without it, the internal admin API key would be leaked to the "
+                "external provider."
+            )
+        resp = requests.post(
+            f"{self._gateway_addr}/register_model",
+            json={
+                "name": cfg.external_model_name,
+                "url": cfg.external_api_url,
+                "model": cfg.external_api_model,
+            },
+            headers={"Authorization": f"Bearer {cfg.openai.admin_api_key}"},
+            timeout=cfg.request_timeout,
+        )
+        resp.raise_for_status()
+        logger.info(
+            "External model registered: name=%s url=%s model=%s "
+            "(requests will be sent to %s/chat/completions)",
+            cfg.external_model_name,
+            cfg.external_api_url,
+            cfg.external_api_model,
+            cfg.external_api_url.rstrip("/"),
+        )
+
+    @property
+    def external_mode(self) -> bool:
+        return self.config.external_api_url is not None
+
     def _start_online_callback_server(self) -> None:
         """Start callback server used by the router to deliver ready trajectories."""
         if self._callback_server is not None:
@@ -990,11 +1051,19 @@ async def chat_completion(
         if extra_body and isinstance(extra_body, dict):
             body.update(extra_body)
 
-        api_key = (
-            session_api_key
-            if session_api_key is not None
-            else self.config.openai.admin_api_key
-        )
+        if self.external_mode:
+            body["model"] = self.config.external_model_name
+            api_key = (
+                session_api_key
+                if session_api_key is not None
+                else self.config.external_api_key or self.config.openai.admin_api_key
+            )
+        else:
+            api_key = (
+                session_api_key
+                if session_api_key is not None
+                else self.config.openai.admin_api_key
+            )
         url = f"{self._gateway_addr}/chat/completions"
         headers = {
             "Content-Type": "application/json",
@@ -1201,6 +1270,19 @@ def _resolve_workflow(
         from areal.api.workflow_api import RolloutWorkflow
         from areal.utils.dynamic_import import import_from_string
 
+        # External mode only supports online mode (workflow=None)
+        if self.external_mode and workflow is not None:
+            raise ValueError(
+                "External model mode only supports online mode (workflow=None). "
+                "Agent-based workflows are not supported with external models."
+            )
+
+        if self.external_mode and group_size > 1:
+            raise ValueError(
+                "External model mode requires group_size=1, "
+                f"got group_size={group_size}."
+            )
+
         # (a) None → online mode: create InferenceServiceWorkflow without agent
         if workflow is None:
             from areal.experimental.inference_service.controller.workflow import (

areal/experimental/inference_service/controller/workflow.py

@@ -98,7 +98,7 @@ async def _export_interactions(
         session: aiohttp.ClientSession,
         session_id: str,
         trajectory_id: int | None = None,
-    ) -> dict[str, InteractionWithTokenLogpReward]:
+    ) -> dict[str, InteractionWithTokenLogpReward] | dict[str, Any]:
         url = f"{self.gateway_addr}/{_EXPORT_TRAJECTORIES_PATHNAME}"
         headers = {"Authorization": f"Bearer {self._admin_api_key}"}
         payload = {
@@ -110,13 +110,18 @@ async def _export_interactions(
         async with session.post(url, json=payload, headers=headers) as resp:
             resp.raise_for_status()
             data = await resp.json()
+
+        # External API trajectories are returned as-is without deserialization
+        if data.get("external_api"):
+            return data
+
         return _deserialize_interactions(data["interactions"])
 
     async def arun_episode(
         self,
         engine: InferenceEngine,
         data: dict[str, Any],
-    ) -> dict[str, InteractionWithTokenLogpReward] | None:
+    ) -> dict[str, InteractionWithTokenLogpReward] | dict[str, Any] | None:
         del engine
         http_session = await workflow_context.get_aiohttp_session()
         await self._grant_capacity(http_session)
@@ -190,23 +195,28 @@ async def _run_offline(
     async def _run_online(
         self,
         http_session: aiohttp.ClientSession,
-    ) -> dict[str, InteractionWithTokenLogpReward] | None:
+    ) -> dict[str, InteractionWithTokenLogpReward] | dict[str, Any] | None:
         logger.debug("Waiting for next ready online trajectory")
         export_request = await self.controller.wait_for_online_trajectory(
             timeout=self.timeout
         )
         if not export_request:
             return None
 
-        interactions = await self._export_interactions(
-            http_session,
-            export_request["session_id"],
-            trajectory_id=export_request["trajectory_id"],
+        session_id = export_request["session_id"]
+        trajectory_id = export_request["trajectory_id"]
+
+        result = await self._export_interactions(
+            http_session, session_id, trajectory_id=trajectory_id
         )
-        if not interactions:
+
+        if isinstance(result, dict) and result.get("external_api"):
+            return result
+
+        if not result:
             return None
 
-        last_id = next(reversed(interactions))
-        last_reward = interactions[last_id].reward
+        last_id = next(reversed(result))
+        last_reward = result[last_id].reward
         stats_tracker.get(workflow_context.stat_scope()).scalar(reward=last_reward)
-        return interactions
+        return result