From dc4175d3b17e54cf853bcdb7197949ac8829d8be Mon Sep 17 00:00:00 2001 From: YusefSyed <211442445+YusefSyed@users.noreply.github.com> Date: Tue, 1 Sep 2026 23:29:04 -0400 Subject: [PATCH] fix(controller): honor agent URL for local workers --- agentlightning/controller/local_reconciler.py | 9 ++- tests/controller/test_local_reconciler_env.py | 64 +++++++++++++++++++ 2 files changed, 68 insertions(+), 5 deletions(-) create mode 100644 tests/controller/test_local_reconciler_env.py diff --git a/agentlightning/controller/local_reconciler.py b/agentlightning/controller/local_reconciler.py index 7ab9acb67..faca4d233 100644 --- a/agentlightning/controller/local_reconciler.py +++ b/agentlightning/controller/local_reconciler.py @@ -205,16 +205,15 @@ async def _spawn_for(self, rollout: Rollout) -> bool: raise ValueError("invalid rollout config: missing config.local.agent_class") agent_class = rollout.config.local.agent_class mode = "train" if rollout.is_train else "val" + agent_url = self._config.agl_server.get("agent_url", None) + agent_base_url = str(agent_url or self._config.agl_server.url).rstrip("/") env = { **os.environ, "AGL_KEY": str(self._config.agl_server.key or ""), "AGL_OPENAI_BASE_URL": ( - f"{self._config.agl_server.url}/proxy/rollout/{rollout.rollout_id}" - f"/attempt/{attempt_id}/mode/{mode}/openai/v1" - ), - "AGL_EVENT_URL": ( - f"{self._config.agl_server.url}/api/rollouts/{rollout.rollout_id}/attempt/{attempt_id}/events" + f"{agent_base_url}/proxy/rollout/{rollout.rollout_id}/attempt/{attempt_id}/mode/{mode}/openai/v1" ), + "AGL_EVENT_URL": (f"{agent_base_url}/api/rollouts/{rollout.rollout_id}/attempt/{attempt_id}/events"), } env.update(_build_env_from_map(rollout.input, rollout.config.local.env_map)) proc = await asyncio.create_subprocess_exec( diff --git a/tests/controller/test_local_reconciler_env.py b/tests/controller/test_local_reconciler_env.py new file mode 100644 index 000000000..a7604b73e --- /dev/null +++ b/tests/controller/test_local_reconciler_env.py @@ -0,0 +1,64 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Unit tests for local reconciler worker environment variables.""" + +import asyncio +from unittest.mock import AsyncMock + +import pytest +from omegaconf import OmegaConf + +from agentlightning.client import AgentLightningAsyncClient +from agentlightning.controller.local_reconciler import LocalReconciler +from agentlightning.schemas import Rollout, RolloutConfig, RolloutLifecycleStatus, RolloutLocalConfig + + +@pytest.mark.parametrize( + ("agent_url", "expected_base_url"), + [ + ("http://agent-gateway:8080/", "http://agent-gateway:8080"), + (None, "http://controller:8080"), + ], +) +@pytest.mark.asyncio +async def test_spawn_uses_agent_url_for_worker_endpoints( + monkeypatch: pytest.MonkeyPatch, + agent_url: str | None, + expected_base_url: str, +) -> None: + api = AsyncMock(spec=AgentLightningAsyncClient) + config = OmegaConf.create( + { + "runner_type": "local", + "agl_server": { + "url": "http://controller:8080", + "agent_url": agent_url, + "key": "secret", + }, + "local_runner": {"maximum_size": 1, "poll_interval": 0.01}, + } + ) + rollout = Rollout( + rollout_id="rollout-1", + input={"question": "1 + 1"}, + config=RolloutConfig(local=RolloutLocalConfig(agent_class="example.Agent")), + status=RolloutLifecycleStatus(created_at=1.0, updated_at=1.0), + ) + reconciler = LocalReconciler(api, config) + patch = AsyncMock(return_value=True) + monkeypatch.setattr(reconciler, "_patch", patch) + proc = AsyncMock(spec=asyncio.subprocess.Process) + proc.pid = 123 + proc.returncode = None + create_subprocess = AsyncMock(return_value=proc) + monkeypatch.setattr(asyncio, "create_subprocess_exec", create_subprocess) + + assert await reconciler._spawn_for(rollout) + + spawn_call = create_subprocess.await_args + assert spawn_call is not None + env = spawn_call.kwargs["env"] + assert env["AGL_KEY"] == "secret" + assert env["AGL_OPENAI_BASE_URL"] == (f"{expected_base_url}/proxy/rollout/rollout-1/attempt/0/mode/train/openai/v1") + assert env["AGL_EVENT_URL"] == f"{expected_base_url}/api/rollouts/rollout-1/attempt/0/events" + patch.assert_awaited_once()