diff --git a/src/agent_env/agents/AGENTS.md b/src/agent_env/agents/AGENTS.md index 0ae4d98a9..0c1c0f558 100644 --- a/src/agent_env/agents/AGENTS.md +++ b/src/agent_env/agents/AGENTS.md @@ -120,8 +120,9 @@ Reference: `enable(MCP_CONFIG_V1, ...)`, and `claude_bridge.py` where it builds ### `cdp-config/v1`, `/ext/cdp-config` For an agent that drives a browser itself. A browser reaches the agent as an endpoint `{name, http_url, ws_url, -headers?, metadata?}` sent to `add`, and adding a name again replaces it; nothing in agent-env sends one yet (the -`browser_connect_cdp` task step will). The SDK implements `add`, `list` and `remove`, and passes the endpoints in +headers?, metadata?}` sent to `add`, and adding a name again replaces it. The `browser_connect_cdp` task step sends +one, `chromium` by default, with `metadata {browser}`. A reset ends the address, so a task runs `browser_connect_cdp` +again after `reset_env`, or gives the turn a `retry_config` that reruns it. The SDK implements `add`, `list` and `remove`, and passes the endpoints in `request.cdp_endpoints` every turn. Take yours with `request.cdp_endpoint()`, or `request.cdp_endpoint(name)` when the task gives several, and connect afresh each turn, since an address can change between turns. With no endpoint, or several and no name, it fails the turn as an `infra_error` (`cdp.no_endpoint`, `cdp.ambiguous`), so the agent never diff --git a/src/agent_env/task/episode/rules.py b/src/agent_env/task/episode/rules.py index 10f3d78c8..8f98ee689 100644 --- a/src/agent_env/task/episode/rules.py +++ b/src/agent_env/task/episode/rules.py @@ -133,6 +133,7 @@ def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[s "deploy_agent": _deploy_agent, "deploy_human_agent": _deploys_an_agent, "register_agent_triggers": _needs_the_agent, + "browser_connect_cdp": _needs_the_agent, "snapshot_agent_state": _needs_the_agent, "collect_artifacts": _collect_artifacts, "register_env_triggers": _register_env_triggers, diff --git a/src/agent_env/task_step/registry.py b/src/agent_env/task_step/registry.py index a4457a320..d51bb1249 100644 --- a/src/agent_env/task_step/registry.py +++ b/src/agent_env/task_step/registry.py @@ -27,6 +27,7 @@ def get_task_step_registry() -> dict[str, type["TaskStep"]]: def _builtin_registry() -> dict[str, type["TaskStep"]]: + from agent_env.task_step.task_steps.browser_connect_cdp import BrowserConnectCdpTaskStep from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep from agent_env.task_step.task_steps.computer_recording import ( EndComputerRecordingTaskStep, @@ -88,6 +89,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]: return { _get_type(ReviewTaskStep): ReviewTaskStep, _get_type(BuildMcpCliTaskStep): BuildMcpCliTaskStep, + _get_type(BrowserConnectCdpTaskStep): BrowserConnectCdpTaskStep, _get_type(CollectArtifactsTaskStep): CollectArtifactsTaskStep, _get_type(StartComputerRecordingTaskStep): StartComputerRecordingTaskStep, _get_type(EndComputerRecordingTaskStep): EndComputerRecordingTaskStep, diff --git a/src/agent_env/task_step/task_steps/browser_connect_cdp.py b/src/agent_env/task_step/task_steps/browser_connect_cdp.py new file mode 100644 index 000000000..170bde50a --- /dev/null +++ b/src/agent_env/task_step/task_steps/browser_connect_cdp.py @@ -0,0 +1,171 @@ +"""Task step that hands a deployed agent a CDP address to an env's website browser, through the agent's cdp-config/v1, +and hides that browser's MCP tools from the agent. A reset ends the address, so a task connects the agent again after +reset_env, or with a retry_config that reruns this step.""" + +from __future__ import annotations + +import logging +from typing import ClassVar, Optional + +import httpx +from agentenv_protocol.client import find_extension_method, invoke_extension +from agentenv_protocol.tool_access import DISABLE, DISABLE_TOOL_EXTENSION_URI, WILDCARD +from agentenv_protocol.tool_context import DEFAULT_ROLE +from agentenv_protocol.website_browser import BROWSER + +from agent_env.a2a_agent.a2a_agent import A2AAgent +from agent_env.entity.refs import EntityRef +from agent_env.env.env import DeployedEnv, DeployedSandboxEnv +from agent_env.providers.sandbox_providers.local_docker.provider import host_url_for +from agent_env.providers.sandbox_providers.sandbox_provider import reachable_url +from agent_env.task_step.context import DeployedAgent, TaskStepContext +from agent_env.task_step.task_step import TaskStep, TaskStepDependency + +logger = logging.getLogger(__name__) + +#: Where the step records each connection it makes, one per agent and name: the role its tools were hidden from. +CONNECTIONS_KEY = "browser_connections" +#: The name the address is added to the agent's cdp-config under when the step names none. +DEFAULT_NAME = "chromium" +#: How long the browser may take to answer cdp, which waits for the browser to be ready. +CDP_TIMEOUT_S = 180 +#: How long the agent may take to answer cdp-config's add. +ADD_TIMEOUT_S = 60 + + +class BrowserConnectCdpTaskStep(TaskStep): + """Hand the agent ``agent_name`` a CDP address to the website browser ``browser`` in the env ``env_id``, the + env's only one when unnamed: added to its cdp-config/v1 as ``name``, with every tool the env serves hidden from + its role unless ``keep_tools``. The address is the browser's own for that agent, as its sandbox reaches it.""" + + type: ClassVar[str] = "browser_connect_cdp" + entity_refs = (EntityRef.env("env_id"),) + + def __init__( + self, + id: str, + version: Optional[int], + env_id: str, + agent_name: Optional[str] = None, + browser: Optional[str] = None, + name: str = DEFAULT_NAME, + keep_tools: bool = False, + depends_on: Optional[list[TaskStepDependency]] = None, + fail_task_on_error: bool = True, + ): + super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) + self.env_id = env_id + self.agent_name = agent_name or TaskStep.DEFAULT_AGENT_NAME + self.browser = browser + self.name = name + self.keep_tools = keep_tools + + def config_problems(self) -> list[str]: + problems = [] + if not isinstance(self.env_id, str) or not self.env_id: + problems.append(f"step {self.id!r}: env_id must be a non-empty string") + if not isinstance(self.name, str) or not self.name: + problems.append(f"step {self.id!r}: name must be a non-empty string") + if not isinstance(self.keep_tools, bool): + problems.append(f"step {self.id!r}: keep_tools must be true or false") + return problems + + def to_dict(self) -> dict: + base = super().to_dict() + base.update(env_id=self.env_id, agent_name=self.agent_name, browser=self.browser, name=self.name, + keep_tools=self.keep_tools) + return base + + @classmethod + def from_dict(cls, data: dict) -> BrowserConnectCdpTaskStep: + return cls( + **cls._base_from_dict(data), + env_id=data["env_id"], + agent_name=data.get("agent_name"), + browser=data.get("browser"), + name=data.get("name") or DEFAULT_NAME, + keep_tools=data.get("keep_tools", False), + ) + + async def execute(self, context: TaskStepContext) -> TaskStepContext: + deployed = _deployed_env(context, self.env_id) + agent = _deployed_agent(context, self.agent_name) + cdp_config = A2AAgent.find_extension(agent.a2a_card or {}, A2AAgent.EXT_CDP_CONFIG) + if cdp_config is None or not agent.a2a_url: + raise RuntimeError(f"agent {agent.agent_name!r} can't take a CDP address: its card has no " + f"{A2AAgent.EXT_CDP_CONFIG}") + browser, card = _browser(deployed, self.browser) + address = await invoke_extension(deployed.environment_url, card, BROWSER, {"user": agent.agent_name}, + CDP_TIMEOUT_S, method="cdp", headers=deployed.control_headers) + http_url, ws_url = _as_agent_reaches(address, deployed, agent) + async with httpx.AsyncClient(timeout=ADD_TIMEOUT_S) as client: + added = await client.post(agent.a2a_url + cdp_config["params"]["endpoint"], json={ + "name": self.name, "http_url": http_url, "ws_url": ws_url, + "metadata": {"browser": browser or deployed.environment_card.get("name")}}) + added.raise_for_status() + role = agent.role or DEFAULT_ROLE + hidden = None if self.keep_tools else WILDCARD + if hidden: + await deployed.invoke(DISABLE_TOOL_EXTENSION_URI, DISABLE, {"role": role, "tools": hidden}) + logger.info(f"agent {agent.agent_name!r} has {self.name!r}, a CDP address to {browser or 'the env'} in env " + f"{deployed.env_id!r}" + (f"; the env's tools are hidden from role {role!r}" if hidden else "")) + # The agent's cdp-config now holds this address under its name, so it replaces any connection recorded so. + records = context.metadata.setdefault(CONNECTIONS_KEY, []) + records[:] = [record for record in records + if (record["agent_name"], record["name"]) != (self.agent_name, self.name)] + [{ + "step_id": self.id, "env_id": self.env_id, "agent_name": self.agent_name, "browser": browser, + "name": self.name, "keep_tools": self.keep_tools, "role": role, "tools_hidden": hidden}] + return context + + +def _deployed_env(context: TaskStepContext, env_id: str) -> DeployedEnv: + deployed = next((d for d in context.deployed_envs if d.env_id == env_id), None) + if deployed is None: + raise RuntimeError(f"env {env_id!r} isn't in context.deployed_envs") + return deployed + + +def _deployed_agent(context: TaskStepContext, agent_name: str) -> DeployedAgent: + agent = next((a for a in context.deployed_agents if a.agent_name == agent_name), None) + if agent is None: + raise RuntimeError(f"agent {agent_name!r} isn't in context.deployed_agents") + return agent + + +def _browser(deployed: DeployedEnv, named: Optional[str]) -> tuple[Optional[str], dict]: + """The website browser to connect to, the one named else the env's only one: its name among the env's children, + or None for the env itself, and its card. A browser computer's driver takes its app's name, so a child is found + among the children, never by name as the driver's own card.""" + card = deployed.environment_card or {} + browsers = [(None, card)] if find_extension_method(card, BROWSER, "cdp") else [] + browsers += [(child.get("name"), child) for child in card.get("children_environments") or [] + if find_extension_method(child, BROWSER, "cdp")] + if named is not None: + found = next((browser for browser in browsers if browser[0] == named), None) + if found is None: + raise RuntimeError(f"env {deployed.env_id!r} has no website browser named {named!r}") + return found + if len(browsers) != 1: + raise RuntimeError(f"env {deployed.env_id!r} has {len(browsers) or 'no'} website browsers" + + (": name one as browser" if browsers else "")) + return browsers[0] + + +def _as_agent_reaches(address: dict, deployed: DeployedEnv, agent: DeployedAgent) -> tuple[str, str]: + """``address``'s URLs as the agent's sandbox reaches them: under the env's URL as the agent reaches that, as + deploy_agent gives it MCP URLs. A URL anywhere else, at a gateway's public address say, is kept.""" + base = deployed.environment_url.rstrip("/") + if isinstance(deployed, DeployedSandboxEnv): + theirs = reachable_url(base, from_sandbox_type=deployed.sandbox_type, to_sandbox_type=agent.sandbox_type) + else: + theirs = host_url_for(base, agent.sandbox_type) + theirs = theirs.rstrip("/") + return _rebased(address["http_url"], base, theirs), _rebased(address["ws_url"], _ws(base), _ws(theirs)) + + +def _rebased(url: str, base: str, theirs: str) -> str: + return theirs + url[len(base):] if url.startswith(base + "/") else url + + +def _ws(url: str) -> str: + return "ws" + url.removeprefix("http") if url.startswith("http") else url diff --git a/tst/data/a2a_agent/Dockerfile b/tst/data/a2a_agent/Dockerfile index a1777fdc0..654e05529 100644 --- a/tst/data/a2a_agent/Dockerfile +++ b/tst/data/a2a_agent/Dockerfile @@ -1,7 +1,7 @@ FROM mirror.gcr.io/library/python:3.12-slim@sha256:423ed6ab25b1921a477529254bfeeabf5855151dc2c3141699a1bfc852199fbf WORKDIR /app COPY agentenv-protocol/ ./agentenv-protocol/ -RUN pip install --no-cache-dir './agentenv-protocol[agent]' && useradd --create-home agent +RUN pip install --no-cache-dir './agentenv-protocol[agent]' 'websockets>=13' && useradd --create-home agent COPY agent.py . USER agent CMD ["python", "agent.py"] diff --git a/tst/data/a2a_agent/agent.py b/tst/data/a2a_agent/agent.py index 2a5b2b34f..612cbf816 100644 --- a/tst/data/a2a_agent/agent.py +++ b/tst/data/a2a_agent/agent.py @@ -13,21 +13,29 @@ It installs skills under ``~/skills//``, a bundle's files downloaded through their read grants, and a ``list-skills`` line in its prompt adds a ``skill : =, ...`` line for each one. A ``run `` line has it run that command in its container and add a ``ran : `` line. + +A ``cdp `` line has it open the URL in a new tab of the browser its cdp-config names so, over raw CDP, and +add a ``cdp : `` line, ``<digest>`` the first 12 hex digits of its address's sha256, so a +reader can tell one address from another without the address. With no endpoint by that name the turn fails. """ import asyncio import base64 import contextlib import hashlib +import itertools +import json import os import re import signal from pathlib import Path +from collections.abc import Mapping from typing import Any from urllib.parse import urlsplit import httpx from agentenv_protocol.a2a_agent import ( + CDP_CONFIG_V1, MCP_CONFIG_V1, SKILL_CONFIG_V1, TRAJECTORY_V1, @@ -48,10 +56,13 @@ enable, extension, ) +from websockets.asyncio.client import connect SEND_FILE = "send-file " LIST_SKILLS = "list-skills" RUN = "run " +CDP = "cdp " +CDP_LOAD_SECONDS = 30 RUN_TIMEOUT_SECONDS = 60 SKILLS_DIR = Path.home() / "skills" LIVE_STEPS = re.compile(r"live-steps (\d+)") @@ -105,6 +116,42 @@ async def _ran(command: str) -> str: return f"ran {command}: {output.decode(errors='replace').strip()}" +async def _opened(endpoint: Mapping[str, Any], name: str, url: str) -> str: + """``url`` opened in a new tab over raw CDP at the browser ``endpoint`` names, as the line that reports it.""" + digest = hashlib.sha256(endpoint["ws_url"].encode()).hexdigest()[:12] + try: + async with connect(endpoint["ws_url"], max_size=None, additional_headers=endpoint.get("headers")) as browser: + ids = itertools.count(1) + + async def call(method: str, params: dict | None = None, session: str | None = None) -> dict: + message_id = next(ids) + await browser.send(json.dumps({"id": message_id, "method": method, "params": params or {}, + **({"sessionId": session} if session else {})})) + while (answer := json.loads(await browser.recv())).get("id") != message_id: + pass + if "error" in answer: + raise RuntimeError(answer["error"].get("message")) + return answer.get("result") or {} + + target = (await call("Target.createTarget", {"url": url}))["targetId"] + session = (await call("Target.attachToTarget", {"targetId": target, "flatten": True}))["sessionId"] + deadline, loaded, title = asyncio.get_running_loop().time() + CDP_LOAD_SECONDS, False, "" + while not loaded and asyncio.get_running_loop().time() < deadline: + # A new tab is about:blank, already complete, until it commits the URL; a page still navigating may + # have no context to evaluate in yet. + with contextlib.suppress(RuntimeError): + state, title, href = (await call("Runtime.evaluate", { + "expression": "[document.readyState, document.title, location.href]", "returnByValue": True}, + session))["result"]["value"] + loaded = state == "complete" and href != "about:blank" + if not loaded: + await asyncio.sleep(0.2) + await call("Target.closeTarget", {"targetId": target}) + return f"cdp {name} {digest}: {title}" + except Exception as e: + return f"cdp {name} {digest} failed: {type(e).__name__}: {e}" + + def _skills() -> list[str]: """One line per installed skill: each file's path and the first 12 hex digits of its sha256.""" if not SKILLS_DIR.is_dir(): @@ -121,7 +168,7 @@ def _skills() -> list[str]: version="1.0.0", ), config=EchoAgentConfig, - extensions=(MCP_CONFIG_V1, enable(TRAJECTORY_V1, live=True), TRIGGERS_V1), + extensions=(MCP_CONFIG_V1, CDP_CONFIG_V1, enable(TRAJECTORY_V1, live=True), TRIGGERS_V1), ) class EchoAgent(AgentEnvAgent): async def run(self, request: TaskRequest[EchoAgentConfig]) -> TaskResult: @@ -137,6 +184,8 @@ async def run(self, request: TaskRequest[EchoAgentConfig]) -> TaskResult: reply = "\n".join([reply, *_skills()]) runs = [line.removeprefix(RUN).strip() for line in prompt.splitlines() if line.startswith(RUN)] reply = "\n".join([reply, *[await _ran(command) for command in runs]]) + opens = [line.removeprefix(CDP).split() for line in prompt.splitlines() if line.startswith(CDP)] + reply = "\n".join([reply, *[await _opened(request.cdp_endpoint(name), name, url) for name, url in opens]]) sends = [line.removeprefix(SEND_FILE).strip() for line in prompt.splitlines() if line.startswith(SEND_FILE)] files = [FilePart(uri=uri, name=uri.rsplit("/", 1)[-1], mime_type="text/plain") for uri in sends] request.trajectory.append({"type": "echo", "input": prompt, "output": reply}) diff --git a/tst/integration/cli/run_bundle_built_agents_test.py b/tst/integration/cli/run_bundle_built_agents_test.py index 7c12fd0ae..8e4b90fec 100644 --- a/tst/integration/cli/run_bundle_built_agents_test.py +++ b/tst/integration/cli/run_bundle_built_agents_test.py @@ -37,7 +37,7 @@ PREFIX = f'''{BASE} WORKDIR /app COPY agentenv-protocol/ ./agentenv-protocol/ -RUN pip install --no-cache-dir './agentenv-protocol[agent]' +RUN pip install --no-cache-dir './agentenv-protocol[agent]' 'websockets>=13' ''' diff --git a/tst/integration/cli/run_bundle_installed_agent_test.py b/tst/integration/cli/run_bundle_installed_agent_test.py index 8ae0f7d18..9bef796f4 100644 --- a/tst/integration/cli/run_bundle_installed_agent_test.py +++ b/tst/integration/cli/run_bundle_installed_agent_test.py @@ -25,7 +25,7 @@ DOCKERFILE = f'''{BASE} WORKDIR /app COPY agentenv-protocol/ ./agentenv-protocol/ -RUN pip install --no-cache-dir './agentenv-protocol[agent]' +RUN pip install --no-cache-dir './agentenv-protocol[agent]' 'websockets>=13' COPY agent.py ./ CMD ["python", "agent.py"] ''' diff --git a/tst/integration/computer/browser_computer_test.py b/tst/integration/computer/browser_computer_test.py index 7946901d5..d0604d568 100644 --- a/tst/integration/computer/browser_computer_test.py +++ b/tst/integration/computer/browser_computer_test.py @@ -7,7 +7,10 @@ saying why and the tools answering with it. Through its cdp address, Playwright and Puppeteer drive it two at once, each on its own session, by http_url and by ws_url; a wrong token is refused, and so is code in the computer without one; one client's close leaves the browser up for the other, which can't open file: URLs; the address outlives a -relaunch, and the computer's reset ends it. Needs Docker and the local registry on +relaunch, and the computer's reset ends it. browser_connect_cdp hands the echo agent its address, with the computer's +tools hidden from its role; after reset_env that address fails until the task runs browser_connect_cdp again, whose +fresh one its next turn opens the site through. Through an app server and a driver restart and a relaunch, the agent's +address keeps working, and two agents connected at once each drive the browser through their own. Needs Docker and the local registry on localhost:5000, and Node 20 or later and npm for the CDP clients.""" import asyncio @@ -19,6 +22,7 @@ import subprocess import time import urllib.parse +import uuid from pathlib import Path import httpx @@ -32,13 +36,18 @@ from agentenv_protocol.computer.conformance.client import image from agentenv_protocol.website_browser import BROWSER +from agent_env.a2a_agent import conversation_store from agent_env.artifact.artifacts.file import FileArtifact from agent_env.env import Env from agent_env.env.envs.computer.utils import inside from agent_env.store.routing import namespace_routing from agent_env.task.teardown import teardown_run from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.task_steps.browser_connect_cdp import BrowserConnectCdpTaskStep +from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep +from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep from agent_env.task_step.task_steps.reset_env import ResetEnvTaskStep +from tst.util.a2a_test_agent import put_test_agent pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] @@ -124,6 +133,8 @@ def log_message(self, *args): time.sleep(float(sys.argv[1])) """ SERVE_START_S = 3 +#: A page that needs no server, so killing the driver, whose child a served site is, leaves it to open. +DATA_PAGE = "data:text/html,<title>Home

Home

" # Every TCP port something in the computer listens on, and whether any process is an X server. LISTENING = """import glob ports = set() @@ -417,6 +428,129 @@ async def test_playwright_and_puppeteer_drive_it_through_its_cdp_address_which_a assert fresh != ada and not report.failed +@pytest.mark.asyncio +async def test_browser_connect_cdp_hands_an_agent_its_address_with_the_tools_hidden_which_a_reset_ends_till_it_reruns( + browser_computer): + with namespace_routing(): + agent = put_test_agent(f"cdp-agent-{uuid.uuid4().hex[:8]}") + deployed = await Env.get(browser_computer.id).deploy( + env_params={"app_settings": {"chromium": {"start_urls": [SITE]}}}) + context = TaskStepContext(deployed_envs=[deployed]) + serving = None + try: + serving = asyncio.ensure_future(inside.run(deployed, ["python3", "-c", SERVE_SITE, "900"], timeout_s=960)) + await asyncio.sleep(SERVE_START_S) + await DeployAgentTaskStep(id="deploy-solver", version=None, env_ids=[], a2a_agent_id=agent.id, + a2a_agent_version=agent.version, agent_name="solver", + sandbox_type="local").execute(context) + await BrowserConnectCdpTaskStep(id="connect", version=None, env_id=browser_computer.id, + agent_name="solver").execute(context) + first, tools = await _opened_over_cdp(context, "first"), await _tool_names(deployed) + serving.cancel() + await ResetEnvTaskStep(id="reset", version=None, env_id=browser_computer.id, + fail_task_on_error=True).execute(context) + serving = asyncio.ensure_future(inside.run(deployed, ["python3", "-c", SERVE_SITE, "900"], timeout_s=960)) + await asyncio.sleep(SERVE_START_S) + stale = await _opened_over_cdp(context, "stale") + await BrowserConnectCdpTaskStep(id="reconnect", version=None, env_id=browser_computer.id, + agent_name="solver").execute(context) + second, tools_after = await _opened_over_cdp(context, "second"), await _tool_names(deployed) + finally: + if serving is not None: + serving.cancel() + report = await teardown_run(context) + + (first_address, first_title), (second_address, second_title) = first, second + assert first_title == second_title == "Home" and first_address != second_address, (first, second) + assert stale[0] == first_address and stale[1].startswith("failed: "), stale + assert tools == tools_after == [] + assert context.metadata["browser_connections"] == [{ + "step_id": "reconnect", "env_id": browser_computer.id, "agent_name": "solver", "browser": "chromium", + "name": "chromium", "keep_tools": False, "role": "default", "tools_hidden": "*"}] + assert not report.failed + + +@pytest.mark.asyncio +async def test_a_cdp_agent_keeps_its_address_through_an_app_server_and_a_driver_restart_and_a_browser_relaunch( + browser_computer): + with namespace_routing(): + agent = put_test_agent(f"cdp-chaos-{uuid.uuid4().hex[:8]}") + deployed = await Env.get(browser_computer.id).deploy( + env_params={"app_settings": {"chromium": {"start_urls": [DATA_PAGE]}}}) + context = TaskStepContext(deployed_envs=[deployed]) + try: + await _connected(context, agent, browser_computer.id, "solver") + first = await _opened_over_cdp(context, "first", url=DATA_PAGE) + await _kill(deployed, "[s]erve_app") + await _ready(deployed) + after_server = await _opened_over_cdp(context, "after-server", url=DATA_PAGE) + await _kill(deployed, "[c]omputer.serve driver.py:PageDriver") + await _ready(deployed) + after_driver = await _opened_over_cdp(context, "after-driver", url=DATA_PAGE) + await protocol_v1.reset_data(f"{deployed.environment_url}/svc/chromium", headers=deployed.control_headers) + after_relaunch = await _opened_over_cdp(context, "after-relaunch", url=DATA_PAGE) + finally: + report = await teardown_run(context) + + assert [title for _, title in (first, after_server, after_driver, after_relaunch)] == ["Home"] * 4 + assert first[0] == after_server[0] == after_driver[0] == after_relaunch[0] + assert not report.failed + + +@pytest.mark.asyncio +async def test_two_cdp_agents_drive_one_browser_at_once_each_through_an_address_of_its_own(browser_computer): + with namespace_routing(): + agent = put_test_agent(f"cdp-pair-{uuid.uuid4().hex[:8]}") + deployed = await Env.get(browser_computer.id).deploy( + env_params={"app_settings": {"chromium": {"start_urls": [DATA_PAGE]}}}) + context = TaskStepContext(deployed_envs=[deployed]) + try: + for name in ("solver", "buyer"): + await _connected(context, agent, browser_computer.id, name) + turns = await asyncio.gather(*(_opened_over_cdp(context, f"{name}-turn-{index}", name, DATA_PAGE) + for name in ("solver", "buyer") for index in range(2))) + record = await _record(deployed) + finally: + report = await teardown_run(context) + + solver, buyer = turns[:2], turns[2:] + assert [title for _, title in turns] == ["Home"] * 4 + assert len({address for address, _ in solver}) == len({address for address, _ in buyer}) == 1 + assert solver[0][0] != buyer[0][0] + assert {connection["user"] for connection in record["cdp_connections"]} == {"solver", "buyer"} + assert [(c["agent_name"], c["name"]) for c in context.metadata["browser_connections"]] == [ + ("solver", "chromium"), ("buyer", "chromium")] + assert not report.failed + + +async def _connected(context: TaskStepContext, agent, env_id: str, name: str) -> None: + """The echo agent deployed as ``name`` on the local provider, with a CDP address to the browser.""" + await DeployAgentTaskStep(id=f"deploy-{name}", version=None, env_ids=[], a2a_agent_id=agent.id, + a2a_agent_version=agent.version, agent_name=name, sandbox_type="local").execute(context) + await BrowserConnectCdpTaskStep(id=f"connect-{name}", version=None, env_id=env_id, agent_name=name).execute(context) + + +async def _opened_over_cdp(context: TaskStepContext, prompt_id: str, agent_name: str = "solver", + url: str = SITE) -> tuple[str, str]: + """``agent_name``'s turn opening ``url`` over its ``chromium`` CDP address: that address's digest, and the title + it saw or, when it couldn't open it, ``failed: ``.""" + await PromptAgentTaskStep(id=prompt_id, version=None, prompt_id=prompt_id, agent_name=agent_name, + prompt=f"cdp chromium {url}", poll_interval_seconds=1).execute(context) + reply = conversation_store.get_conversation(context.metadata["a2a_conversations"][prompt_id])["messages"][1] + line = "\n".join(part["text"] for part in reply["parts"] if part["kind"] == "text").splitlines()[-1] + match = re.fullmatch(r"cdp chromium (\w+)( failed)?: (.*)", line) + assert match, line + return match.group(1), f"failed: {match.group(3)}" if match.group(2) else match.group(3) + + +async def _tool_names(deployed) -> list[str]: + """The tools the env's MCP lists to a caller with no role.""" + async with streamablehttp_client(deployed.mcp_url) as (read, write, _): + async with ClientSession(read, write) as session: + await session.initialize() + return [tool.name for tool in (await session.list_tools()).tools] + + async def _cdp(deployed, user: str) -> dict: """``user``'s CDP address, asked of the browser as agentenv asks: through the driver, with the control token.""" card = (await _get(deployed, "/.well-known/agent-env.json")).json() diff --git a/tst/unit/task_step/test_browser_connect_cdp.py b/tst/unit/task_step/test_browser_connect_cdp.py new file mode 100644 index 000000000..9dbc645f0 --- /dev/null +++ b/tst/unit/task_step/test_browser_connect_cdp.py @@ -0,0 +1,144 @@ +"""browser_connect_cdp asks the env's website browser for the agent's CDP address, adds it to the agent's cdp-config as +the agent's sandbox reaches it, and hides every tool the env serves from the agent's role unless kept; it refuses an +agent that takes no CDP address and an env whose browser it can't tell; and connecting under a name already connected +replaces its record. Only the HTTP boundary is mocked.""" + +from __future__ import annotations + +import json + +import httpx +import pytest + +from agent_env.env.env import DeployedSandboxEnv +from agent_env.task_step.context import DeployedAgent, TaskStepContext +from agent_env.task_step.task_steps.browser_connect_cdp import BrowserConnectCdpTaskStep + +_ENV = "http://localhost:41000" +_AGENT = "http://localhost:42000" +_BROWSER = {"uri": "urn:agentenv:browser/v1", "params": { + "endpoint": "/svc/chromium/agentenv/ext/browser", "methods": {"cdp": {"endpoint": "/svc/chromium/agentenv/ext/browser/cdp"}}}} +_TOOL_ACCESS = {"uri": "urn:agentenv:disable-tool/v1", "params": {"endpoint": "/tools/disable", + "methods": {"disable": {"method": "POST"}}}} +_CDP_CONFIG = {"uri": "urn:agentenv:cdp-config/v1", "params": {"endpoint": "/ext/cdp-config"}} + + +@pytest.mark.asyncio +async def test_the_agent_gets_the_browsers_address_as_its_sandbox_reaches_it_and_its_role_none_of_the_envs_tools( + monkeypatch): + sent = _mock(monkeypatch, tokens=["t1"]) + context = TaskStepContext(deployed_envs=[_env()], deployed_agents=[_agent()], metadata={}) + + await BrowserConnectCdpTaskStep(id="connect", version=None, env_id="computer", agent_name="solver").execute( + context) + + asked, added, hidden = sent + assert (str(asked.url), json.loads(asked.content)) == (f"{_ENV}/svc/chromium/agentenv/ext/browser/cdp", + {"user": "solver"}) + assert asked.headers["authorization"] == "Bearer control" + assert (str(added.url), json.loads(added.content)) == (f"{_AGENT}/ext/cdp-config", { + "name": "chromium", "http_url": "http://host.docker.internal:41000/svc/chromium/cdp/t1/", + "ws_url": "ws://host.docker.internal:41000/svc/chromium/cdp/t1/devtools/browser", + "metadata": {"browser": "chromium"}}) + assert (str(hidden.url), json.loads(hidden.content)) == (f"{_ENV}/tools/disable", {"role": "default", "tools": "*"}) + assert context.metadata["browser_connections"] == [{ + "step_id": "connect", "env_id": "computer", "agent_name": "solver", "browser": "chromium", "name": "chromium", + "keep_tools": False, "role": "default", "tools_hidden": "*"}] + + +@pytest.mark.asyncio +async def test_a_role_kept_tools_a_name_and_a_gateways_address_are_the_tasks(monkeypatch): + sent = _mock(monkeypatch, tokens=["t1"], base="https://gateway.test/browser/chromium") + context = TaskStepContext(deployed_envs=[_env()], deployed_agents=[_agent(role="buyer")], metadata={}) + + await BrowserConnectCdpTaskStep(id="connect", version=None, env_id="computer", agent_name="solver", + browser="chromium", name="shop", keep_tools=True).execute(context) + + _, added = sent + assert json.loads(added.content)["name"] == "shop" + assert json.loads(added.content)["http_url"] == "https://gateway.test/browser/chromium/cdp/t1/" + assert context.metadata["browser_connections"][0] == { + "step_id": "connect", "env_id": "computer", "agent_name": "solver", "browser": "chromium", "name": "shop", + "keep_tools": True, "role": "buyer", "tools_hidden": None} + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("env", "agent", "browser", "refusal"), [ + ("one", {}, None, "its card has no urn:agentenv:cdp-config/v1"), + ("none", _CDP_CONFIG, None, "has no website browsers"), + ("two", _CDP_CONFIG, None, "has 2 website browsers: name one as browser"), + ("one", _CDP_CONFIG, "firefox", "has no website browser named 'firefox'"), +]) +async def test_an_agent_without_cdp_config_or_a_browser_it_cant_tell_is_refused_before_any_request( + monkeypatch, env, agent, browser, refusal): + sent = _mock(monkeypatch, tokens=["t1"]) + children = {"one": [_app("chromium")], "none": [], "two": [_app("chromium"), _app("brave")]}[env] + context = TaskStepContext(deployed_envs=[_env(children)], deployed_agents=[_agent(extensions=[agent] if agent + else [])], metadata={}) + + with pytest.raises(RuntimeError, match=refusal): + await BrowserConnectCdpTaskStep(id="connect", version=None, env_id="computer", agent_name="solver", + browser=browser).execute(context) + assert sent == [] + + +@pytest.mark.asyncio +async def test_a_connection_under_a_name_already_connected_replaces_its_record(monkeypatch): + sent = _mock(monkeypatch, tokens=["t1", "t2"]) + other = _env() + other.env_id = "other" + context = TaskStepContext(deployed_envs=[_env(), other], deployed_agents=[_agent()], metadata={}) + + for env_id in ("computer", "other"): + await BrowserConnectCdpTaskStep(id=f"connect-{env_id}", version=None, env_id=env_id, + agent_name="solver").execute(context) + + assert len(sent) == 6 + assert [(record["env_id"], record["name"]) for record in context.metadata["browser_connections"]] == [ + ("other", "chromium")] + + +def test_it_round_trips_and_a_blank_name_is_a_config_problem(): + step = BrowserConnectCdpTaskStep(id="connect", version=None, env_id="computer", agent_name="solver", + browser="chromium", name="shop", keep_tools=True) + + assert BrowserConnectCdpTaskStep.from_dict(step.to_dict()).to_dict() == step.to_dict() + assert BrowserConnectCdpTaskStep(id="c", version=None, env_id="computer", name="").config_problems() == [ + "step 'c': name must be a non-empty string"] + + +def _app(name: str) -> dict: + return {"name": name, "capabilities": {"extensions": [_BROWSER]}} + + +def _env(children: list | None = None) -> DeployedSandboxEnv: + """A browser computer's deployment: a driver serving a data plane and tool access on its own card, named after + its browser app among its children, as a browser computer's driver is.""" + return DeployedSandboxEnv( + env_id="computer", env_version=1, sandbox_id="sb", sandbox_type="local", control_token="control", + environment_card_url=f"{_ENV}/.well-known/agent-env.json", + environment_card={"name": "chromium", "capabilities": {"extensions": [_TOOL_ACCESS], "operations": ["reset"]}, + "children_environments": [_app("chromium")] if children is None else children}) + + +def _agent(role: str | None = None, extensions: list | None = None) -> DeployedAgent: + return DeployedAgent(agent_name="solver", api_url=_AGENT, a2a_url=_AGENT, sandbox_type="local", role=role, + a2a_card={"capabilities": {"extensions": [_CDP_CONFIG] if extensions is None else extensions}}) + + +def _mock(monkeypatch, *, tokens: list[str], base: str = f"{_ENV}/svc/chromium") -> list[httpx.Request]: + """Answer the browser's cdp with each of ``tokens`` in turn, at ``base``, and anything else as done; return the + requests.""" + sent, real, left = [], httpx.AsyncClient, iter(tokens) + + def handle(request: httpx.Request) -> httpx.Response: + sent.append(request) + if request.url.path.endswith("/browser/cdp"): + token = next(left) + ws = "wss" + base.removeprefix("https") if base.startswith("https") else "ws" + base.removeprefix("http") + return httpx.Response(200, json={"http_url": f"{base}/cdp/{token}/", + "ws_url": f"{ws}/cdp/{token}/devtools/browser"}) + return httpx.Response(200, json={"status": "added"} if request.url.path == "/ext/cdp-config" else {"ok": True}) + + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle))) + return sent