[CI] Add E2E CI for Tinker API with SkyRLTrainBackend

.github/workflows/gpu_e2e_ci_tinker.yaml

@@ -0,0 +1,47 @@
+name: SkyRL-GPU-E2E-CI-Tinker
+
+on:
+  schedule:
+    - cron: '17 8 * * *'   # Every day at 08:17 UTC, offset from the other nightly E2Es
+  workflow_dispatch:
+
+permissions:
+  checks: write   # for status checks to appear
+  contents: read
+
+jobs:
+
+  skyrl_gpu_e2e_test_tinker:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        shell: bash
+        working-directory: .
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python
+        # This is the version of the action for setting up Python, not the Python version.
+        uses: actions/setup-python@v5
+        with:
+          # Semantic version range syntax or exact version of a Python version
+          python-version: '3.12'
+          cache: 'pip'
+      - name: Install the latest version of uv
+        uses: astral-sh/setup-uv@v6
+        with:
+          activate-environment: true
+      - name: Install basic dependencies
+        run: uv pip install anyscale==0.24.79 typer==0.9.0
+      - name: Install envsubst
+        run: sudo apt-get update && sudo apt-get install -y gettext-base
+      - name: Tinker rl_loop convergence test
+        env:
+          ANYSCALE_CLI_TOKEN: ${{ secrets.ANYSCALE_CLI_TOKEN }}
+          ANYSCALE_HOST: https://console.anyscale.com
+          WANDB_API_KEY: ${{ secrets.WANDB_API_KEY }}
+        run: |
+          envsubst < ci/anyscale_gpu_e2e_test_tinker.yaml > ci/anyscale_gpu_e2e_test_tinker_envsubst.yaml
+          anyscale job submit -f ci/anyscale_gpu_e2e_test_tinker_envsubst.yaml --timeout 12000
+          anyscale job wait --cloud sky-anyscale-aws-us-east-1 --name skyrl-train-gpu-e2e-test-tinker --timeout 12000
+          rm -f ci/anyscale_gpu_e2e_test_tinker_envsubst.yaml

ci/anyscale_gpu_e2e_test_tinker.yaml

@@ -0,0 +1,10 @@
+name: skyrl-train-gpu-e2e-test-tinker
+entrypoint: bash ci/gpu_e2e_test_run_tinker.sh
+image_uri: novaskyai/skyrl-train-ray-2.51.1-py3.12-cu12.8 # (Optional) Exclusive with `containerfile`.
+cloud: sky-anyscale-aws-us-east-1
+ray_version: "2.51.1"
+compute_config: l4_ci
+working_dir: . # (Optional) Use current working directory "." as the working_dir. Can be any local path or remote .zip file in cloud storage.
+env_vars:
+  WANDB_API_KEY: $WANDB_API_KEY
+max_retries: 1 # (Optional) Maximum number of times the job will be retried before being marked failed. Defaults to `1`.

ci/gpu_e2e_test_run_tinker.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+bash tests/train/gpu_e2e_test/gsm8k_tinker.sh

skyrl/backends/skyrl_train/inference_servers/common.py

@@ -13,6 +13,12 @@
 
 logger = logging.getLogger(__name__)
 
+# Stride between successive server actors' (or groups') start_port values.
+# Each actor's `find_and_reserve_port` increments by 1 on conflict, so this
+# stride must be larger than the max number of conflicts an actor could see
+# inside its window.
+SERVER_PORT_STRIDE = 100
+
 
 @dataclass
 class ServerInfo:
@@ -85,8 +91,9 @@ def find_and_reserve_port(start_port: int) -> Tuple[int, socket.socket]:
         (port, socket) -- caller must close the socket before rebinding.
     """
     port = start_port
+    end_port = start_port + SERVER_PORT_STRIDE
     sock: socket.socket | None = None
-    while True:
+    while port < end_port:
         try:
             sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@@ -97,5 +104,7 @@ def find_and_reserve_port(start_port: int) -> Tuple[int, socket.socket]:
             if sock:
                 sock.close()
             port += 1
-            if port > 65535:
-                raise RuntimeError(f"No available port found starting from {start_port}")
+    raise RuntimeError(
+        f"No available port found in [{start_port}, {end_port}). "
+        f"Free up the port range or raise SERVER_PORT_STRIDE."
+    )

skyrl/backends/skyrl_train/inference_servers/remote_inference_client.py

@@ -1100,16 +1100,26 @@ async def update_lora_from_disk(
         lora_path: str,
     ) -> Dict[str, Any]:
         """
-        Update LoRA adapter weights by loading from disk on all backend servers via /v1/load_lora_adapter.
+        Update LoRA adapter weights by loading from disk on all backend servers
+        via the SkyRL custom /skyrl/v1/load_lora_adapter endpoint.
 
-        Always loads under self.active_lora_name so the same slot is reused across
-        weight syncs.
+        Always loads under self.active_lora_name so the same lora_int_id slot
+        is reused across weight syncs.
 
-        After loading, generation requests will automatically use the LoRA adapter
-        by setting the model name to the LoRA adapter name.
+        TODO(aaron): switch back to /v1/load_lora_adapter (with "load_inplace": True)
+        once the upstream fix in https://github.com/vllm-project/vllm/pull/41482
+        lands in a vLLM release we depend on.
 
-        TODO(aaron): remove _unload_on_server and add back "load_inplace": True to payload after
-        vllm lora disk load bug is fixed.
+        The custom endpoint (defined in vllm_server_actor.py) wraps add_lora
+        with load_inplace=True (so the engine reloads the freshly-written
+        safetensors) and then resets the cached LoRARequest's load_inplace=False
+        (so subsequent generates don't reload from disk on every step). This
+        avoids two vLLM 0.19.0 bugs that surface under colocate_all + tp=1 +
+        num_engines>=2 — see vllm_server_actor.py:_skyrl_load_lora_adapter for
+        the detailed explanation.
+
+        After loading, generation requests will automatically use the LoRA
+        adapter by setting the model name to the LoRA adapter name.
 
         Args:
             lora_path: Path to the LoRA adapter on disk (must be accessible from servers).
@@ -1121,30 +1131,17 @@ async def update_lora_from_disk(
             raise ValueError("active_lora_name must be set on RemoteInferenceClient before loading a LoRA adapter.")
 
         lora_name = self.active_lora_name
-        # Both endpoints return plain text on success or JSON ErrorResponse on failure,
-        # so we use a session.post directly rather than _call_all_servers (which expects JSON).
         session = await self._get_session()
 
-        async def _unload_on_server(server_url: str) -> None:
-            """Remove the cached LoRARequest server-side. 404 on the first sync is expected."""
-            url = f"{server_url}/v1/unload_lora_adapter"
-            async with session.post(url, json={"lora_name": lora_name}) as resp:
-                if resp.status == 404:
-                    return
-                if resp.status >= 400:
-                    body = await resp.json()
-                    raise_for_status(resp, body)
-
         async def _load_on_server(server_url: str):
-            url = f"{server_url}/v1/load_lora_adapter"
+            url = f"{server_url}/skyrl/v1/load_lora_adapter"
             payload = {"lora_name": lora_name, "lora_path": lora_path}
             async with session.post(url, json=payload) as resp:
                 if resp.status >= 400:
                     body = await resp.json()
                     raise_for_status(resp, body)
-                return server_url, {"status": resp.status, "body": await resp.text()}
+                return server_url, await resp.json()
 
-        await asyncio.gather(*[_unload_on_server(url) for url in self.server_urls])
         results = await asyncio.gather(*[_load_on_server(url) for url in self.server_urls])
 
         logger.info(f"Loaded LoRA adapter '{lora_name}' from {lora_path}")

skyrl/backends/skyrl_train/inference_servers/server_group.py

@@ -11,6 +11,7 @@
 from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
 
 from skyrl.backends.skyrl_train.inference_servers.common import (
+    SERVER_PORT_STRIDE,
     ServerInfo,
 )
 from skyrl.backends.skyrl_train.inference_servers.protocols import ServerActorProtocol
@@ -184,7 +185,7 @@ def _create_actors(self) -> List[Any]:
 
             actor = ServerActorClass.remote(
                 self._cli_args,
-                self._start_port + server_idx,
+                self._start_port + server_idx * SERVER_PORT_STRIDE,
                 server_idx=server_idx,
                 bundle_indices=bundle_indices,
                 dp_size=self._num_servers if self._enable_dp else -1,

skyrl/backends/skyrl_train/inference_servers/setup.py

@@ -14,6 +14,7 @@
     get_ray_pg_ready_with_timeout,
 )
 
+from .common import SERVER_PORT_STRIDE
 from .remote_inference_client import RemoteInferenceClient
 from .server_group import ServerGroup
 from .utils import (
@@ -107,7 +108,7 @@ def create_inference_servers(
             ServerGroup(
                 cli_args=copy.deepcopy(pd_cli_args),
                 num_servers=ie_cfg.data_parallel_size,
-                start_port=VLLM_START_PORT + i * servers_per_group,
+                start_port=VLLM_START_PORT + i * servers_per_group * SERVER_PORT_STRIDE,
                 placement_group=prefill_pg,
                 placement_group_bundle_offset=i * gpus_per_server * servers_per_group,
                 enable_dp=ie_cfg.data_parallel_size > 1,
@@ -125,7 +126,7 @@ def create_inference_servers(
             ServerGroup(
                 cli_args=copy.deepcopy(pd_cli_args),
                 num_servers=ie_cfg.data_parallel_size,
-                start_port=VLLM_START_PORT + (num_prefill + i) * servers_per_group,
+                start_port=VLLM_START_PORT + (num_prefill + i) * servers_per_group * SERVER_PORT_STRIDE,
                 placement_group=decode_pg,
                 placement_group_bundle_offset=decode_bundle_offset + i * gpus_per_server * servers_per_group,
                 enable_dp=ie_cfg.data_parallel_size > 1,
@@ -183,7 +184,7 @@ def create_inference_servers(
                 cli_args=cli_args,
                 num_servers=ie_cfg.data_parallel_size,
                 placement_group=placement_group,
-                start_port=VLLM_START_PORT + i * ie_cfg.data_parallel_size,
+                start_port=VLLM_START_PORT + i * ie_cfg.data_parallel_size * SERVER_PORT_STRIDE,
                 enable_dp=ie_cfg.data_parallel_size > 1,
                 distributed_executor_backend=ie_cfg.distributed_executor_backend,
                 placement_group_bundle_offset=i * gpus_per_server * ie_cfg.data_parallel_size,

skyrl/backends/skyrl_train/inference_servers/vllm_server_actor.py

@@ -22,6 +22,7 @@
     init_app_state,
 )
 from vllm.inputs import TokensPrompt
+from vllm.lora.request import LoRARequest
 from vllm.sampling_params import SamplingParams as VLLMSamplingParams
 from vllm.usage.usage_lib import UsageContext
 from vllm.utils import random_uuid
@@ -360,6 +361,47 @@ async def _reset_prefix_cache(request: Request):
             await engine.reset_prefix_cache()
             return {"status": "ok"}
 
+        @app.post("/skyrl/v1/load_lora_adapter")
+        async def _skyrl_load_lora_adapter(request: Request):
+            """Load a LoRA adapter from disk, replacing any existing adapter
+            under the same name in place. Used by RemoteInferenceClient.update_lora_from_disk.
+
+            TODO(aaron): remove this endpoint and route update_lora_from_disk back
+            through /v1/load_lora_adapter once the upstream fix in
+            https://github.com/vllm-project/vllm/pull/41482 lands in a vLLM release we depend on.
+            """
+            body = await request.json()
+            lora_name = body.get("lora_name")
+            lora_path = body.get("lora_path")
+            if not lora_name or not lora_path:
+                raise HTTPException(
+                    status_code=400,
+                    detail="Both 'lora_name' and 'lora_path' must be provided.",
+                )
+
+            models = request.app.state.openai_serving_models
+            async with models.lora_resolver_lock[lora_name]:
+                lora_int_id = (
+                    models.lora_requests[lora_name].lora_int_id
+                    if lora_name in models.lora_requests
+                    else models.lora_id_counter.inc(1)
+                )
+                lora_request = LoRARequest(
+                    lora_name=lora_name,
+                    lora_int_id=lora_int_id,
+                    lora_path=lora_path,
+                    load_inplace=True,
+                )
+                await models.engine_client.add_lora(lora_request)
+                lora_request.load_inplace = False
+                models.lora_requests[lora_name] = lora_request
+
+            return {
+                "status": "ok",
+                "lora_name": lora_name,
+                "lora_int_id": lora_int_id,
+            }
+
         # NOTE (sumanthrh): We use a custom generate endpoint /skyrl/v1/generate because the native
         # endpoint /inference/v1/generate does not support returning routed expert IDs.
         # TODO (sumanthrh): Migrate back to /inference/v1/generate once this is fixed on the vllm side

tests/train/gpu_e2e_test/gsm8k_tinker.sh

@@ -0,0 +1,65 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+RUN_NAME="run_$(date +%Y%m%d%H)"
+PROJECT_NAME="gsm8k_tinker_ci"
+SCRIPT_DIR=$(dirname $(realpath $0))
+SKYRL_REPO_ROOT=$(realpath "$SCRIPT_DIR/../../..")
+LOG_DIR="$HOME/tinker_logs/$RUN_NAME"
+mkdir -p "$LOG_DIR"
+
+# TODO: tighten thresholds after 3-5 nightly runs (5% allowance from min observed),
+# matching the convention in gsm8k_colocate.sh.
+REWARD_MIN_VALUE=0.0
+
+BACKEND_CONFIG='{"trainer.placement.colocate_all": true, "trainer.placement.policy_num_gpus_per_node": 4, "trainer.micro_forward_batch_size_per_gpu": 8, "trainer.micro_train_batch_size_per_gpu": 8, "generator.inference_engine.num_engines": 4, "generator.inference_engine.tensor_parallel_size": 1, "generator.inference_engine.backend": "vllm", "generator.inference_engine.run_engines_locally": true, "generator.inference_engine.weight_sync_backend": "nccl", "generator.inference_engine.async_engine": true, "generator.inference_engine.gpu_memory_utilization": 0.8, "generator.batched": true}'
+
+# Start tinker server in its own process group so we can clean up the engine subprocess too.
+setsid uv run --extra tinker --extra fsdp -m skyrl.tinker.api \
+  --base-model "Qwen/Qwen3-0.6B" --backend fsdp --port 8000 \
+  --backend-config "$BACKEND_CONFIG" >"$LOG_DIR/server.log" 2>&1 &
+SERVER_PID=$!
+trap 'kill -TERM -- -$SERVER_PID 2>/dev/null || true; sleep 5; kill -KILL -- -$SERVER_PID 2>/dev/null || true' EXIT
+
+deadline=$(( $(date +%s) + 1800 ))
+until curl -sSf http://localhost:8000/docs >/dev/null 2>&1; do
+  if (( $(date +%s) > deadline )); then
+    echo "Tinker server did not become ready within 30 minutes" >&2
+    tail -n 200 "$LOG_DIR/server.log" >&2 || true
+    exit 1
+  fi
+  if ! kill -0 $SERVER_PID 2>/dev/null; then
+    echo "Tinker server exited early" >&2
+    tail -n 200 "$LOG_DIR/server.log" >&2 || true
+    exit 1
+  fi
+  sleep 5
+done
+
+COOKBOOK_DIR="$HOME/tinker-cookbook"
+[ -d "$COOKBOOK_DIR" ] || git clone --depth 1 https://github.com/thinking-machines-lab/tinker-cookbook.git "$COOKBOOK_DIR"
+
+# math_rl.train builds on tinker_cookbook/rl/train.py and exposes wandb_project /
+# wandb_name natively, so we get the same wandb-driven flow as the other E2E
+# nightlies (no client-side metrics publisher needed).
+cd "$COOKBOOK_DIR"
+TINKER_API_KEY=tml-dummy uv run --extra math-rl --extra wandb --with tinker --with datasets --with torch \
+  python -m tinker_cookbook.recipes.math_rl.train \
+  base_url=http://localhost:8000 \
+  model_name="Qwen/Qwen3-0.6B" \
+  env=gsm8k \
+  log_path="$LOG_DIR" \
+  groups_per_batch=512 \
+  group_size=4 \
+  max_tokens=512 \
+  max_steps=14 \
+  eval_every=10000 \
+  save_every=10000 \
+  wandb_project="$PROJECT_NAME" \
+  wandb_name="$RUN_NAME" \
+  behavior_if_log_dir_exists=delete
+
+cd "$SKYRL_REPO_ROOT"
+uv run --isolated --extra fsdp "$SCRIPT_DIR/get_summary.py" \
+  --run_name "$RUN_NAME" --project_name "$PROJECT_NAME" \
+  --asserts "env/all/reward/total >= $REWARD_MIN_VALUE"