feat(proxy): add proxy gateway and online RL training mode (#947)

* feat(proxy): add proxy gateway and online RL training mode

Add a lightweight proxy gateway that routes external user requests to
session-pinned backend proxy workers, enabling online RL training where
real users drive the rollout loop instead of scripted agents.

Key changes:
- New proxy gateway server with session routing, API-key-based auth,
  readiness queue for online-mode worker coordination, and session
  refresh with API key reuse
- New _OnlineAgent that registers workers as ready and blocks until
  an external user completes a full session lifecycle
- OpenAIProxyWorkflow gains mode="online" alongside inline/subproc
- RolloutController.start_proxy_gateway() for gateway lifecycle
- RLTrainer._init_proxy_for_online() with _EmptyDataLoader for
  online mode where data comes from external users
- Fix export_trajectories race condition: session_id is now required
  in the request body (resolved via admin auth) so export always
  targets the correct session even after API key remapping
- Worker start_session accepts caller-provided api_key for refresh/
  reuse with conflict detection (409 for active, cleanup for finished)
- Example scripts and documentation for the online RL workflow
- 36 new tests covering gateway, worker, and online agent

* fix

Author: garrett4wade

areal/api/cli_args.py

@@ -1378,17 +1378,18 @@ class OpenAIProxyConfig:
         default="inline",
         metadata={
             "help": (
-                "OpenAI proxy mode: 'inline' (in-process) or 'subproc' (subprocess). "
+                "OpenAI proxy mode: 'inline' (in-process), 'subproc' (subprocess), "
+                "or 'online' (external user sessions for online RL training). "
                 "`inline` mode runs the provided agent workflow directly in the same process. "
-                "It can use the provided `base_url` and `http_client` to reduce overhead. "
-                "`subproc` mode launches a separate process to run the agent with `OPENAI_BASE_URL` environment variable, "
-                "which offers more flexible deployment options at the cost of larger overhead."
+                "`subproc` mode launches a separate process to run the agent. "
+                "`online` mode waits for external users to complete sessions via "
+                "the proxy gateway URL, enabling online RL training."
             ),
-            "choices": ["inline", "subproc"],
+            "choices": ["inline", "subproc", "online"],
         },
     )
     tool_call_parser: str = field(
-        default="qwen3",
+        default="qwen",
         metadata={"help": "Parser for tool calls in model output."},
     )
     reasoning_parser: str = field(

areal/engine/sglang_remote.py

@@ -271,6 +271,9 @@ def get_version(self) -> int:
         """Get the current weight version."""
         return self._engine.get_version()
 
+    def set_proxy_gateway_addr(self, addr: str) -> None:
+        return self._engine.set_proxy_gateway_addr(addr)
+
     async def agenerate(self, req: ModelRequest) -> ModelResponse:
         """Asynchronously generate a response for the given request."""
         return await self._engine.agenerate(req)

areal/engine/vllm_remote.py

@@ -315,6 +315,9 @@ def get_version(self) -> int:
         """Get the current weight version."""
         return self._engine.get_version()
 
+    def set_proxy_gateway_addr(self, addr: str) -> None:
+        self._engine.set_proxy_gateway_addr(addr)
+
     async def agenerate(self, req: ModelRequest) -> ModelResponse:
         """Asynchronously generate a response for the given request."""
         return await self._engine.agenerate(req)

areal/experimental/openai/client.py

@@ -1042,7 +1042,7 @@ def __init__(
         self,
         engine: TRolloutEngine,
         tokenizer: "PreTrainedTokenizerFast",
-        tool_call_parser: str = "qwen3",
+        tool_call_parser: str = "qwen",
         reasoning_parser: str = "qwen3",
         engine_max_tokens: int | None = None,
         chat_template_type: str = "hf",

areal/experimental/openai/proxy/client_session.py

@@ -139,8 +139,14 @@ async def export_interactions(
         """Export interactions for this session via HTTP.
 
         This method should be called after the session context exits
-        (i.e., after `__aexit__` has ended the RL session), since
-        `/export_trajectories` waits for the session to finish.
+        (i.e., after ``__aexit__`` has ended the RL session), since
+        ``/export_trajectories`` waits for the session to finish.
+
+        The request always includes the explicit ``session_id`` so that
+        the server resolves the correct session regardless of any
+        API-key-to-session remapping that may have occurred during a
+        refresh cycle.  Admin auth is used because the session key is
+        not guaranteed to still map to this session.
 
         Parameters
         ----------
@@ -153,17 +159,22 @@ async def export_interactions(
         -------
         dict[str, InteractionWithTokenLogpReward]
             Dictionary mapping interaction IDs to their data
-        """
 
-        if self._session_api_key is None:
-            raise ValueError("Session API key is not set")
+        Raises
+        ------
+        ValueError
+            If ``session_id`` has not been set on this client.
+        """
+        if self.session_id is None:
+            raise ValueError("session_id must be set before exporting interactions")
 
         url = f"{self.base_url}{EXPORT_TRAJECTORIES_PATHNAME}"
         payload = {
+            "session_id": self.session_id,
             "discount": discount,
             "style": style,
         }
-        headers = self._session_auth_headers()
+        headers = self._admin_auth_headers()
         async with self._session.post(url, json=payload, headers=headers) as resp:
             resp.raise_for_status()
             data = await resp.json()

areal/experimental/openai/proxy/online_agent.py

@@ -0,0 +1,86 @@
+"""Internal agent for online training mode.
+
+``_OnlineAgent`` is NOT intended for direct use. It is automatically
+created by ``OpenAIProxyWorkflow`` when ``mode="online"``.  It registers
+the assigned proxy worker as "ready" on the proxy gateway, then blocks
+until an external user completes a full session lifecycle.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import aiohttp
+
+from areal.infra import workflow_context
+
+from .proxy_gateway import CompletedSessionInfo
+from .server import INTERNAL_WAIT_FOR_SESSION_PATHNAME
+
+
+class _OnlineAgent:
+    """Internal agent that waits for external user sessions.
+
+    Registers the assigned proxy worker as "ready" on the proxy
+    gateway, then blocks until an external user completes a full session
+    lifecycle (start_session → interact → set_reward → end_session) on
+    that worker.
+
+    Parameters
+    ----------
+    proxy_gateway_addr : str
+        HTTP address of the proxy gateway server.
+    admin_api_key : str
+        Admin API key for authenticating with the proxy gateway.
+    timeout : float
+        Maximum seconds to wait for a session completion.
+    """
+
+    def __init__(
+        self,
+        proxy_gateway_addr: str,
+        admin_api_key: str,
+        timeout: float = 3600.0,
+    ):
+        self.proxy_gateway_addr = proxy_gateway_addr
+        self.admin_api_key = admin_api_key
+        self.timeout = timeout
+
+    async def run(
+        self, data: dict[str, Any], **extra_kwargs: Any
+    ) -> CompletedSessionInfo:
+        """Wait for an external user to complete a session.
+
+        Parameters
+        ----------
+        data : dict
+            Ignored in online mode (dataloader yields empty dicts).
+        extra_kwargs : dict
+            Provided by ``OpenAIProxyWorkflow``:
+
+            - ``base_url``: proxy worker address
+            - ``api_key``: admin API key
+            - ``http_client``: ``httpx.AsyncClient`` (unused here)
+
+        Returns
+        -------
+        CompletedSessionInfo
+            Session credentials for trajectory export.
+        """
+        base_url = extra_kwargs["base_url"]  # proxy worker addr
+
+        url = f"{self.proxy_gateway_addr}/{INTERNAL_WAIT_FOR_SESSION_PATHNAME}"
+        headers = {"Authorization": f"Bearer {self.admin_api_key}"}
+        payload = {"worker_addr": base_url}
+
+        timeout = aiohttp.ClientTimeout(total=self.timeout)
+        session = await workflow_context.get_aiohttp_session()
+        async with session.post(
+            url,
+            json=payload,
+            headers=headers,
+            timeout=timeout,
+        ) as resp:
+            resp.raise_for_status()
+            result = await resp.json()
+        return CompletedSessionInfo(**result)