Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions resources_servers/interactive_browser/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,11 @@ Every acquired provider session is released exactly once — including when the
CDP connect fails after the session was created, and however often `close()` is
called (`tests/test_registry_and_sessions.py`).

A session seeded by an Environment Server carries its episode (`rollout_id`,
`attempt`) to the provider as session metadata, and `remote_cdp` logs the
provider's session id next to it when the session is acquired. That log line is
the join between a provider-side session and the rollout record that held it.

## Run

### 1. Backend contract tests (no GPU, no Gym serving stack)
Expand Down
17 changes: 12 additions & 5 deletions resources_servers/interactive_browser/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,7 @@ async def seed_session(
raise ValueError("resources_session_id is already bound to another episode or task")
return ResourcesSeedSessionResponse(resources_session_id=session_id)
spec = BrowserSeedSessionRequest.model_validate(body.task_data)
await self._open_session(session_id, spec)
await self._open_session(session_id, spec, episode_id=body.episode_id)
self._typed_identity[session_id] = (body.episode_id, body.task_id)
return ResourcesSeedSessionResponse(resources_session_id=session_id)

Expand Down Expand Up @@ -249,7 +249,9 @@ async def _release(self, session_id: str) -> None:
# let the close report success silently.
logger.warning("could not close the browser for session %s", session_id, exc_info=True)

async def _open_session(self, session_id: str, body: BrowserSeedSessionRequest) -> None:
async def _open_session(
self, session_id: str, body: BrowserSeedSessionRequest, episode_id: Optional[EpisodeId] = None
) -> None:
# Resolve a repo-relative initial_url (e.g. "site/index.html") to an
# absolute file:// URI, so example tasks don't hard-code machine paths.
initial_url = body.initial_url
Expand All @@ -260,9 +262,14 @@ async def _open_session(self, session_id: str, body: BrowserSeedSessionRequest)
# old browser first so we don't leak a session/process.
await self._release(session_id)

# The rollout id travels with the session so a remote provider can tag
# (and later account for) the browser it hands out.
backend = create_backend(self.config.backend, session_metadata={"rollout_session_id": session_id})
# Identifiers travel with the session so a remote provider can tag (and later
# account for) the browser it hands out. The episode is what the training side
# records; the session id alone is internal to whoever seeded it.
session_metadata = {"rollout_session_id": session_id}
if episode_id is not None:
session_metadata["rollout_id"] = episode_id.rollout_id
session_metadata["attempt"] = str(episode_id.attempt)
backend = create_backend(self.config.backend, session_metadata=session_metadata)
# `open()` unwinds its own partial state — including any provider
# session it acquired — before it raises.
await backend.open(initial_url)
Expand Down
10 changes: 9 additions & 1 deletion resources_servers/interactive_browser/browser/remote_cdp.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,13 @@ async def _connect(self, playwright):
provider_options=dict(self._provider_options),
)
handle = await self._provider.acquire(spec)
# The one line that joins a provider's session to the rollout that held it.
LOGGER.info(
"Provider %r session %s acquired for %s",
getattr(self._provider, "name", type(self._provider).__name__),
handle.session_id or "?",
spec.metadata,
)
if not handle.cdp_url:
# Release before raising: the provider may already hold a session
# even though it failed to give us a usable endpoint.
Expand Down Expand Up @@ -130,8 +137,9 @@ async def _release(self) -> None:
# Loud, not fatal: a session the provider still holds is a leaked
# resource, and silence here is how a run walks into its quota.
LOGGER.warning(
"Provider %r failed to release session %s: %r",
"Provider %r failed to release session %s for %s: %r",
getattr(self._provider, "name", type(self._provider).__name__),
handle.session_id or "?",
self.session_metadata,
exc,
)
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,22 @@ async def test_session_is_released_once_however_often_close_is_called(cdp_endpoi
assert provider.released[0].session_id == "s1"


@pytest.mark.asyncio
async def test_acquiring_a_session_logs_which_rollout_holds_it(cdp_endpoint, caplog):
"""The provider's session id and the rollout share one log line, so a leaked cloud
session can be traced to its rollout without matching timestamps."""
provider = FakeProvider(cdp_endpoint)
backend = RemoteCDPBackend(provider, session_metadata={"rollout_id": "rollout-7", "attempt": "0"})

with caplog.at_level("INFO", logger="browser.remote_cdp"):
await backend.open("about:blank")
await backend.close()

acquired = [r.getMessage() for r in caplog.records if "acquired for" in r.getMessage()]
assert len(acquired) == 1
assert "s1" in acquired[0] and "rollout-7" in acquired[0]


@pytest.mark.asyncio
async def test_session_is_released_when_the_cdp_connect_fails():
# Port 1 refuses connections: the session was acquired but is unusable, and
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ def browsers(monkeypatch):

def _create_backend(config, session_metadata=None):
browser = _Browser(open_delay_s=settings["open_delay_s"])
browser.session_metadata = session_metadata
created.append(browser)
return browser

Expand Down Expand Up @@ -304,3 +305,25 @@ def test_mcp_tool_calls_reach_the_browser_opened_under_the_callers_id(browsers)
assert called["result"].get("isError") is not True, called
assert len(browsers.created) == 1
assert server._session_id_to_state["resources-session"].answer == "42"


def test_the_browser_is_tagged_with_the_episode_the_training_side_records(browsers) -> None:
"""The resources_session_id is minted inside the Environment Server; only the episode
reaches the training record, so that is what a provider-side session must carry."""
client = TestClient(_server().setup_webserver())

client.post("/seed_session", json=_seed_json(episode={"rollout_id": "rollout-7", "attempt": 2}))

assert browsers.created[0].session_metadata == {
"rollout_session_id": "resources-session",
"rollout_id": "rollout-7",
"attempt": "2",
}


def test_an_agent_run_seed_has_no_episode_to_tag(browsers) -> None:
client = TestClient(_server().setup_webserver())

client.post("/seed_session", json={"initial_url": "https://example.com/"})

assert set(browsers.created[0].session_metadata) == {"rollout_session_id"}
Loading