From e5082eb62bbe9d1431a4ec1800039d7f07c9f7fb Mon Sep 17 00:00:00 2001 From: William Loo Date: Tue, 3 Nov 2026 12:52:03 -0800 Subject: [PATCH 1/2] feat(website): sync the clock of, snapshot, and configure a multi env's websites as its MCP servers (FD-4683) A website env, and a multi env's websites, get the harness treatment MCP servers get: sync_env_clock syncs them (and tells them main's timezone), snapshot_env exports each through its data plane and fails on a site whose card can't be read, and an apply_server_config broadcast reaches them; a site without the clock, a data plane or the config extension is skipped. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/env/legacy_protocol.py | 53 +++++- .../task_steps/apply_server_config.py | 75 ++++---- .../task_step/task_steps/snapshot_env.py | 90 ++++++--- .../task_step/task_steps/sync_env_clock.py | 57 ++++-- tst/unit/env/test_legacy_protocol.py | 19 ++ .../env_gateway_provider_test.py | 12 ++ tst/unit/task_step/sync_env_clock_test.py | 153 ++++++++++++++- .../task_step/test_apply_server_config.py | 91 +++++++++ tst/unit/task_step/test_snapshot_env.py | 174 ++++++++++++++++++ 9 files changed, 635 insertions(+), 89 deletions(-) diff --git a/src/agent_env/env/legacy_protocol.py b/src/agent_env/env/legacy_protocol.py index 16899025b..8790157ac 100644 --- a/src/agent_env/env/legacy_protocol.py +++ b/src/agent_env/env/legacy_protocol.py @@ -9,7 +9,8 @@ import asyncio import logging -from typing import TYPE_CHECKING, Optional +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Optional import httpx from agentenv_protocol import RPC_PATH, DataPart, client as protocol_v1 @@ -33,20 +34,62 @@ def environment_base_url(gateway_url: Optional[str], environment_name: str, mcp: return f"{gateway_url}/svc/{prefix}{environment_name}" +@dataclass(frozen=True) +class BackingServer: + """A server behind an env's gateway, by its environment_name: an MCP server, at ``/svc/mcp-{name}``, or a website, + at ``/svc/{name}``.""" + + environment_name: str + mcp: bool = True + + +def backing_servers(env: Any) -> list[BackingServer]: + """The servers behind env's gateway: a MultiEnv's MCP servers then its websites, a WebsiteEnv's website, or else + the env's one MCP server; empty for an env with none.""" + from agent_env.env.envs.website import WebsiteEnv + + if isinstance(env, WebsiteEnv): + return [BackingServer(env.environment_name, mcp=False)] + servers = [BackingServer(e.environment_name) for e in getattr(env, "mcp_server_envs", None) or []] + servers += [BackingServer(e.environment_name, mcp=False) for e in getattr(env, "website_envs", None) or []] + if not servers and getattr(env, "environment_name", None): + servers = [BackingServer(env.environment_name)] + return servers + + +def shared_names(servers: list[BackingServer]) -> set[str]: + """The environment_names an MCP server and a website in ``servers`` share, which a stored env card can't tell apart.""" + return {s.environment_name for s in servers if s.mcp} & {s.environment_name for s in servers if not s.mcp} + + +def mcp_wins(servers: list[BackingServer]) -> list[BackingServer]: + """``servers`` without each website that shares its environment_name with an MCP server, as a MultiEnv routes an + artifact of that name to the MCP server.""" + shared = shared_names(servers) + return [s for s in servers if s.mcp or s.environment_name not in shared] + + async def v1_base_url(deployed: Optional[DeployedEnv], gateway_url: Optional[str], environment_name: str, mcp: bool = True) -> Optional[str]: - """A child env's v1 data-plane base from the stored env card, or None for legacy; without a stored card, the live probe decides.""" + """A child env's v1 data-plane base from the stored env card, or None for legacy; without a stored card, the live probe + decides. A website's probe is a card read that is None only on a 404 and raises on any other failure, so a website whose + card could not be read is never taken for one serving no data plane.""" if deployed is None or not deployed.environment_card: base_url = environment_base_url(gateway_url, environment_name, mcp=mcp) + if not mcp: + _, card = await child_env_card(None, gateway_url, environment_name, mcp=False) + return base_url if card is not None else None return base_url if await protocol_v1.supports_v1(base_url) else None child = deployed.get_child_env_card(environment_name) return deployed.environment_url + child.get("url", RPC_PATH).removesuffix(RPC_PATH) if child else None -async def child_env_card(deployed: Optional[DeployedEnv], gateway_url: Optional[str], environment_name: str, timeout: int = 10) -> tuple[str, Optional[dict]]: - """An MCP child env's card and the base its endpoints resolve against: from the stored env card, or without one a live read (None on 404).""" +async def child_env_card(deployed: Optional[DeployedEnv], gateway_url: Optional[str], environment_name: str, timeout: int = 10, + mcp: bool = True) -> tuple[str, Optional[dict]]: + """A child env's card and the base its endpoints resolve against: from the stored env card, or without one a live read + at the MCP server's gateway path, or with ``mcp=False`` the website's (None on 404).""" if deployed is None or not deployed.environment_card: - base_url = environment_base_url((gateway_url or "").rstrip("/"), environment_name) + base_url = environment_base_url((gateway_url or "").rstrip("/"), environment_name, mcp=mcp) try: return base_url, await protocol_v1.get_card(base_url, timeout=timeout) except httpx.HTTPStatusError as e: diff --git a/src/agent_env/task_step/task_steps/apply_server_config.py b/src/agent_env/task_step/task_steps/apply_server_config.py index 41f82ffd0..c8c8e1f08 100644 --- a/src/agent_env/task_step/task_steps/apply_server_config.py +++ b/src/agent_env/task_step/task_steps/apply_server_config.py @@ -1,15 +1,15 @@ """Task step that applies a task's behavior config to deployed servers. At setup (after the env is deployed + universe loaded, before the agent runs) -this step invokes each backing MCP server's ``set_*`` config extensions over the -gateway. Extensions are advertised on the server's ``EnvironmentCard`` (served at +this step invokes each backing MCP server's or website's ``set_*`` config extensions +over the gateway. Extensions are advertised on the server's ``EnvironmentCard`` (served at ``/.well-known/agent-env.json``) under ``capabilities.extensions[]`` and invoked at their advertised REST endpoint (e.g. ``/agentenv/ext/set_errors``). Config lives on the backing server, not on the gateway itself. Each server's card comes from the env card stored on the deployed record, whose extension endpoints the -gateway has already rewritten to its ``/svc/mcp-{service}`` proxy; a record without a -stored card reads it live there. The agent never sees the card or these endpoints; +gateway has already rewritten to its ``/svc/mcp-{service}`` (or a website's ``/svc/{service}``) +proxy; a record without a stored card reads it live there. The agent never sees the card or these endpoints; only this harness step invokes them. """ @@ -76,7 +76,7 @@ class ApplyServerConfigStep(TaskStep): entity_refs = (EntityRef.env("env_id"),) # A directive whose ``service`` is this sentinel is a broadcast: at execution - # it expands to one directive per MCP server in the env (same uri + args). + # it expands to one directive per MCP server and website in the env (same uri + args). # Lets a caller apply an env-wide config (e.g. set_acting_user across every # server) without enumerating services itself. Pairs naturally with # ``tolerate_unadvertised`` so servers that don't opt in are simply skipped. @@ -131,44 +131,51 @@ def from_dict(cls, data: dict) -> "ApplyServerConfigStep": tolerate_unadvertised=data.get("tolerate_unadvertised", False), ) - def _expand_directives(self, deployed) -> list[ConfigDirective]: + def _expand_directives(self, deployed) -> tuple[list[ConfigDirective], set[str], set[int]]: """Expand any broadcast directive (service == ``*``) into one directive - per MCP server in the env, resolved from the env definition. The env's - child service names aren't otherwise enumerable here (the deployed card - list carries no roster), so we read them from ``Env.get`` — matching how - deploy_env / load_artifact resolve the env. Non-broadcast directives - pass through untouched.""" + per MCP server and website in the env, resolved from the env definition, + and return them with the names that are websites, whose cards a live read + finds at ``/svc/{name}``, and the positions of the directives a broadcast + made for a website. Those are skipped, as under ``tolerate_unadvertised``, + when the website has no card or doesn't advertise the URI, so a website + the task didn't name never fails it; a directive naming a website, and + every MCP server, stays strict. The env's child service names aren't otherwise + enumerable here (the deployed card list carries no roster), so we read + them from ``Env.get`` — matching how deploy_env / load_artifact resolve + the env. A website sharing its name with an MCP server is left to the MCP + server, as the stored env card can't tell the two apart. Non-broadcast + directives pass through untouched.""" if not any(d.service == self._BROADCAST for d in self.directives): - return self.directives + return self.directives, set(), set() + from agent_env.env import legacy_protocol from agent_env.env.env import Env env = Env.get(deployed.env_id, deployed.env_version) - environment_names = [ - e.environment_name for e in (getattr(env, "mcp_server_envs", None) or []) - ] # A single-server env (a bare MCPServerEnv) exposes `environment_name` and no - # `mcp_server_envs`; mirror snapshot_env / exporter enumeration and fall - # back to it, else a broadcast silently matches nothing on single-server envs. - if not environment_names and getattr(env, "environment_name", None): - environment_names = [env.environment_name] + # `mcp_server_envs`; backing_servers falls back to it, as snapshot_env does, + # else a broadcast silently matches nothing on single-server envs. + servers = legacy_protocol.mcp_wins(legacy_protocol.backing_servers(env)) # A broadcast that resolves to zero services would apply nothing while the # step reports success: the exact silent misconfig this step prevents. Fail # loud (tolerate_unadvertised governs advertisement, not missing servers). - if not environment_names: + if not servers: raise ApplyServerConfigError( - f"apply_server_config: broadcast directive matched no MCP servers " - f"in env {self.env_id!r} (env exposes neither mcp_server_envs nor a " + f"apply_server_config: broadcast directive matched no MCP servers or websites " + f"in env {self.env_id!r} (env exposes neither mcp_server_envs, website_envs nor a " f"environment_name); nothing to apply, likely a task authoring error" ) expanded: list[ConfigDirective] = [] + lenient: set[int] = set() for d in self.directives: if d.service != self._BROADCAST: expanded.append(d) continue - for svc in environment_names: - expanded.append(ConfigDirective(service=svc, uri=d.uri, args=d.args)) - return expanded + for server in servers: + if not server.mcp: + lenient.add(len(expanded)) + expanded.append(ConfigDirective(service=server.environment_name, uri=d.uri, args=d.args)) + return expanded, {server.environment_name for server in servers if not server.mcp}, lenient async def execute(self, context: TaskStepContext) -> TaskStepContext: deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None) @@ -180,7 +187,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: # actually armed with — a partial apply must not look like a no-op. changes = context.metadata.setdefault("server_config_changes", []) # Directives skipped because the target server doesn't advertise the - # extension (only when tolerate_unadvertised). Kept separate from + # extension (when tolerate_unadvertised, or for a website a broadcast reached). Kept separate from # `changes` so the audit trail never shows a skip as an applied change. skipped = context.metadata.setdefault("server_config_skipped", []) # Expand broadcast (service="*") directives to one-per-env-service. @@ -188,7 +195,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: # network), so wrap it to keep the broadcast path's failure contract the # same ApplyServerConfigError as the per-directive HTTP calls below. try: - directives = self._expand_directives(deployed) + directives, websites, lenient = self._expand_directives(deployed) except ApplyServerConfigError: raise except Exception as e: @@ -204,15 +211,17 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: # card: a server predating env-card support, one whose card name differs, or one # whose card the deploy couldn't read. cards: dict[str, tuple[str, Optional[dict]]] = {} - for directive in directives: + for position, directive in enumerate(directives): + tolerate = self.tolerate_unadvertised or position in lenient try: if directive.service not in cards: cards[directive.service] = await legacy_protocol.child_env_card( - deployed, gateway_url_of(deployed), directive.service, timeout=self.timeout_seconds + deployed, gateway_url_of(deployed), directive.service, timeout=self.timeout_seconds, + mcp=directive.service not in websites, ) base_url, card = cards[directive.service] if card is None: - if not self.tolerate_unadvertised: + if not tolerate: raise ApplyServerConfigError( f"Service {directive.service!r} has no env card (env={self.env_id}), so {directive.uri!r} " f"can't be applied: it serves none, its card isn't named {directive.service!r}, or the deploy couldn't read it" @@ -220,7 +229,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: logger.warning( f"apply_server_config: service {directive.service!r} serves no " f"environment card (env={self.env_id}); skipping {directive.uri!r} " - f"(tolerate_unadvertised)" + f"({'tolerate_unadvertised' if self.tolerate_unadvertised else 'broadcast to a website'})" ) skipped.append({ "step_id": self.id, @@ -233,11 +242,11 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: continue ext = protocol_v1.find_extension(card, directive.uri) if ext is None: - if self.tolerate_unadvertised: + if tolerate: logger.warning( f"apply_server_config: {directive.uri!r} not advertised on " f"service {directive.service!r} (env={self.env_id}); skipping " - f"(tolerate_unadvertised)" + f"({'tolerate_unadvertised' if self.tolerate_unadvertised else 'broadcast to a website'})" ) skipped.append({ "step_id": self.id, diff --git a/src/agent_env/task_step/task_steps/snapshot_env.py b/src/agent_env/task_step/task_steps/snapshot_env.py index 48fc9d6dc..9415dac98 100644 --- a/src/agent_env/task_step/task_steps/snapshot_env.py +++ b/src/agent_env/task_step/task_steps/snapshot_env.py @@ -13,7 +13,7 @@ from dataclasses import dataclass from datetime import datetime, timezone from pathlib import PurePosixPath -from typing import ClassVar, Optional +from typing import TYPE_CHECKING, ClassVar, Optional from urllib.parse import urlparse import httpx @@ -37,6 +37,9 @@ from agent_env.entity.scopes import FieldScope from agent_env.task_step.task_step import TaskStep, TaskStepDependency +if TYPE_CHECKING: + from agent_env.env.legacy_protocol import BackingServer + logger = logging.getLogger(__name__) DEFAULT_EXPORT_TIMEOUT_SECONDS = 600 @@ -167,6 +170,9 @@ class SnapshotEnvTaskStep(TaskStep): Output: ``context.metadata["env_snapshotted_universes"][self.id]`` = ``{"id", "version"}``. Fails the step if any service fails to export. + Websites are snapshotted through their v1 data plane; a website serving + none is left out, and one sharing its name with an MCP server is left to + the MCP server, as a MultiEnv loads that name's artifact into it. ``include_env_changes`` also stores who changed what (urn:agentenv:changes/v1) under ``metadata["env_changes"]``. @@ -469,28 +475,36 @@ def _resolve_loaded_universe_ref(deployed) -> Optional[tuple[str, Optional[int]] return None return (su["id"], su.get("version")) if su and su.get("id") else None + @classmethod + def _enumerate_environments(cls, env) -> list[str]: + """environment names for a multi env, a website env or a single MCP server env.""" + return [server.environment_name for server in cls._enumerate_servers(env)] + @staticmethod - def _enumerate_environments(env) -> list[str]: - """environment names for a multi env or a single MCP server env.""" - if getattr(env, "mcp_server_envs", None): - return [e.environment_name for e in env.mcp_server_envs] - if getattr(env, "environment_name", None): - return [env.environment_name] - raise RuntimeError( - f"Env '{env.id}' v{env.version} has no MCP services to snapshot " - "(expected mcp_server_envs or environment_name)" - ) + def _enumerate_servers(env) -> list[BackingServer]: + """The servers to snapshot: the env's MCP servers and websites, a website sharing its name with an MCP + server left to it.""" + from agent_env.env import legacy_protocol + + servers = legacy_protocol.mcp_wins(legacy_protocol.backing_servers(env)) + if not servers: + raise RuntimeError( + f"Env '{env.id}' v{env.version} has no MCP services or websites to snapshot " + "(expected mcp_server_envs, website_envs or environment_name)" + ) + return servers @staticmethod async def _export_environment_to_file( gateway_url: Optional[str], environment_name: str, tmp_path: str, timeout_seconds: float, deployed=None, - data_get_params: Optional[dict] = None, - ) -> str | tuple[GetDataResponse, Optional[_SnapshotUpload]]: - """Export one service; return the suffix written (".json"/".zip") to - ``tmp_path``, or the object url a FilePart names when the service uploaded the - bundle itself (nothing written to ``tmp_path`` then — export_one registers it), or, for - an environment whose card declares a file tree, its data/get answer and the upload it - answered under, which export_one keeps file by file. ``data_get_params`` go to its data/get. + data_get_params: Optional[dict] = None, mcp: bool = True, + ) -> Optional[str | tuple[GetDataResponse, Optional[_SnapshotUpload]]]: + """Export one service, an MCP server or with ``mcp=False`` a website; return the + suffix written (".json"/".zip") to ``tmp_path``, or the object url a FilePart names + when the service uploaded the bundle itself (nothing written to ``tmp_path`` then — + export_one registers it), or, for an environment whose card declares a file tree, its + data/get answer and the upload it answered under, which export_one keeps file by file, + or None for a website serving no v1 data plane. ``data_get_params`` go to its data/get. v1 ``get_data``, handed a namespace grant when the service takes the data-objects form: an upload through it → its object url; ``DataPart`` → json; ``FilePart`` → @@ -499,10 +513,12 @@ async def _export_environment_to_file( guard. Errors are caught by the caller (fails just this service).""" from agent_env.env import legacy_protocol - base_url = await legacy_protocol.v1_base_url(deployed, gateway_url, environment_name, mcp=True) + base_url = await legacy_protocol.v1_base_url(deployed, gateway_url, environment_name, mcp=mcp) + if base_url is None and not mcp: + return None if base_url is not None: headers = deployed.control_headers_for(base_url) if deployed is not None else None - card_base, card = await legacy_protocol.child_env_card(deployed, gateway_url, environment_name) + card_base, card = await legacy_protocol.child_env_card(deployed, gateway_url, environment_name, mcp=mcp) if returns_file_tree(card or {}): upload = await asyncio.to_thread(_snapshot_upload, card or {}, timeout_seconds, getattr(deployed, "sandbox_type", None), FILE_TREE_LIMITS) @@ -611,7 +627,10 @@ async def snapshot_env_state( from agent_env.artifact import FileArtifact, EnvironmentArtifact, EnvironmentUniverseArtifact from agent_env.env.env import gateway_url_of - services = cls._enumerate_environments(env) if gateway_url else _card_environments(deployed) + from agent_env.env.legacy_protocol import BackingServer + + servers = cls._enumerate_servers(env) if gateway_url else [BackingServer(name) for name in _card_environments(deployed)] + services = [server.environment_name for server in servers] gateway = gateway_url.rstrip("/") if gateway_url else None # `gateway` identifies which deployment was exported: env= alone doesn't when one env_id has several. logger.info( @@ -643,9 +662,12 @@ async def snapshot_env_state( # The stored card describes the deployment's own gateway; an override pointing elsewhere keeps the live probe. card_record = deployed if deployed is not None and (gateway is None or (gateway_url_of(deployed) or "").rstrip("/") == gateway) else None sem = asyncio.Semaphore(concurrency) + no_data_plane: set[str] = set() - async def export_one(environment_name: str): - """Returns EnvironmentArtifact on success, (public_err, log_err) on failure.""" + async def export_one(server: BackingServer): + """Returns EnvironmentArtifact on success, (public_err, log_err) on failure, None for a website serving no data plane.""" + environment_name = server.environment_name + website = {} if server.mcp else {"mcp": False} async with sem: # Suffix (.json/.zip) is reported by _export_environment_to_file. fd, tmp_path = tempfile.mkstemp(prefix=f"{environment_name}-") @@ -656,7 +678,7 @@ async def export_one(environment_name: str): try: result = await cls._export_environment_to_file( gateway, environment_name, tmp_path, export_timeout_seconds, - deployed=card_record, data_get_params=(data_get_params or {}).get(environment_name), + deployed=card_record, data_get_params=(data_get_params or {}).get(environment_name), **website, ) except httpx.HTTPStatusError as e: body_snippet = "" @@ -671,6 +693,9 @@ async def export_one(environment_name: str): return public, f"{public}: {e}" except ValueError as e: return str(e), str(e) + if result is None: + no_data_plane.add(environment_name) + return None if isinstance(result, tuple): try: @@ -744,7 +769,7 @@ def _register_uploaded_bundle(): except OSError: pass - results = await asyncio.gather(*(export_one(name) for name in services)) + results = await asyncio.gather(*(export_one(server) for server in servers)) environment_artifacts = [] skipped: dict[str, tuple[str, str]] = {} @@ -767,9 +792,18 @@ def _register_uploaded_bundle(): f"{len(skipped)}/{len(services)} services were not exported " f"(missing={sorted(skipped)}): {public_summary}" ) - if empty := [name for name, result in zip(services, results) if result is None]: - if len(empty) == len(services): - raise RuntimeError(f"{', '.join(map(repr, empty))} returned no files to snapshot") + without_data_plane = [name for name in services if name in no_data_plane] + if without_data_plane: + logger.warning( + f"snapshot_env: websites {without_data_plane} of {env.id} serve no v1 data plane; " + "their state is not in the snapshot" + ) + empty = [name for name, result in zip(services, results) if result is None and name not in no_data_plane] + if not environment_artifacts: + reasons = [f"its websites {without_data_plane} serve no v1 data plane"] if without_data_plane else [] + reasons += [f"{', '.join(map(repr, empty))} returned no files to snapshot"] if empty else [] + raise RuntimeError(f"Env '{env.id}' v{env.version} exported no state: {'; '.join(reasons)}") + if empty: logger.info(f"snapshot_env: {empty} returned no files to keep, so {snapshot_id} leaves them out") universe = await asyncio.to_thread( diff --git a/src/agent_env/task_step/task_steps/sync_env_clock.py b/src/agent_env/task_step/task_steps/sync_env_clock.py index 76ff196e8..c80cef8a8 100644 --- a/src/agent_env/task_step/task_steps/sync_env_clock.py +++ b/src/agent_env/task_step/task_steps/sync_env_clock.py @@ -1,16 +1,16 @@ """Arm a deployed env gateway's virtual clock, then sync its backing servers to it. Arms the clock (set_time) and reads the gateway's server-reachable env_get_time_url (state), both as the -stored env card declares clock/v1, then invokes sync_time on each backing server whose child env card -advertises it. Non-advertisers are skipped when tolerate_missing_sync_time (default True). Place right -before prompt_agent so the virtual timeline starts at agent-start. +stored env card declares clock/v1, then invokes sync_time on each backing server, MCP server or website, +whose child env card advertises it. Non-advertisers are skipped when tolerate_missing_sync_time (default +True), and websites always. Place right before prompt_agent so the virtual timeline starts at agent-start. """ from __future__ import annotations import logging from datetime import datetime, timezone as _utc_tz -from typing import ClassVar, Optional +from typing import TYPE_CHECKING, ClassVar, Optional from zoneinfo import ZoneInfo, ZoneInfoNotFoundError import httpx @@ -22,6 +22,9 @@ from agent_env.entity.refs import EntityRef from agent_env.task_step.task_step import TaskStep, TaskStepDependency +if TYPE_CHECKING: + from agent_env.env.legacy_protocol import BackingServer + logger = logging.getLogger(__name__) @@ -107,16 +110,15 @@ def from_dict(cls, data: dict) -> SyncEnvClockTaskStep: timezone=data.get("timezone"), ) - def _service_names(self, deployed) -> list[str]: - """Backing MCP server names for the env (mirrors apply_server_config's enumeration).""" + def _backing_servers(self, deployed) -> list[BackingServer]: + """The env's backing MCP servers and websites (``legacy_protocol.backing_servers``). A website sharing its + name with an MCP server has its card read live, since the stored env card can't tell the two apart.""" + from agent_env.env import legacy_protocol from agent_env.env.env import Env - env = Env.get(deployed.env_id, deployed.env_version) - names = [e.environment_name for e in (getattr(env, "mcp_server_envs", None) or [])] - if not names and getattr(env, "environment_name", None): - names = [env.environment_name] - return names + return legacy_protocol.backing_servers(Env.get(deployed.env_id, deployed.env_version)) async def execute(self, context: TaskStepContext) -> TaskStepContext: + from agent_env.env import legacy_protocol from agent_env.env.env import gateway_url_of deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None) @@ -131,10 +133,17 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: state = await deployed.invoke(EXT_CLOCK_URI, "state", timeout=self.timeout_seconds) env_get_time_url = state.get("env_get_time_url") + servers = self._backing_servers(deployed) + shared = legacy_protocol.shared_names(servers) synced, skipped = [], [] - for service in self._service_names(deployed): - outcome = await self._sync_one(deployed, gateway_url_of(deployed), service, env_get_time_url) - (synced if outcome.get("synced") else skipped).append({"service": service, "environment": service, **outcome}) + for server in servers: + service = server.environment_name + record = None if service in shared and not server.mcp else deployed + outcome = await self._sync_one(record, gateway_url_of(deployed), service, env_get_time_url, mcp=server.mcp, + tolerate=self.tolerate_missing_sync_time or not server.mcp, + headers=deployed.control_headers) + entry = {"service": service, "environment": service, **({} if server.mcp else {"website": True}), **outcome} + (synced if outcome.get("synced") else skipped).append(entry) configuration = { "step_id": self.id, "env_id": self.env_id, @@ -151,17 +160,24 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: f"({[(s['service'], s.get('reason')) for s in skipped]}) — every tool still answers wall time") return context - async def _sync_one(self, deployed, gateway_url: str, service: str, env_get_time_url: Optional[str]) -> dict: - """Invoke sync_time on one server if its card advertises clock/v1; skip (or raise) per the flag.""" + async def _sync_one(self, deployed, gateway_url: str, service: str, env_get_time_url: Optional[str], mcp: bool = True, + tolerate: Optional[bool] = None, headers: Optional[dict] = None) -> dict: + """Invoke sync_time on one server, an MCP server or with ``mcp=False`` a website, if its card advertises + clock/v1; skip a server with no card or no clock/v1 when ``tolerate`` (tolerate_missing_sync_time by default), + else raise. ``execute`` tolerates every website, so one that doesn't follow the clock never fails the step; one + that follows it but can't take the task's timezone does, as an MCP server does. ``headers`` are the env's control + headers, ``deployed``'s by default; ``execute`` passes them because it hands a website that shares its name with + an MCP server no record, so its card is read live.""" + tolerate = self.tolerate_missing_sync_time if tolerate is None else tolerate from agent_env.env import legacy_protocol - base_url, card = await legacy_protocol.child_env_card(deployed, gateway_url, service, timeout=self.timeout_seconds) + base_url, card = await legacy_protocol.child_env_card(deployed, gateway_url, service, timeout=self.timeout_seconds, mcp=mcp) if card is None: - if self.tolerate_missing_sync_time: + if tolerate: return {"synced": False, "reason": "no_env_card"} raise RuntimeError(f"{service} has no env card: it serves none, its card isn't named {service!r}, or the deploy couldn't read it") if protocol_v1.find_extension(card, EXT_CLOCK_URI) is None: - if self.tolerate_missing_sync_time: + if tolerate: logger.info(f"sync_env_clock: {service} does not advertise {EXT_CLOCK_URI}; skipping") return {"synced": False, "reason": "not_advertised"} raise RuntimeError(f"{service} does not advertise {EXT_CLOCK_URI}") @@ -176,5 +192,6 @@ async def _sync_one(self, deployed, gateway_url: str, service: str, env_get_time params["timezone"] = self.timezone result = await protocol_v1.invoke_extension(base_url, card, EXT_CLOCK_URI, params=params, - timeout=self.timeout_seconds, headers=deployed.control_headers) + timeout=self.timeout_seconds, + headers=deployed.control_headers if headers is None else headers) return {"synced": True, "result": result} diff --git a/tst/unit/env/test_legacy_protocol.py b/tst/unit/env/test_legacy_protocol.py index 9508e8d2e..7d107ad34 100644 --- a/tst/unit/env/test_legacy_protocol.py +++ b/tst/unit/env/test_legacy_protocol.py @@ -177,3 +177,22 @@ async def test_a_v1_services_state_is_the_data_it_answers_with_else_its_export_s assert await legacy_protocol.service_state(None, "http://gw", "slack") == state assert ("GET /svc/mcp-slack/export-state" in asked) is reads_export_state + + +def test_backing_servers_are_an_envs_mcp_servers_then_its_websites(): + from types import SimpleNamespace + + from agent_env.env.envs.website import WebsiteEnv + from agent_env.env.legacy_protocol import BackingServer + + named = lambda *names: [SimpleNamespace(environment_name=n) for n in names] + multi = SimpleNamespace(mcp_server_envs=named("slack", "email"), website_envs=named("shop", "slack")) + + assert legacy_protocol.backing_servers(multi) == [BackingServer("slack"), BackingServer("email"), + BackingServer("shop", mcp=False), BackingServer("slack", mcp=False)] + assert legacy_protocol.shared_names(legacy_protocol.backing_servers(multi)) == {"slack"} + assert legacy_protocol.mcp_wins(legacy_protocol.backing_servers(multi)) == [ + BackingServer("slack"), BackingServer("email"), BackingServer("shop", mcp=False)] + assert legacy_protocol.backing_servers(WebsiteEnv(id="w", version=1, environment_name="shop")) == [BackingServer("shop", mcp=False)] + assert legacy_protocol.backing_servers(SimpleNamespace(environment_name="notes")) == [BackingServer("notes")] + assert legacy_protocol.backing_servers(SimpleNamespace()) == [] diff --git a/tst/unit/providers/env_providers/env_gateway_provider_test.py b/tst/unit/providers/env_providers/env_gateway_provider_test.py index a2f9958bc..f604e42cf 100644 --- a/tst/unit/providers/env_providers/env_gateway_provider_test.py +++ b/tst/unit/providers/env_providers/env_gateway_provider_test.py @@ -2466,6 +2466,18 @@ def test_a_one_image_website_env_is_fronted_with_only_its_server_image(): assert topology.website_images == [server] +def test_a_websites_server_reaches_the_gateways_clock_at_the_url_the_gateway_hands_it(): + """sync_env_clock hands each server http://gateway:/clock/time: the gateway's compose service is that host, + listening on that port, on the one network the website's server is on.""" + from agent_env.env.gateway.constants import GATEWAY_COMPOSE_HOST + + services = yaml.safe_load(_render([], [WebsiteConfig(backend_image="shop-be", frontend_image=None, environment_name="shop")]))["services"] + gateway, site = services[GATEWAY_COMPOSE_HOST], services["shop-website-backend"] + + assert "MCP_PORT=18765" in gateway["environment"] + assert gateway["networks"] == site["networks"] == ["env-network"] + + def _sites(handler): """Route the deploy's site checks to ``handler(request) -> httpx.Response``.""" real = httpx.AsyncClient diff --git a/tst/unit/task_step/sync_env_clock_test.py b/tst/unit/task_step/sync_env_clock_test.py index 5ab7bee8e..5b83c51d7 100644 --- a/tst/unit/task_step/sync_env_clock_test.py +++ b/tst/unit/task_step/sync_env_clock_test.py @@ -5,6 +5,7 @@ from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, EnvCapabilityUnsupported from agent_env.env.gateway.constants import EXT_CLOCK_URI +from agent_env.env.legacy_protocol import BackingServer from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_steps import sync_env_clock as mod from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep, resolve_virtual_time @@ -94,7 +95,7 @@ async def test_sync_one_skips_or_raises_for_a_child_env_missing_from_the_stored_ @pytest.mark.asyncio async def test_execute_syncs_each_child_env_through_the_stored_card(monkeypatch): sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) - monkeypatch.setattr(SyncEnvClockTaskStep, "_service_names", lambda self, deployed: ["email"]) + monkeypatch.setattr(SyncEnvClockTaskStep, "_backing_servers", lambda self, deployed: [BackingServer("email")]) context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]}))]) await _step().execute(context) @@ -124,6 +125,96 @@ async def test_execute_keeps_the_arm_failure_message(monkeypatch): await _step().execute(context) +@pytest.mark.asyncio +async def test_execute_syncs_a_multi_envs_websites_alongside_its_mcp_servers(monkeypatch): + from types import SimpleNamespace + from unittest.mock import patch + + sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name="email")], + website_envs=[SimpleNamespace(environment_name="shop"), SimpleNamespace(environment_name="blog")]) + context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT], "shop": [_CLOCK_EXT], "blog": []}))]) + + with patch("agent_env.env.env.Env.get", return_value=env): + await _step().execute(context) + + assert [(r.method, str(r.url)) for r in sent][2:] == [ + ("POST", f"{_CARD}/svc/mcp-email/agentenv/ext/sync_time"), ("POST", f"{_CARD}/svc/shop/agentenv/ext/sync_time"), + ] + record = context.metadata["clock_configurations"][0] + assert [(s["service"], s.get("website")) for s in record["synced"]] == [("email", None), ("shop", True)] + assert [(s["service"], s["website"], s["reason"]) for s in record["skipped"]] == [("blog", True, "not_advertised")] + + +@pytest.mark.asyncio +async def test_execute_syncs_a_single_website_env(monkeypatch): + from unittest.mock import patch + + from agent_env.env.envs.website import WebsiteEnv + + sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"shop": [_CLOCK_EXT]}))]) + + with patch("agent_env.env.env.Env.get", return_value=WebsiteEnv(id="e1", version=1, environment_name="shop")): + await _step().execute(context) + + assert [(r.method, str(r.url)) for r in sent][2:] == [("POST", f"{_CARD}/svc/shop/agentenv/ext/sync_time")] + assert json.loads(sent[2].content) == {"env_get_time_url": CLOCK_URL} + + +@pytest.mark.asyncio +async def test_sync_one_reads_a_websites_card_live_at_its_own_gateway_path(monkeypatch): + read = [] + + async def fake_card(base_url, **kw): + read.append(base_url) + return {"capabilities": {"extensions": []}} + + monkeypatch.setattr(mod.protocol_v1, "get_card", fake_card) + assert await _step()._sync_one(None, "http://gw", "shop", CLOCK_URL, mcp=False) == {"synced": False, "reason": "not_advertised"} + assert read == ["http://gw/svc/shop"] + + +@pytest.mark.asyncio +async def test_execute_reads_a_website_sharing_an_mcp_servers_name_live(monkeypatch): + from types import SimpleNamespace + from unittest.mock import patch + + sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name="email")], + website_envs=[SimpleNamespace(environment_name="email")]) + context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]}))]) + + with patch("agent_env.env.env.Env.get", return_value=env): + await _step().execute(context) + + assert [(r.method, str(r.url)) for r in sent][2:] == [ + ("POST", f"{_CARD}/svc/mcp-email/agentenv/ext/sync_time"), ("GET", "http://gw/svc/email/.well-known/agent-env.json"), + ] + + +@pytest.mark.asyncio +async def test_strict_execute_skips_websites_but_not_mcp_servers_that_do_not_follow_the_clock(monkeypatch): + from types import SimpleNamespace + from unittest.mock import patch + + _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name="email")], + website_envs=[SimpleNamespace(environment_name="shop"), SimpleNamespace(environment_name="legacy")]) + context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT], "shop": []}))]) + + with patch("agent_env.env.env.Env.get", return_value=env): + await _step(tolerate_missing_sync_time=False).execute(context) + + record = context.metadata["clock_configurations"][0] + assert [s["service"] for s in record["synced"]] == ["email"] + assert [(s["service"], s["reason"]) for s in record["skipped"]] == [("shop", "not_advertised"), ("legacy", "no_env_card")] + + unsynced = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [], "shop": [_CLOCK_EXT]}))]) + with patch("agent_env.env.env.Env.get", return_value=env), pytest.raises(RuntimeError, match="email does not advertise"): + await _step(tolerate_missing_sync_time=False).execute(unsynced) + + _CLOCK_EXT = {"uri": EXT_CLOCK_URI, "params": {"endpoint": "/agentenv/ext/sync_time", "methods": {"sync_time": {"method": "POST"}}}} _ZONE_EXT = {"uri": EXT_CLOCK_URI, "params": {"endpoint": "/agentenv/ext/sync_time", "methods": {"sync_time": {"method": "POST", "request": { "type": "object", "properties": {"env_get_time_url": {"type": "string"}, "timezone": {"type": "string"}}, "required": ["env_get_time_url"]}}}}} @@ -226,7 +317,7 @@ def test_without_a_zone_virtual_time_is_passed_through_untouched(): @pytest.mark.asyncio async def test_execute_arms_the_gateway_at_the_local_instant_and_tells_every_server_the_zone(monkeypatch): sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) - monkeypatch.setattr(SyncEnvClockTaskStep, "_service_names", lambda self, deployed: ["email"]) + monkeypatch.setattr(SyncEnvClockTaskStep, "_backing_servers", lambda self, deployed: [BackingServer("email")]) context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_ZONE_EXT]}))]) await _zoned().execute(context) @@ -241,7 +332,7 @@ async def test_execute_arms_the_gateway_at_the_local_instant_and_tells_every_ser @pytest.mark.parametrize("tolerate", [True, False]) async def test_a_server_that_cannot_take_the_zone_fails_the_step_before_any_sync_is_sent(monkeypatch, tolerate): sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) - monkeypatch.setattr(SyncEnvClockTaskStep, "_service_names", lambda self, deployed: ["email"]) + monkeypatch.setattr(SyncEnvClockTaskStep, "_backing_servers", lambda self, deployed: [BackingServer("email")]) context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_OLD_EXT]}))]) with pytest.raises(RuntimeError, match="takes no timezone"): @@ -264,6 +355,62 @@ async def test_a_server_that_does_not_advertise_clock_is_still_skipped_when_a_zo assert out == {"synced": False, "reason": "not_advertised"} +@pytest.mark.asyncio +async def test_a_zoned_step_tells_a_website_that_takes_the_zone_and_fails_on_one_that_cannot(monkeypatch): + """A website is tolerated for not following the clock, but one that follows it without taking the zone fails the + step, as an MCP server does.""" + from types import SimpleNamespace + from unittest.mock import patch + + env = SimpleNamespace(mcp_server_envs=[], website_envs=[SimpleNamespace(environment_name="shop")]) + sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"shop": [_ZONE_EXT]}))]) + with patch("agent_env.env.env.Env.get", return_value=env): + await _zoned().execute(context) + assert json.loads(sent[2].content) == {"env_get_time_url": CLOCK_URL, "timezone": "America/New_York"} + assert context.metadata["clock_configurations"][0]["synced"][0]["website"] is True + + _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + older = TaskStepContext(deployed_envs=[_carded(await _composed_card({"shop": [_OLD_EXT]}))]) + with patch("agent_env.env.env.Env.get", return_value=env), pytest.raises(RuntimeError, match="shop advertises .* takes no timezone"): + await _zoned().execute(older) + + +@pytest.mark.asyncio +async def test_a_website_sharing_an_mcp_servers_name_is_synced_with_the_envs_control_headers(monkeypatch): + """Its card is read live, with no record, and its sync_time still carries the env's control token.""" + import dataclasses + from types import SimpleNamespace + from unittest.mock import patch + + import httpx + + sent, real = [], httpx.AsyncClient + website_card = {"name": "email", "url": "/agentenv", "capabilities": {"extensions": [_CLOCK_EXT]}} + + def handle(request: httpx.Request) -> httpx.Response: + sent.append(request) + if request.url.path.endswith("/clock/state"): + return httpx.Response(200, json={"env_get_time_url": CLOCK_URL}) + if request.url.path == "/svc/email/.well-known/agent-env.json": + return httpx.Response(200, json=website_card) + return httpx.Response(200, json={}) + + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle))) + env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name="email")], + website_envs=[SimpleNamespace(environment_name="email")]) + record = dataclasses.replace(_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]})), control_token="ctl") + context = TaskStepContext(deployed_envs=[record]) + + with patch("agent_env.env.env.Env.get", return_value=env): + await _step().execute(context) + + website_sync = [r for r in sent if str(r.url) == "http://gw/svc/email/agentenv/ext/sync_time"] + assert len(website_sync) == 1 + assert {k: website_sync[0].headers.get(k) for k in record.control_headers} == record.control_headers + assert [s["service"] for s in context.metadata["clock_configurations"][0]["synced"]] == ["email", "email"] + + def test_a_padded_zone_is_stored_and_sent_stripped(): s = _zoned(timezone=" America/New_York ") assert s.timezone == "America/New_York" and s.to_dict()["timezone"] == "America/New_York" diff --git a/tst/unit/task_step/test_apply_server_config.py b/tst/unit/task_step/test_apply_server_config.py index f9badb26c..ddeb96806 100644 --- a/tst/unit/task_step/test_apply_server_config.py +++ b/tst/unit/task_step/test_apply_server_config.py @@ -589,6 +589,97 @@ async def test_live_read_strips_a_trailing_slash_from_the_gateway_url(): assert [url for url, _ in gw.post_calls] == [f"{_GATEWAY}/svc/mcp-slack/agentenv/ext/set_errors"] +# --- websites --- + + +def _fake_multi_env(mcp_names, website_names): + return SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name=s) for s in mcp_names], + website_envs=[SimpleNamespace(environment_name=s) for s in website_names]) + + +class _KeyedGateway(_MockGateway): + """A gateway serving each card at /svc//.well-known/agent-env.json, where a website's key is its bare name.""" + + async def get(self, url, *, timeout=None, params=None): + self.get_calls.append(url) + for key, card in self.cards.items(): + if url.endswith(f"/svc/{key}/.well-known/agent-env.json"): + return _resp(200, url, card) + return _resp(404, url, {"error": "no card"}) + + +@pytest.mark.asyncio +async def test_stored_card_broadcast_reaches_websites_and_leaves_a_shared_name_to_the_mcp_server(): + gw = _MockGateway({}) + ext = _ext(_SET_ACTING_USER_URI, "/agentenv/ext/set_acting_user") + ctx = _carded_context(await _composed_card({"mcp-slack": [ext], "shop": [ext]})) + step = _step(directives=[{"service": "*", "uri": _SET_ACTING_USER_URI, "args": {"user_email": "a@x.com"}}]) + with _patch_httpx(gw), patch("agent_env.env.env.Env.get", return_value=_fake_multi_env(["slack"], ["shop", "slack"])): + out = await step.execute(ctx) + + assert gw.get_calls == [] + assert [url for url, _ in gw.post_calls] == [f"{_CARD}/svc/mcp-slack/agentenv/ext/set_acting_user", + f"{_CARD}/svc/shop/agentenv/ext/set_acting_user"] + assert [c["service"] for c in out.metadata["server_config_changes"]] == ["slack", "shop"] + + +@pytest.mark.asyncio +async def test_live_read_broadcast_reads_a_websites_card_at_its_own_gateway_path(): + ext = _ext(_SET_ACTING_USER_URI, "/agentenv/ext/set_acting_user") + gw = _KeyedGateway({"mcp-slack": _card(ext), "shop": _card(ext)}) + step = _step(directives=[{"service": "*", "uri": _SET_ACTING_USER_URI, "args": {"user_email": "a@x.com"}}]) + with _patch_httpx(gw), patch("agent_env.env.env.Env.get", return_value=_fake_multi_env(["slack"], ["shop"])): + await step.execute(_make_context()) + + assert gw.get_calls == [f"{_GATEWAY}/svc/mcp-slack/.well-known/agent-env.json", f"{_GATEWAY}/svc/shop/.well-known/agent-env.json"] + assert [url for url, _ in gw.post_calls] == [f"{_GATEWAY}/svc/mcp-slack/agentenv/ext/set_acting_user", + f"{_GATEWAY}/svc/shop/agentenv/ext/set_acting_user"] + + +@pytest.mark.asyncio +async def test_broadcast_reaches_a_single_website_env(): + from agent_env.env.envs.website import WebsiteEnv + + gw = _MockGateway({}) + ctx = _carded_context(await _composed_card({"shop": [_ext(_SET_ACTING_USER_URI, "/agentenv/ext/set_acting_user")]})) + step = _step(directives=[{"service": "*", "uri": _SET_ACTING_USER_URI, "args": {"user_email": "a@x.com"}}]) + with _patch_httpx(gw), patch("agent_env.env.env.Env.get", return_value=WebsiteEnv(id="env-x", version=1, environment_name="shop")): + await step.execute(ctx) + + assert [url for url, _ in gw.post_calls] == [f"{_CARD}/svc/shop/agentenv/ext/set_acting_user"] + + +@pytest.mark.asyncio +async def test_strict_broadcast_skips_websites_without_a_card_or_the_extension(): + gw = _MockGateway({}) + ext = _ext(_SET_ERRORS_URI, "/agentenv/ext/set_errors") + ctx = _carded_context(await _composed_card({"mcp-slack": [ext], "shop": []})) + step = _step(directives=[{"service": "*", "uri": _SET_ERRORS_URI, "args": {"error_rate": 1.0}}]) + with _patch_httpx(gw), patch("agent_env.env.env.Env.get", return_value=_fake_multi_env(["slack"], ["shop", "legacy"])): + out = await step.execute(ctx) + + assert [url for url, _ in gw.post_calls] == [f"{_CARD}/svc/mcp-slack/agentenv/ext/set_errors"] + assert [c["service"] for c in out.metadata["server_config_changes"]] == ["slack"] + assert [(s["service"], s["reason"]) for s in out.metadata["server_config_skipped"]] == [ + ("shop", "extension_not_advertised"), ("legacy", "no_env_card")] + + +@pytest.mark.asyncio +async def test_strict_broadcast_still_fails_for_an_mcp_server_or_a_named_website_lacking_the_extension(): + ext = _ext(_SET_ERRORS_URI, "/agentenv/ext/set_errors") + broadcast = {"service": "*", "uri": _SET_ERRORS_URI, "args": {"error_rate": 1.0}} + unadvertising_mcp = _carded_context(await _composed_card({"mcp-slack": [], "shop": [ext]})) + with _patch_httpx(_MockGateway({})), patch("agent_env.env.env.Env.get", return_value=_fake_multi_env(["slack"], ["shop"])): + with pytest.raises(ApplyServerConfigError, match="not advertised on card for service 'slack'"): + await _step(directives=[broadcast]).execute(unadvertising_mcp) + + unadvertising_shop = _carded_context(await _composed_card({"mcp-slack": [ext], "shop": []})) + named = {"service": "shop", "uri": _SET_ERRORS_URI, "args": {"error_rate": 1.0}} + with _patch_httpx(_MockGateway({})), patch("agent_env.env.env.Env.get", return_value=_fake_multi_env(["slack"], ["shop"])): + with pytest.raises(ApplyServerConfigError, match="not advertised on card for service 'shop'"): + await _step(directives=[broadcast, named]).execute(unadvertising_shop) + + # --- schema / validation (pure, no I/O) --- diff --git a/tst/unit/task_step/test_snapshot_env.py b/tst/unit/task_step/test_snapshot_env.py index 877244c37..380690fc2 100644 --- a/tst/unit/task_step/test_snapshot_env.py +++ b/tst/unit/task_step/test_snapshot_env.py @@ -1367,3 +1367,177 @@ def shared_credentials_env(self) -> dict[str, str]: store = _SharingS3Store(client=object(), bucket="bucket") monkeypatch.setattr(snapshot_mod, "get_config", lambda: type("Cfg", (), {"get_object_store": lambda self: store})()) + + +# ── websites ───────────────────────────────────────────────────────────────── + +def _multi_env(mcp_names, website_names): + return SimpleNamespace(id="env-1", version=3, mcp_server_envs=[_FakeServiceEnv(n) for n in mcp_names], + website_envs=[_FakeServiceEnv(n) for n in website_names]) + + +def test_enumerate_includes_websites_and_leaves_a_shared_name_to_the_mcp_server(): + servers = SnapshotEnvTaskStep._enumerate_servers(_multi_env(["slack", "gmail"], ["shop", "slack"])) + assert [(s.environment_name, s.mcp) for s in servers] == [("slack", True), ("gmail", True), ("shop", False)] + + +def test_enumerate_a_single_website_env_and_a_multi_env_of_websites_only(): + from agent_env.env.envs.website import WebsiteEnv + + site = WebsiteEnv(id="site-env", version=1, environment_name="shop") + assert [(s.environment_name, s.mcp) for s in SnapshotEnvTaskStep._enumerate_servers(site)] == [("shop", False)] + assert SnapshotEnvTaskStep._enumerate_environments(_multi_env([], ["shop"])) == ["shop"] + + +@pytest.mark.asyncio +async def test_export_of_a_website_reads_its_data_plane_from_the_stored_card(monkeypatch, tmp_path): + record = _carded(await _card({"mcp-slack": [], "shop": []})) + sent = _mock_http(monkeypatch, rpc_result={"parts": [{"kind": "data", "data": {"orders": 2}}]}) + + out = str(tmp_path / "export-tmp") + suffix = await SnapshotEnvTaskStep._export_environment_to_file("https://gw-1", "shop", out, 30, record, mcp=False) + + assert suffix == ".json" + assert [(r.method, str(r.url)) for r in sent] == [("POST", f"{_CARD}/svc/shop/agentenv")] + with open(out) as f: + assert json.load(f) == {"orders": 2} + + +@pytest.mark.asyncio +async def test_export_of_a_website_without_a_stored_card_probes_its_own_gateway_path(monkeypatch, tmp_path): + import dataclasses + + record = dataclasses.replace(_carded({}), environment_card=None) + sent = _mock_http(monkeypatch, rpc_result={"parts": [{"kind": "data", "data": {"orders": 2}}]}, + body=b'{"name": "shop", "url": "/agentenv"}') + + suffix = await SnapshotEnvTaskStep._export_environment_to_file("https://gw-1", "shop", str(tmp_path / "out"), 30, record, mcp=False) + + assert suffix == ".json" + assert [(r.method, str(r.url)) for r in sent] == [ + ("GET", "https://gw-1/svc/shop/.well-known/agent-env.json"), + ("GET", "https://gw-1/svc/shop/.well-known/agent-env.json"), + ("POST", "https://gw-1/svc/shop/agentenv"), + ] + + +def _serve_website_card(monkeypatch, card_response): + """A gateway whose website ``shop`` answers its card read with ``card_response`` (a response or an exception to + raise) and whose MCP server ``slack`` serves a v1 data plane; returns the requests sent.""" + import httpx + + sent, real = [], httpx.AsyncClient + + def handle(request: httpx.Request) -> httpx.Response: + sent.append(request) + if request.url.path == f"/svc/shop{WELL_KNOWN_PATH}": + if isinstance(card_response, Exception): + raise card_response + return card_response + if request.url.path.endswith("/agentenv"): + return httpx.Response(200, json={"jsonrpc": "2.0", "id": 1, "result": {"parts": [{"kind": "data", "data": {}}]}}) + return httpx.Response(200, json={}) + + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle))) + return sent + + +def _card_read_failures(): + import httpx + + return [httpx.Response(500, json={}), httpx.Response(503, json={}), httpx.ReadTimeout("card read timed out"), + httpx.ConnectError("connection refused")] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("card_response", _card_read_failures(), ids=["500", "503", "timeout", "connect-error"]) +async def test_export_of_a_website_whose_card_read_fails_raises_rather_than_skipping_it(monkeypatch, tmp_path, card_response): + """Without a stored card, a website card read that fails is no proof the website serves no data plane.""" + import httpx + + sent = _serve_website_card(monkeypatch, card_response) + + with pytest.raises(httpx.HTTPError): + await SnapshotEnvTaskStep._export_environment_to_file("https://gw-1", "shop", str(tmp_path / "out"), 30, mcp=False) + assert [str(r.url) for r in sent] == [f"https://gw-1/svc/shop{WELL_KNOWN_PATH}"] + + +@pytest.mark.asyncio +async def test_export_of_a_website_without_a_card_is_skipped(monkeypatch, tmp_path): + import httpx + + sent = _serve_website_card(monkeypatch, httpx.Response(404, json={})) + + assert await SnapshotEnvTaskStep._export_environment_to_file("https://gw-1", "shop", str(tmp_path / "out"), 30, mcp=False) is None + assert [str(r.url) for r in sent] == [f"https://gw-1/svc/shop{WELL_KNOWN_PATH}"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("card_response", _card_read_failures(), ids=["500", "503", "timeout", "connect-error"]) +async def test_core_fails_a_snapshot_whose_website_card_cannot_be_read(monkeypatch, card_response): + """Another server exporting does not let the snapshot through without a website it could not check.""" + real_export = SnapshotEnvTaskStep.__dict__["_export_environment_to_file"] + _install_core_artifact_fakes(monkeypatch, EnvironmentUniverseArtifact) + monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", real_export) + _serve_website_card(monkeypatch, card_response) + + with pytest.raises(RuntimeError, match=r"1/2 services were not exported \(missing=\['shop'\]\)"): + await SnapshotEnvTaskStep.snapshot_env_state( + env=_multi_env(["slack"], ["shop"]), gateway_url="https://gw-1", snapshot_id="snap-x", + ) + + +@pytest.mark.asyncio +async def test_export_of_a_website_serving_no_data_plane_exports_nothing_and_sends_nothing(monkeypatch, tmp_path): + record = _carded(await _card({"mcp-slack": []})) + sent = _mock_http(monkeypatch) + + assert await SnapshotEnvTaskStep._export_environment_to_file("https://gw-1", "shop", str(tmp_path / "out"), 30, record, mcp=False) is None + assert sent == [] + + +@pytest.mark.asyncio +async def test_core_snapshots_websites_through_their_data_plane_and_leaves_out_those_without_one(monkeypatch): + calls = [] + + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **website): + calls.append((name, website)) + if name == "legacy-site": + return None + with open(tmp_path, "w") as f: + json.dump({"ok": name}, f) + return ".json" + + universe_puts = {} + + class _FakeUniverse: + id = "snap-x" + version = 1 + + @staticmethod + def put(id, *, environment_artifacts, metadata=None): + universe_puts["names"] = [sa.environment_name for sa in environment_artifacts] + return _FakeUniverse() + + _install_core_artifact_fakes(monkeypatch, _FakeUniverse) + monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(fake_export)) + + result = await SnapshotEnvTaskStep.snapshot_env_state( + env=_multi_env(["slack"], ["shop", "slack", "legacy-site"]), gateway_url="https://gw", snapshot_id="snap-x", + ) + + assert sorted(calls) == [("legacy-site", {"mcp": False}), ("shop", {"mcp": False}), ("slack", {})] + assert universe_puts["names"] == ["slack", "shop"] + assert result.environments_snapshotted == ["slack", "shop"] and result.total == 3 + + +@pytest.mark.asyncio +async def test_core_raises_when_no_service_exported_state(monkeypatch): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, **website): + return None + + _install_core_artifact_fakes(monkeypatch, EnvironmentUniverseArtifact) + monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(fake_export)) + + with pytest.raises(RuntimeError, match=r"exported no state: its websites \['shop'\] serve no v1 data plane"): + await SnapshotEnvTaskStep.snapshot_env_state(env=_multi_env([], ["shop"]), gateway_url="https://gw", snapshot_id="snap-x") From b18049d5eeaed225444da4c5da146f8850b4e1ef Mon Sep 17 00:00:00 2001 From: William Loo Date: Tue, 3 Nov 2026 12:52:03 -0800 Subject: [PATCH 2/2] fix(website): sync_env_clock, snapshot_env and broadcasts reach each server the env card lists, at its own route (FD-4683) The servers behind an env's gateway come from its card's children_environments, each at its own /svc route, through new protocol helpers (ChildEnv, children, child_at, child_named, serves), not from the env's definition: a website sharing an MCP server's name is its own server, and a card without data/get in its operations serves no v1 data plane. The definition is read only for servers that serve no card, and for a record whose card can't be read, logged. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/agentenv-protocol/README.md | 10 ++ .../src/agentenv_protocol/client.py | 80 +++++++++ .../agentenv-protocol/tests/test_client.py | 52 ++++++ src/agent_env/env/legacy_protocol.py | 105 ++++++++++-- .../task_steps/apply_server_config.py | 81 +++++---- .../task_step/task_steps/snapshot_env.py | 86 +++++----- .../task_step/task_steps/sync_env_clock.py | 44 ++--- tst/unit/env/test_legacy_protocol.py | 100 ++++++++++- tst/unit/task/local_run_write_fence_test.py | 2 +- tst/unit/task_step/sync_env_clock_test.py | 80 ++++++--- .../task_step/test_apply_server_config.py | 30 +++- tst/unit/task_step/test_snapshot_env.py | 156 ++++++++++++++---- 12 files changed, 631 insertions(+), 195 deletions(-) diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index b6cc8c649..4e3b0de48 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -198,6 +198,16 @@ with any other transport are kept beside the SDK's entry. Card URLs are paths, r address the card was fetched from. On the client side, `client.mcp_path(card)` returns the declared path, or `/mcp` for a card without one. +A composed card, such as a gateway's, nests each backing environment's card under +`children_environments`, its `url` and endpoints rooted at the route the composer serves it at. +`client.children(card)` returns each as a `ChildEnv` (its `name`, `url` and `card`, with its +`extensions`, `operations` and `serves(method)`, and `base_url(address)`, its data plane resolved +against the composed card's address). Two children may share a name, as a website and an MCP +server can, so `client.child_at(card, url)` finds one by its url or route; `client.child_named(card, +name)` raises `AmbiguousChildError` for a shared name, and `client.find_child` returns the first. +`client.serves(card, method)` says whether a card serves a data-plane method by its declared +`operations`, None for a card that declares none. + Extensions are invoked from what the card advertises. `client.find_extension_method(card, uri, method)` returns one advertised method, whose `endpoint` is the method's own or else the extension's, and `client.invoke_extension(base_url, card, uri, params, method=...)` calls it with diff --git a/packages/agentenv-protocol/src/agentenv_protocol/client.py b/packages/agentenv-protocol/src/agentenv_protocol/client.py index dcddaf541..a1e352879 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/client.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/client.py @@ -7,6 +7,7 @@ import logging from collections.abc import Iterator from contextvars import ContextVar +from dataclasses import dataclass, field import httpx @@ -210,6 +211,83 @@ def find_tool(card: dict, name: str) -> dict | None: return None +def serves(card: dict, method: str) -> bool | None: + """Whether a card serves the data-plane ``method`` (``data/get`` say): True or False by its declared + ``capabilities.operations``, or None when it declares none, as an older card doesn't, so the caller decides as it + did before cards declared them (a probe, say).""" + declared = operations(card) + return None if declared is None else method in declared + + +@dataclass(frozen=True) +class ChildEnv: + """One child env a composed card nests under ``children_environments``: its ``name`` and ``url`` as its card gives + them, and that ``card``, which says what it offers. A composer roots each child's ``url`` (and its endpoints) at + the route it serves the child at, so two children sharing a name have distinct urls; ``url`` is None for a child + whose card names none.""" + + name: str | None + url: str | None + card: dict = field(hash=False, repr=False) + + @property + def extensions(self) -> list[dict]: + """The extensions the child's card advertises (``capabilities.extensions``).""" + return list((self.card.get("capabilities") or {}).get("extensions") or []) + + @property + def operations(self) -> list[str] | None: + """The data-plane methods the child's card declares (``operations``), or None when it declares none.""" + return operations(self.card) + + def find_extension(self, uri: str) -> dict | None: + """The extension the child's card advertises under ``uri``, or None (``find_extension``).""" + return find_extension(self.card, uri) + + def serves(self, method: str) -> bool | None: + """Whether the child's card serves the data-plane ``method``, None when it declares no operations (``serves``).""" + return serves(self.card, method) + + def base_url(self, address: str) -> str: + """Where the child's data plane is, its ``url`` without the RPC path, resolved against ``address``, the composed + card's: a path joins onto it, a full URL stands alone, and a child naming no url is at ``address`` itself.""" + url = (self.url or RPC_PATH).removesuffix(RPC_PATH) + return url if "://" in url else f"{address.rstrip('/')}{url}" + + +def children(card: dict) -> list[ChildEnv]: + """Each child env a composed card nests under ``children_environments``, in its order; none for a leaf card, or + one whose ``children_environments`` is missing or null. An entry that isn't an object is skipped.""" + return [ChildEnv(name=child.get("name"), url=child.get("url"), card=child) + for child in card.get("children_environments") or [] if isinstance(child, dict)] + + +def child_at(card: dict, url: str) -> ChildEnv | None: + """The child env of a composed card at ``url``: the one whose ``url`` is ``url``, or whose data plane is there + (``url`` without the RPC path, ``/svc/mcp-slack`` for ``/svc/mcp-slack/agentenv`` say), a trailing slash aside; + None when no child is. A composer serves each child at its own route, so this is the lookup two children sharing a + name can't confuse.""" + wanted = url.rstrip("/") + for child in children(card): + if child.url is not None and wanted in (child.url.rstrip("/"), child.url.removesuffix(RPC_PATH).rstrip("/")): + return child + return None + + +class AmbiguousChildError(LookupError): + """More than one child env of a composed card has the name asked for, so the name can't say which is meant.""" + + +def child_named(card: dict, name: str) -> ChildEnv | None: + """The one child env of a composed card named ``name``, or None when none is; raises ``AmbiguousChildError`` when + several are, as a website and an MCP server sharing a name are, which ``child_at`` tells apart by their urls.""" + found = [child for child in children(card) if child.name == name] + if len(found) > 1: + raise AmbiguousChildError(f"{len(found)} child envs are named {name!r} ({', '.join(str(c.url) for c in found)}); " + "look one up by its url") + return found[0] if found else None + + def find_child(card: dict, name: str) -> dict | None: """Return the nested child EnvironmentCard named `name`, or None. @@ -218,6 +296,8 @@ def find_child(card: dict, name: str) -> dict | None: rewritten to be reachable from the composed card's base_url. Callers resolve a service by name here, then use the child card with `find_extension`/`invoke_extension`. Tolerates a missing/null `children_environments` so single/leaf cards return None rather than raising. + Two children sharing `name` return the first; `child_named` refuses that, and `child_at` + finds a child by its url, which tells them apart. """ for child in card.get("children_environments") or []: if child.get("name") == name: diff --git a/packages/agentenv-protocol/tests/test_client.py b/packages/agentenv-protocol/tests/test_client.py index edcd38798..bfc65e9a7 100644 --- a/packages/agentenv-protocol/tests/test_client.py +++ b/packages/agentenv-protocol/tests/test_client.py @@ -190,6 +190,58 @@ def test_find_child_tolerates_missing_children(): assert protocol_v1.find_child({"children_environments": None}, "slack") is None +_SHARED_NAME = {"name": "gw", "children_environments": [ + {"name": "mail", "url": "/svc/mcp-mail/agentenv", "capabilities": {"operations": ["data/get"], "extensions": [ + {"uri": "urn:agentenv:clock/v1"}]}}, + {"name": "mail", "url": "/svc/mail/agentenv", "capabilities": {"operations": []}}, + {"name": "docs", "capabilities": {}}, + "not a card", +]} + + +def test_children_are_each_nested_card_by_its_name_url_and_card(): + found = protocol_v1.children(_SHARED_NAME) + + assert [(c.name, c.url) for c in found] == [("mail", "/svc/mcp-mail/agentenv"), ("mail", "/svc/mail/agentenv"), ("docs", None)] + assert found[0].card is _SHARED_NAME["children_environments"][0] + assert [c.extensions for c in found] == [[{"uri": "urn:agentenv:clock/v1"}], [], []] + assert [c.operations for c in found] == [["data/get"], [], None] + assert [c.serves("data/get") for c in found] == [True, False, None] + assert found[0].find_extension("urn:agentenv:clock/v1") is not None and found[1].find_extension("urn:agentenv:clock/v1") is None + assert protocol_v1.children({"name": "leaf"}) == [] and protocol_v1.children({"children_environments": None}) == [] + + +def test_a_childs_base_url_is_its_url_without_the_rpc_path_against_the_composed_cards_address(): + mcp, website, docs = protocol_v1.children(_SHARED_NAME) + + assert mcp.base_url("https://env.example/") == "https://env.example/svc/mcp-mail" + assert website.base_url("https://env.example") == "https://env.example/svc/mail" + assert docs.base_url("https://env.example") == "https://env.example" + assert protocol_v1.ChildEnv("x", "https://x.example/agentenv", {}).base_url("https://env.example") == "https://x.example" + + +def test_child_at_finds_a_child_by_its_url_or_data_plane_route_where_two_share_a_name(): + assert protocol_v1.child_at(_SHARED_NAME, "/svc/mail").url == "/svc/mail/agentenv" + assert protocol_v1.child_at(_SHARED_NAME, "/svc/mcp-mail/").url == "/svc/mcp-mail/agentenv" + assert protocol_v1.child_at(_SHARED_NAME, "/svc/mail/agentenv").url == "/svc/mail/agentenv" + assert protocol_v1.child_at(_SHARED_NAME, "/svc/docs") is None + assert protocol_v1.child_at({"name": "leaf"}, "/svc/mail") is None + + +def test_child_named_refuses_a_name_two_children_share(): + assert protocol_v1.child_named(_SHARED_NAME, "docs").name == "docs" + assert protocol_v1.child_named(_SHARED_NAME, "gmail") is None + with pytest.raises(protocol_v1.AmbiguousChildError, match="2 child envs are named 'mail'"): + protocol_v1.child_named(_SHARED_NAME, "mail") + assert protocol_v1.find_child(_SHARED_NAME, "mail")["url"] == "/svc/mcp-mail/agentenv" + + +def test_serves_is_what_a_card_declares_of_a_method_and_none_when_it_declares_no_operations(): + assert protocol_v1.serves({"capabilities": {"operations": ["data/get"]}}, "data/get") is True + assert protocol_v1.serves({"capabilities": {"operations": ["data/add"]}}, "data/get") is False + assert protocol_v1.serves({"capabilities": {}}, "data/get") is None + + def test_find_extension_ignores_children(): # set-errors/v1 is advertised only on the children, never at the top level assert protocol_v1.find_extension(_CARD_WITH_CHILDREN, "urn:agentenv:set-errors/v1") is None diff --git a/src/agent_env/env/legacy_protocol.py b/src/agent_env/env/legacy_protocol.py index 8790157ac..f5ca7a002 100644 --- a/src/agent_env/env/legacy_protocol.py +++ b/src/agent_env/env/legacy_protocol.py @@ -8,12 +8,15 @@ from __future__ import annotations import asyncio +import dataclasses import logging from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Optional import httpx -from agentenv_protocol import RPC_PATH, DataPart, client as protocol_v1 +from agentenv_protocol import METHOD_GET, RPC_PATH, DataPart, client as protocol_v1 + +from agent_env.env.gateway.constants import EXT_WEBSITE_URI if TYPE_CHECKING: from agent_env.env.env import DeployedEnv @@ -36,16 +39,28 @@ def environment_base_url(gateway_url: Optional[str], environment_name: str, mcp: @dataclass(frozen=True) class BackingServer: - """A server behind an env's gateway, by its environment_name: an MCP server, at ``/svc/mcp-{name}``, or a website, - at ``/svc/{name}``.""" + """A server behind an env's address, by its environment_name: an MCP server or, with ``mcp=False``, a website. + + One the env's card lists (``card_servers``) carries that ``child`` and the ``address`` its paths resolve against. + One the card lists none for, a server serving no card, carries the ``address`` alone: it has no card, and none is + read for it. One named by the env definition of a record whose card can't be read carries neither, so its card is + read live at its gateway path, ``/svc/mcp-{name}`` or a website's ``/svc/{name}`` (``server_card``).""" environment_name: str mcp: bool = True + child: Optional[protocol_v1.ChildEnv] = None + address: Optional[str] = None + + @property + def card(self) -> Optional[dict]: + """The server's card as the env's card lists it, or None.""" + return self.child.card if self.child is not None else None -def backing_servers(env: Any) -> list[BackingServer]: - """The servers behind env's gateway: a MultiEnv's MCP servers then its websites, a WebsiteEnv's website, or else - the env's one MCP server; empty for an env with none.""" +def definition_servers(env: Any) -> list[BackingServer]: + """The servers an env definition names: a MultiEnv's MCP servers then its websites, a WebsiteEnv's website, or + else the env's one MCP server; empty for an env with none. Read only for what the env's card can't say + (``backing_servers``).""" from agent_env.env.envs.website import WebsiteEnv if isinstance(env, WebsiteEnv): @@ -57,16 +72,74 @@ def backing_servers(env: Any) -> list[BackingServer]: return servers -def shared_names(servers: list[BackingServer]) -> set[str]: - """The environment_names an MCP server and a website in ``servers`` share, which a stored env card can't tell apart.""" - return {s.environment_name for s in servers if s.mcp} & {s.environment_name for s in servers if not s.mcp} - - -def mcp_wins(servers: list[BackingServer]) -> list[BackingServer]: - """``servers`` without each website that shares its environment_name with an MCP server, as a MultiEnv routes an - artifact of that name to the MCP server.""" - shared = shared_names(servers) - return [s for s in servers if s.mcp or s.environment_name not in shared] +def card_servers(card: dict, address: str) -> list[BackingServer]: + """The servers ``card``, read at ``address``, lists, each with its own card: a leaf card's env itself; else each + child the composed card nests, after the env itself when its own card serves ``data/get``, as a computer's driver + fronting its apps does. A child is a website when its card carries ``website/v1`` or its url is at a website's + gateway route, ``/svc/{name}`` (an MCP server's is ``/svc/mcp-{name}``), so a website and an MCP server sharing a + name are two servers.""" + own = protocol_v1.ChildEnv(name=card.get("name"), url=card.get("url"), card=card) + if not isinstance(card.get("children_environments"), list): + return [_listed(own, address)] + servers = [_listed(own, address)] if protocol_v1.serves(card, METHOD_GET) else [] + return servers + [_listed(child, address) for child in protocol_v1.children(card)] + + +def _listed(child: protocol_v1.ChildEnv, address: str) -> BackingServer: + route, url = f"/svc/{child.name}", (child.url or "").rstrip("/") + website = child.find_extension(EXT_WEBSITE_URI) is not None or url == route or url.startswith(f"{route}/") + return BackingServer(child.name, mcp=not website, child=child, address=address) + + +async def backing_servers(deployed: Optional[DeployedEnv], gateway_url: Optional[str], env: Any, + timeout: int = 10) -> list[BackingServer]: + """The servers behind an env as its card lists them (``card_servers``), then, behind a gateway, each server its + definition ``env`` names that the card lists none of the same name and kind for: one serving no card, which the + gateway leaves off its card, so it has none. + + The card is ``deployed``'s stored one when ``gateway_url`` is its own gateway or None, else read live at + ``gateway_url`` (an override, or a record that kept no card). A record with no card URL, an older one, or no + record, or a live read that fails, leaves the definition alone: its servers, each read live at its gateway path, + logged once.""" + from agent_env.env.env import gateway_url_of + + gateway = gateway_url.rstrip("/") if gateway_url else None + card, address, why = None, None, "it has no record" + if deployed is not None and deployed.environment_card and (gateway is None or (gateway_url_of(deployed) or "").rstrip("/") == gateway): + card, address = deployed.environment_card, deployed.environment_url + elif deployed is not None and deployed.environment_card_url and gateway: + try: + card, address = await protocol_v1.get_card(gateway, timeout=timeout), gateway + except httpx.HTTPError as e: + why = f"its card at {gateway} could not be read ({type(e).__name__})" + elif deployed is not None: + why = "its record has no card URL" + named = definition_servers(env) + if not isinstance(card, dict): + logger.info(f"env {getattr(env, 'id', None)!r}: {why}, so its servers are the ones its definition names, " + f"each read live at its gateway path: {[(s.environment_name, s.mcp) for s in named]}") + return named + servers = card_servers(card, address) + if not gateway: + return servers + listed = {(s.environment_name, s.mcp) for s in servers} + return servers + [dataclasses.replace(s, address=address) for s in named if (s.environment_name, s.mcp) not in listed] + + +async def server_card(server: BackingServer, gateway_url: Optional[str], timeout: int = 10) -> tuple[Optional[str], Optional[dict]]: + """A server's card and the base its endpoints resolve against: as the env's card lists it, None for a server the + card lists none for, else read live at its gateway path (``child_env_card``).""" + if server.address is not None: + return server.address, server.card + return await child_env_card(None, gateway_url, server.environment_name, timeout=timeout, mcp=server.mcp) + + +async def server_v1_base_url(server: BackingServer, gateway_url: Optional[str]) -> Optional[str]: + """A server's v1 data-plane base: its card's ``url`` resolved against the env's address, None for a server the + env's card lists none for, else as the live probe at its gateway path finds it (``v1_base_url``).""" + if server.address is not None: + return server.child.base_url(server.address) if server.child is not None else None + return await v1_base_url(None, gateway_url, server.environment_name, mcp=server.mcp) async def v1_base_url(deployed: Optional[DeployedEnv], gateway_url: Optional[str], environment_name: str, mcp: bool = True) -> Optional[str]: diff --git a/src/agent_env/task_step/task_steps/apply_server_config.py b/src/agent_env/task_step/task_steps/apply_server_config.py index c8c8e1f08..91cc24167 100644 --- a/src/agent_env/task_step/task_steps/apply_server_config.py +++ b/src/agent_env/task_step/task_steps/apply_server_config.py @@ -9,14 +9,16 @@ Config lives on the backing server, not on the gateway itself. Each server's card comes from the env card stored on the deployed record, whose extension endpoints the gateway has already rewritten to its ``/svc/mcp-{service}`` (or a website's ``/svc/{service}``) -proxy; a record without a stored card reads it live there. The agent never sees the card or these endpoints; -only this harness step invokes them. +proxy; a record without a stored card reads it live there. A broadcast reaches each server +the env card lists, by that server's own card and route, so a website sharing its name with +an MCP server is reached too (``legacy_protocol.backing_servers``). The agent never sees the +card or these endpoints; only this harness step invokes them. """ from __future__ import annotations import logging -from typing import Any, ClassVar, Optional +from typing import TYPE_CHECKING, Any, ClassVar, Optional import httpx from agentenv_protocol import client as protocol_v1 @@ -25,6 +27,9 @@ from agent_env.entity.refs import EntityRef from agent_env.task_step.task_step import TaskStep, TaskStepDependency +if TYPE_CHECKING: + from agent_env.env.legacy_protocol import BackingServer + logger = logging.getLogger(__name__) @@ -131,51 +136,40 @@ def from_dict(cls, data: dict) -> "ApplyServerConfigStep": tolerate_unadvertised=data.get("tolerate_unadvertised", False), ) - def _expand_directives(self, deployed) -> tuple[list[ConfigDirective], set[str], set[int]]: - """Expand any broadcast directive (service == ``*``) into one directive - per MCP server and website in the env, resolved from the env definition, - and return them with the names that are websites, whose cards a live read - finds at ``/svc/{name}``, and the positions of the directives a broadcast - made for a website. Those are skipped, as under ``tolerate_unadvertised``, - when the website has no card or doesn't advertise the URI, so a website - the task didn't name never fails it; a directive naming a website, and - every MCP server, stays strict. The env's child service names aren't otherwise - enumerable here (the deployed card list carries no roster), so we read - them from ``Env.get`` — matching how deploy_env / load_artifact resolve - the env. A website sharing its name with an MCP server is left to the MCP - server, as the stored env card can't tell the two apart. Non-broadcast - directives pass through untouched.""" + async def _expand_directives(self, deployed) -> list[tuple[ConfigDirective, Optional[BackingServer], bool]]: + """Each directive with the server it targets and whether it is lenient: a broadcast (service ``*``) expanded + into one directive per MCP server and website the env card lists (``legacy_protocol.backing_servers``), each + targeting that server by its own card and route. A directive a broadcast made for a website is lenient: it is + skipped, as under ``tolerate_unadvertised``, when the website has no card or doesn't advertise the URI, so a + website the task didn't name never fails it. A named directive, and every MCP server, stays strict; a named one + targets the server of that name, the MCP server when a website shares it, as a MultiEnv's load does, or, without + a broadcast to list them, is looked up by name (None).""" if not any(d.service == self._BROADCAST for d in self.directives): - return self.directives, set(), set() + return [(d, None, False) for d in self.directives] from agent_env.env import legacy_protocol - from agent_env.env.env import Env + from agent_env.env.env import Env, gateway_url_of - env = Env.get(deployed.env_id, deployed.env_version) - # A single-server env (a bare MCPServerEnv) exposes `environment_name` and no - # `mcp_server_envs`; backing_servers falls back to it, as snapshot_env does, - # else a broadcast silently matches nothing on single-server envs. - servers = legacy_protocol.mcp_wins(legacy_protocol.backing_servers(env)) + servers = await legacy_protocol.backing_servers(deployed, gateway_url_of(deployed), Env.get(deployed.env_id, deployed.env_version), + timeout=self.timeout_seconds) # A broadcast that resolves to zero services would apply nothing while the # step reports success: the exact silent misconfig this step prevents. Fail # loud (tolerate_unadvertised governs advertisement, not missing servers). if not servers: raise ApplyServerConfigError( f"apply_server_config: broadcast directive matched no MCP servers or websites " - f"in env {self.env_id!r} (env exposes neither mcp_server_envs, website_envs nor a " - f"environment_name); nothing to apply, likely a task authoring error" + f"in env {self.env_id!r} (its card lists no child envs, and its definition names neither " + f"mcp_server_envs, website_envs nor an environment_name); nothing to apply, likely a task authoring error" ) - expanded: list[ConfigDirective] = [] - lenient: set[int] = set() + expanded: list[tuple[ConfigDirective, Optional[BackingServer], bool]] = [] for d in self.directives: if d.service != self._BROADCAST: - expanded.append(d) + named = [s for s in servers if s.environment_name == d.service] + expanded.append((d, next((s for s in named if s.mcp), named[0] if named else None), False)) continue for server in servers: - if not server.mcp: - lenient.add(len(expanded)) - expanded.append(ConfigDirective(service=server.environment_name, uri=d.uri, args=d.args)) - return expanded, {server.environment_name for server in servers if not server.mcp}, lenient + expanded.append((ConfigDirective(service=server.environment_name, uri=d.uri, args=d.args), server, not server.mcp)) + return expanded async def execute(self, context: TaskStepContext) -> TaskStepContext: deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None) @@ -195,7 +189,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: # network), so wrap it to keep the broadcast path's failure contract the # same ApplyServerConfigError as the per-directive HTTP calls below. try: - directives, websites, lenient = self._expand_directives(deployed) + directives = await self._expand_directives(deployed) except ApplyServerConfigError: raise except Exception as e: @@ -210,16 +204,17 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: # multiple tools) against the same server. A None card marks a child env with no # card: a server predating env-card support, one whose card name differs, or one # whose card the deploy couldn't read. - cards: dict[str, tuple[str, Optional[dict]]] = {} - for position, directive in enumerate(directives): - tolerate = self.tolerate_unadvertised or position in lenient + cards: dict[Any, tuple[Optional[str], Optional[dict]]] = {} + for directive, server, lenient in directives: + tolerate = self.tolerate_unadvertised or lenient + key = directive.service if server is None else server try: - if directive.service not in cards: - cards[directive.service] = await legacy_protocol.child_env_card( - deployed, gateway_url_of(deployed), directive.service, timeout=self.timeout_seconds, - mcp=directive.service not in websites, - ) - base_url, card = cards[directive.service] + if key not in cards: + cards[key] = (await legacy_protocol.child_env_card(deployed, gateway_url_of(deployed), directive.service, + timeout=self.timeout_seconds) + if server is None else + await legacy_protocol.server_card(server, gateway_url_of(deployed), timeout=self.timeout_seconds)) + base_url, card = cards[key] if card is None: if not tolerate: raise ApplyServerConfigError( diff --git a/src/agent_env/task_step/task_steps/snapshot_env.py b/src/agent_env/task_step/task_steps/snapshot_env.py index 9415dac98..abb5573f3 100644 --- a/src/agent_env/task_step/task_steps/snapshot_env.py +++ b/src/agent_env/task_step/task_steps/snapshot_env.py @@ -475,36 +475,43 @@ def _resolve_loaded_universe_ref(deployed) -> Optional[tuple[str, Optional[int]] return None return (su["id"], su.get("version")) if su and su.get("id") else None - @classmethod - def _enumerate_environments(cls, env) -> list[str]: - """environment names for a multi env, a website env or a single MCP server env.""" - return [server.environment_name for server in cls._enumerate_servers(env)] - @staticmethod - def _enumerate_servers(env) -> list[BackingServer]: - """The servers to snapshot: the env's MCP servers and websites, a website sharing its name with an MCP - server left to it.""" - from agent_env.env import legacy_protocol - - servers = legacy_protocol.mcp_wins(legacy_protocol.backing_servers(env)) + def _snapshot_servers(servers: list[BackingServer], env) -> list[BackingServer]: + """The servers to snapshot of ``servers`` (``legacy_protocol.backing_servers``), one per environment_name: a + universe names each environment by it, and a load hands an environment of a name an MCP server and a website + share to the MCP server, so a server whose name one kept earlier holds (the env itself, else an MCP server) is + left out, and logged.""" if not servers: raise RuntimeError( f"Env '{env.id}' v{env.version} has no MCP services or websites to snapshot " - "(expected mcp_server_envs, website_envs or environment_name)" + "(its card lists no child envs, and its definition names no mcp_server_envs, website_envs or environment_name)" ) - return servers + if not all(server.environment_name for server in servers): + raise RuntimeError(f"Env '{env.id}' v{env.version} has a card naming no environment to snapshot") + kept: dict[str, BackingServer] = {} + for server in sorted(servers, key=lambda s: not s.mcp): + if (holder := kept.get(server.environment_name)) is not None: + logger.warning(f"snapshot_env: {'website' if not server.mcp else 'server'} {server.environment_name!r} " + f"of {env.id} is not snapshotted: it shares its name with the {'MCP server' if holder.mcp else 'website'} " + "kept, and a universe names an environment by its name") + continue + kept[server.environment_name] = server + return list(kept.values()) @staticmethod async def _export_environment_to_file( gateway_url: Optional[str], environment_name: str, tmp_path: str, timeout_seconds: float, deployed=None, - data_get_params: Optional[dict] = None, mcp: bool = True, + data_get_params: Optional[dict] = None, mcp: bool = True, server: Optional[BackingServer] = None, ) -> Optional[str | tuple[GetDataResponse, Optional[_SnapshotUpload]]]: """Export one service, an MCP server or with ``mcp=False`` a website; return the suffix written (".json"/".zip") to ``tmp_path``, or the object url a FilePart names when the service uploaded the bundle itself (nothing written to ``tmp_path`` then — export_one registers it), or, for an environment whose card declares a file tree, its data/get answer and the upload it answered under, which export_one keeps file by file, - or None for a website serving no v1 data plane. ``data_get_params`` go to its data/get. + or None for a service serving no v1 data plane: a website with no card, or a service whose card declares + operations without data/get. ``data_get_params`` go to its data/get. ``server`` is the service as the env's + card lists it (``legacy_protocol.backing_servers``), its card and route; without it, the service is looked up + by name on ``deployed``'s stored card, or probed live at its gateway path. v1 ``get_data``, handed a namespace grant when the service takes the data-objects form: an upload through it → its object url; ``DataPart`` → json; ``FilePart`` → @@ -513,12 +520,22 @@ async def _export_environment_to_file( guard. Errors are caught by the caller (fails just this service).""" from agent_env.env import legacy_protocol - base_url = await legacy_protocol.v1_base_url(deployed, gateway_url, environment_name, mcp=mcp) + if server is not None: + mcp = server.mcp + base_url = await legacy_protocol.server_v1_base_url(server, gateway_url) + else: + base_url = await legacy_protocol.v1_base_url(deployed, gateway_url, environment_name, mcp=mcp) + card_base, card = None, None + if base_url is not None: + card_base, card = (await legacy_protocol.server_card(server, gateway_url) if server is not None else + await legacy_protocol.child_env_card(deployed, gateway_url, environment_name, mcp=mcp)) + if card is not None and protocol_v1.serves(card, METHOD_GET) is False: + logger.info(f"snapshot_env: {environment_name}'s card declares {protocol_v1.operations(card)}, no {METHOD_GET}") + return None if base_url is None and not mcp: return None if base_url is not None: headers = deployed.control_headers_for(base_url) if deployed is not None else None - card_base, card = await legacy_protocol.child_env_card(deployed, gateway_url, environment_name, mcp=mcp) if returns_file_tree(card or {}): upload = await asyncio.to_thread(_snapshot_upload, card or {}, timeout_seconds, getattr(deployed, "sandbox_type", None), FILE_TREE_LIMITS) @@ -625,13 +642,12 @@ async def snapshot_env_state( A classmethod so callers that have no step instance can reuse it.""" from agent_env.artifact import FileArtifact, EnvironmentArtifact, EnvironmentUniverseArtifact + from agent_env.env import legacy_protocol from agent_env.env.env import gateway_url_of - from agent_env.env.legacy_protocol import BackingServer - - servers = cls._enumerate_servers(env) if gateway_url else [BackingServer(name) for name in _card_environments(deployed)] - services = [server.environment_name for server in servers] gateway = gateway_url.rstrip("/") if gateway_url else None + servers = cls._snapshot_servers(await legacy_protocol.backing_servers(deployed, gateway, env), env) + services = [server.environment_name for server in servers] # `gateway` identifies which deployment was exported: env= alone doesn't when one env_id has several. logger.info( f"snapshot_env: env={env.id} v{env.version} snapshot_id={snapshot_id} " @@ -665,7 +681,7 @@ async def snapshot_env_state( no_data_plane: set[str] = set() async def export_one(server: BackingServer): - """Returns EnvironmentArtifact on success, (public_err, log_err) on failure, None for a website serving no data plane.""" + """Returns EnvironmentArtifact on success, (public_err, log_err) on failure, None for a service serving no data plane.""" environment_name = server.environment_name website = {} if server.mcp else {"mcp": False} async with sem: @@ -678,7 +694,8 @@ async def export_one(server: BackingServer): try: result = await cls._export_environment_to_file( gateway, environment_name, tmp_path, export_timeout_seconds, - deployed=card_record, data_get_params=(data_get_params or {}).get(environment_name), **website, + deployed=card_record, data_get_params=(data_get_params or {}).get(environment_name), + server=server, **website, ) except httpx.HTTPStatusError as e: body_snippet = "" @@ -795,12 +812,12 @@ def _register_uploaded_bundle(): without_data_plane = [name for name in services if name in no_data_plane] if without_data_plane: logger.warning( - f"snapshot_env: websites {without_data_plane} of {env.id} serve no v1 data plane; " + f"snapshot_env: {without_data_plane} of {env.id} serve no v1 data plane; " "their state is not in the snapshot" ) empty = [name for name, result in zip(services, results) if result is None and name not in no_data_plane] if not environment_artifacts: - reasons = [f"its websites {without_data_plane} serve no v1 data plane"] if without_data_plane else [] + reasons = [f"{without_data_plane} serve no v1 data plane"] if without_data_plane else [] reasons += [f"{', '.join(map(repr, empty))} returned no files to snapshot"] if empty else [] raise RuntimeError(f"Env '{env.id}' v{env.version} exported no state: {'; '.join(reasons)}") if empty: @@ -883,25 +900,6 @@ def registered_upload(object_url: str, file_id: str, filename: str, content_type id=file_id, description=description, filename=filename, content_type=content_type, object_url=object_url)) -def _card_environments(deployed) -> list[str]: - """The environments a deployment without a gateway serves, as the card it recorded names them: the env itself when - it has no children; else each child its name reaches that exports, and the env too, first, when it exports beside - them, as a computer fronting its apps does.""" - card = deployed.environment_card or {} - children = card.get("children_environments") - if not isinstance(children, list): - names = [card.get("name")] - else: - names = [child.get("name") for child in children if deployed.get_child_env_card(child.get("name")) is child - and ((operations := protocol_v1.operations(child)) is None - or METHOD_GET in operations)] - if METHOD_GET in (protocol_v1.operations(card) or []): - names.insert(0, card.get("name")) - if not all(names): - raise RuntimeError(f"Deployed env '{deployed.env_id}' recorded a card naming no environment to snapshot") - return names - - def _changes_route(deployed, override_url: Optional[str]) -> tuple[str, str]: """The verb and URL of the env's changes: an override keeps `{override}/changes`, else the stored card names them.""" if override_url: diff --git a/src/agent_env/task_step/task_steps/sync_env_clock.py b/src/agent_env/task_step/task_steps/sync_env_clock.py index c80cef8a8..8ae463f56 100644 --- a/src/agent_env/task_step/task_steps/sync_env_clock.py +++ b/src/agent_env/task_step/task_steps/sync_env_clock.py @@ -1,9 +1,10 @@ """Arm a deployed env gateway's virtual clock, then sync its backing servers to it. Arms the clock (set_time) and reads the gateway's server-reachable env_get_time_url (state), both as the -stored env card declares clock/v1, then invokes sync_time on each backing server, MCP server or website, -whose child env card advertises it. Non-advertisers are skipped when tolerate_missing_sync_time (default -True), and websites always. Place right before prompt_agent so the virtual timeline starts at agent-start. +stored env card declares clock/v1, then invokes sync_time on each backing server, MCP server or website, the +env card lists (``legacy_protocol.backing_servers``) whose own card advertises it, at the route that card gives. +Non-advertisers are skipped when tolerate_missing_sync_time (default True), and websites always. Place right +before prompt_agent so the virtual timeline starts at agent-start. """ from __future__ import annotations @@ -110,15 +111,15 @@ def from_dict(cls, data: dict) -> SyncEnvClockTaskStep: timezone=data.get("timezone"), ) - def _backing_servers(self, deployed) -> list[BackingServer]: - """The env's backing MCP servers and websites (``legacy_protocol.backing_servers``). A website sharing its - name with an MCP server has its card read live, since the stored env card can't tell the two apart.""" + async def _backing_servers(self, deployed) -> list[BackingServer]: + """The env's backing MCP servers and websites, as its card lists them (``legacy_protocol.backing_servers``): + a website sharing its name with an MCP server is a server of its own, at its own route.""" from agent_env.env import legacy_protocol - from agent_env.env.env import Env - return legacy_protocol.backing_servers(Env.get(deployed.env_id, deployed.env_version)) + from agent_env.env.env import Env, gateway_url_of + return await legacy_protocol.backing_servers(deployed, gateway_url_of(deployed), Env.get(deployed.env_id, deployed.env_version), + timeout=self.timeout_seconds) async def execute(self, context: TaskStepContext) -> TaskStepContext: - from agent_env.env import legacy_protocol from agent_env.env.env import gateway_url_of deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None) @@ -133,15 +134,11 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: state = await deployed.invoke(EXT_CLOCK_URI, "state", timeout=self.timeout_seconds) env_get_time_url = state.get("env_get_time_url") - servers = self._backing_servers(deployed) - shared = legacy_protocol.shared_names(servers) synced, skipped = [], [] - for server in servers: + for server in await self._backing_servers(deployed): service = server.environment_name - record = None if service in shared and not server.mcp else deployed - outcome = await self._sync_one(record, gateway_url_of(deployed), service, env_get_time_url, mcp=server.mcp, - tolerate=self.tolerate_missing_sync_time or not server.mcp, - headers=deployed.control_headers) + outcome = await self._sync_one(deployed, gateway_url_of(deployed), service, env_get_time_url, mcp=server.mcp, + tolerate=self.tolerate_missing_sync_time or not server.mcp, server=server) entry = {"service": service, "environment": service, **({} if server.mcp else {"website": True}), **outcome} (synced if outcome.get("synced") else skipped).append(entry) @@ -161,17 +158,20 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: return context async def _sync_one(self, deployed, gateway_url: str, service: str, env_get_time_url: Optional[str], mcp: bool = True, - tolerate: Optional[bool] = None, headers: Optional[dict] = None) -> dict: + tolerate: Optional[bool] = None, server: Optional[BackingServer] = None) -> dict: """Invoke sync_time on one server, an MCP server or with ``mcp=False`` a website, if its card advertises clock/v1; skip a server with no card or no clock/v1 when ``tolerate`` (tolerate_missing_sync_time by default), else raise. ``execute`` tolerates every website, so one that doesn't follow the clock never fails the step; one - that follows it but can't take the task's timezone does, as an MCP server does. ``headers`` are the env's control - headers, ``deployed``'s by default; ``execute`` passes them because it hands a website that shares its name with - an MCP server no record, so its card is read live.""" + that follows it but can't take the task's timezone does, as an MCP server does. ``server`` is the server as the + env's card lists it (``legacy_protocol.server_card``); without it, the card is ``service``'s on ``deployed``'s + stored card, or read live at its gateway path.""" tolerate = self.tolerate_missing_sync_time if tolerate is None else tolerate from agent_env.env import legacy_protocol - base_url, card = await legacy_protocol.child_env_card(deployed, gateway_url, service, timeout=self.timeout_seconds, mcp=mcp) + if server is not None: + base_url, card = await legacy_protocol.server_card(server, gateway_url, timeout=self.timeout_seconds) + else: + base_url, card = await legacy_protocol.child_env_card(deployed, gateway_url, service, timeout=self.timeout_seconds, mcp=mcp) if card is None: if tolerate: return {"synced": False, "reason": "no_env_card"} @@ -193,5 +193,5 @@ async def _sync_one(self, deployed, gateway_url: str, service: str, env_get_time result = await protocol_v1.invoke_extension(base_url, card, EXT_CLOCK_URI, params=params, timeout=self.timeout_seconds, - headers=deployed.control_headers if headers is None else headers) + headers=deployed.control_headers) return {"synced": True, "result": result} diff --git a/tst/unit/env/test_legacy_protocol.py b/tst/unit/env/test_legacy_protocol.py index 7d107ad34..ef8fee3f8 100644 --- a/tst/unit/env/test_legacy_protocol.py +++ b/tst/unit/env/test_legacy_protocol.py @@ -179,7 +179,7 @@ async def test_a_v1_services_state_is_the_data_it_answers_with_else_its_export_s assert ("GET /svc/mcp-slack/export-state" in asked) is reads_export_state -def test_backing_servers_are_an_envs_mcp_servers_then_its_websites(): +def test_definition_servers_are_an_envs_mcp_servers_then_its_websites(): from types import SimpleNamespace from agent_env.env.envs.website import WebsiteEnv @@ -188,11 +188,93 @@ def test_backing_servers_are_an_envs_mcp_servers_then_its_websites(): named = lambda *names: [SimpleNamespace(environment_name=n) for n in names] multi = SimpleNamespace(mcp_server_envs=named("slack", "email"), website_envs=named("shop", "slack")) - assert legacy_protocol.backing_servers(multi) == [BackingServer("slack"), BackingServer("email"), - BackingServer("shop", mcp=False), BackingServer("slack", mcp=False)] - assert legacy_protocol.shared_names(legacy_protocol.backing_servers(multi)) == {"slack"} - assert legacy_protocol.mcp_wins(legacy_protocol.backing_servers(multi)) == [ - BackingServer("slack"), BackingServer("email"), BackingServer("shop", mcp=False)] - assert legacy_protocol.backing_servers(WebsiteEnv(id="w", version=1, environment_name="shop")) == [BackingServer("shop", mcp=False)] - assert legacy_protocol.backing_servers(SimpleNamespace(environment_name="notes")) == [BackingServer("notes")] - assert legacy_protocol.backing_servers(SimpleNamespace()) == [] + assert legacy_protocol.definition_servers(multi) == [BackingServer("slack"), BackingServer("email"), + BackingServer("shop", mcp=False), BackingServer("slack", mcp=False)] + assert legacy_protocol.definition_servers(WebsiteEnv(id="w", version=1, environment_name="shop")) == [BackingServer("shop", mcp=False)] + assert legacy_protocol.definition_servers(SimpleNamespace(environment_name="notes")) == [BackingServer("notes")] + assert legacy_protocol.definition_servers(SimpleNamespace()) == [] + + +_COMPOSED = {"name": "gw", "url": RPC_PATH, "capabilities": {"extensions": []}, "children_environments": [ + {"name": "slack", "url": "/svc/mcp-slack/agentenv", "capabilities": {"extensions": []}}, + {"name": "slack", "url": "/svc/slack/agentenv", "capabilities": {"extensions": []}}, + {"name": "docs", "url": "/svc/mcp-docs/agentenv", "capabilities": {"extensions": [{"uri": "urn:agentenv:website/v1"}]}}, +]} + + +def test_card_servers_are_the_cards_children_a_website_told_apart_by_its_route_or_its_website_extension(): + servers = legacy_protocol.card_servers(_COMPOSED, "https://env.example") + + assert [(s.environment_name, s.mcp, s.address) for s in servers] == [ + ("slack", True, "https://env.example"), ("slack", False, "https://env.example"), ("docs", False, "https://env.example")] + assert [s.child.base_url(s.address) for s in servers] == [ + "https://env.example/svc/mcp-slack", "https://env.example/svc/slack", "https://env.example/svc/mcp-docs"] + + +def test_card_servers_of_a_leaf_card_is_the_env_itself_and_a_composed_card_serving_data_get_comes_first(): + leaf = {"name": "notes", "url": RPC_PATH} + assert [(s.environment_name, s.card) for s in legacy_protocol.card_servers(leaf, "https://n")] == [("notes", leaf)] + driver = {**_COMPOSED, "name": "desktop", "capabilities": {"operations": ["data/get"]}} + assert [s.environment_name for s in legacy_protocol.card_servers(driver, "https://d")] == ["desktop", "slack", "slack", "docs"] + + +def _record(card: dict | None, card_url: str | None = "https://env.example/.well-known/agent-env.json"): + from agent_env.env.env import DeployedGatewayEnv + + return DeployedGatewayEnv(env_id="e", env_version=1, gateway_url="http://gw", mcp_url="http://gw/mcp", db_web_url=None, + sandbox_id="sb", environment_card_url=card_url, environment_card=card) + + +@pytest.mark.asyncio +async def test_backing_servers_are_the_stored_cards_then_those_the_definition_names_serving_no_card(): + from types import SimpleNamespace + + env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name=n) for n in ("slack", "legacy")], + website_envs=[SimpleNamespace(environment_name="slack")]) + + servers = await legacy_protocol.backing_servers(_record(_COMPOSED), "http://gw", env) + + assert [(s.environment_name, s.mcp, s.card is not None, s.address) for s in servers] == [ + ("slack", True, True, "https://env.example"), ("slack", False, True, "https://env.example"), + ("docs", False, True, "https://env.example"), ("legacy", True, False, "https://env.example")] + assert await legacy_protocol.server_card(servers[-1], "http://gw") == ("https://env.example", None) + assert await legacy_protocol.server_v1_base_url(servers[-1], "http://gw") is None + + +@pytest.mark.asyncio +async def test_backing_servers_read_the_card_live_without_a_stored_one_or_for_another_gateway(monkeypatch): + from types import SimpleNamespace + + read = [] + + async def get_card(base_url, **kw): + read.append(base_url) + return _COMPOSED + + monkeypatch.setattr(legacy_protocol.protocol_v1, "get_card", get_card) + env = SimpleNamespace(environment_name="slack") + for record, gateway in ((_record(None), "http://gw/"), (_record(_COMPOSED), "http://elsewhere")): + servers = await legacy_protocol.backing_servers(record, gateway, env) + assert [(s.environment_name, s.address) for s in servers] == [ + ("slack", gateway.rstrip("/")), ("slack", gateway.rstrip("/")), ("docs", gateway.rstrip("/"))] + assert read == ["http://gw", "http://elsewhere"] + + +@pytest.mark.asyncio +async def test_a_record_whose_card_cant_be_read_falls_back_to_its_definition_logged_once(monkeypatch, caplog): + from types import SimpleNamespace + + async def get_card(base_url, **kw): + raise httpx.ConnectError("refused") + + monkeypatch.setattr(legacy_protocol.protocol_v1, "get_card", get_card) + env = SimpleNamespace(id="e", mcp_server_envs=[SimpleNamespace(environment_name="slack")], + website_envs=[SimpleNamespace(environment_name="slack")]) + with caplog.at_level("INFO", logger="agent_env.env.legacy_protocol"): + for record in (_record(None, card_url=None), _record(None), None): + servers = await legacy_protocol.backing_servers(record, "http://gw", env) + assert [(s.environment_name, s.mcp, s.address) for s in servers] == [("slack", True, None), ("slack", False, None)] + + assert [r.getMessage().split(",")[0] for r in caplog.records] == [ + "env 'e': its record has no card URL", "env 'e': its card at http://gw could not be read (ConnectError)", + "env 'e': it has no record"] diff --git a/tst/unit/task/local_run_write_fence_test.py b/tst/unit/task/local_run_write_fence_test.py index d3f60293e..108709d94 100644 --- a/tst/unit/task/local_run_write_fence_test.py +++ b/tst/unit/task/local_run_write_fence_test.py @@ -264,7 +264,7 @@ def __init__(self, id, version=None): self.mcp_server_envs = [SimpleNamespace(environment_name="gmail")] -async def _export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): +async def _export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): Path(tmp_path).write_text(json.dumps({"service": name})) return ".json" diff --git a/tst/unit/task_step/sync_env_clock_test.py b/tst/unit/task_step/sync_env_clock_test.py index 5b83c51d7..a9a109e2a 100644 --- a/tst/unit/task_step/sync_env_clock_test.py +++ b/tst/unit/task_step/sync_env_clock_test.py @@ -1,11 +1,11 @@ """Unit tests for SyncEnvClockTaskStep — round-trip + the per-server sync_time fan-out (mocked).""" import json +from types import SimpleNamespace import pytest from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, EnvCapabilityUnsupported from agent_env.env.gateway.constants import EXT_CLOCK_URI -from agent_env.env.legacy_protocol import BackingServer from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_steps import sync_env_clock as mod from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep, resolve_virtual_time @@ -95,7 +95,7 @@ async def test_sync_one_skips_or_raises_for_a_child_env_missing_from_the_stored_ @pytest.mark.asyncio async def test_execute_syncs_each_child_env_through_the_stored_card(monkeypatch): sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) - monkeypatch.setattr(SyncEnvClockTaskStep, "_backing_servers", lambda self, deployed: [BackingServer("email")]) + monkeypatch.setattr("agent_env.env.env.Env.get", lambda *a, **k: SimpleNamespace(environment_name="email")) context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]}))]) await _step().execute(context) @@ -176,21 +176,64 @@ async def fake_card(base_url, **kw): @pytest.mark.asyncio -async def test_execute_reads_a_website_sharing_an_mcp_servers_name_live(monkeypatch): - from types import SimpleNamespace +async def test_execute_syncs_a_website_sharing_an_mcp_servers_name_at_its_own_route(monkeypatch): + """The card lists the two as children at distinct urls, so each is synced by its own card, with no card read.""" from unittest.mock import patch sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name="email")], website_envs=[SimpleNamespace(environment_name="email")]) - context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]}))]) + context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_CLOCK_EXT], "email": [_CLOCK_EXT]}))]) with patch("agent_env.env.env.Env.get", return_value=env): await _step().execute(context) assert [(r.method, str(r.url)) for r in sent][2:] == [ - ("POST", f"{_CARD}/svc/mcp-email/agentenv/ext/sync_time"), ("GET", "http://gw/svc/email/.well-known/agent-env.json"), + ("POST", f"{_CARD}/svc/mcp-email/agentenv/ext/sync_time"), ("POST", f"{_CARD}/svc/email/agentenv/ext/sync_time"), ] + record = context.metadata["clock_configurations"][0] + assert [(s["service"], s.get("website")) for s in record["synced"]] == [("email", None), ("email", True)] + + +@pytest.mark.asyncio +async def test_execute_without_a_card_url_syncs_the_servers_the_definition_names_read_live(monkeypatch): + """An older record kept no card URL, so its servers come from its env definition and each card is read live at + its gateway path.""" + from unittest.mock import patch + + import httpx + + sent, real = [], httpx.AsyncClient + cards = {"/svc/mcp-email/.well-known/agent-env.json": {"name": "email", "capabilities": {"extensions": [_CLOCK_EXT]}}, + "/svc/shop/.well-known/agent-env.json": {"name": "shop", "capabilities": {"extensions": []}}} + + def handle(request: httpx.Request) -> httpx.Response: + sent.append(request) + if request.url.path == "/clock/state": + return httpx.Response(200, json={"env_get_time_url": CLOCK_URL}) + return httpx.Response(200, json=cards.get(request.url.path, {})) + + monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle))) + record = _cardless() + + async def invoke(uri, method, params=None, **kw): + return {"env_get_time_url": CLOCK_URL} if method == "state" else {} + + record.invoke = invoke + env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name="email")], + website_envs=[SimpleNamespace(environment_name="shop")]) + context = TaskStepContext(deployed_envs=[record]) + + with patch("agent_env.env.env.Env.get", return_value=env): + await _step().execute(context) + + assert [(r.method, str(r.url)) for r in sent] == [ + ("GET", "http://gw/svc/mcp-email/.well-known/agent-env.json"), ("POST", "http://gw/svc/mcp-email/agentenv/ext/sync_time"), + ("GET", "http://gw/svc/shop/.well-known/agent-env.json"), + ] + cfg = context.metadata["clock_configurations"][0] + assert [s["service"] for s in cfg["synced"]] == ["email"] + assert [(s["service"], s["reason"]) for s in cfg["skipped"]] == [("shop", "not_advertised")] @pytest.mark.asyncio @@ -317,7 +360,7 @@ def test_without_a_zone_virtual_time_is_passed_through_untouched(): @pytest.mark.asyncio async def test_execute_arms_the_gateway_at_the_local_instant_and_tells_every_server_the_zone(monkeypatch): sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) - monkeypatch.setattr(SyncEnvClockTaskStep, "_backing_servers", lambda self, deployed: [BackingServer("email")]) + monkeypatch.setattr("agent_env.env.env.Env.get", lambda *a, **k: SimpleNamespace(environment_name="email")) context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_ZONE_EXT]}))]) await _zoned().execute(context) @@ -332,7 +375,7 @@ async def test_execute_arms_the_gateway_at_the_local_instant_and_tells_every_ser @pytest.mark.parametrize("tolerate", [True, False]) async def test_a_server_that_cannot_take_the_zone_fails_the_step_before_any_sync_is_sent(monkeypatch, tolerate): sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) - monkeypatch.setattr(SyncEnvClockTaskStep, "_backing_servers", lambda self, deployed: [BackingServer("email")]) + monkeypatch.setattr("agent_env.env.env.Env.get", lambda *a, **k: SimpleNamespace(environment_name="email")) context = TaskStepContext(deployed_envs=[_carded(await _composed_card({"mcp-email": [_OLD_EXT]}))]) with pytest.raises(RuntimeError, match="takes no timezone"): @@ -378,34 +421,19 @@ async def test_a_zoned_step_tells_a_website_that_takes_the_zone_and_fails_on_one @pytest.mark.asyncio async def test_a_website_sharing_an_mcp_servers_name_is_synced_with_the_envs_control_headers(monkeypatch): - """Its card is read live, with no record, and its sync_time still carries the env's control token.""" import dataclasses - from types import SimpleNamespace from unittest.mock import patch - import httpx - - sent, real = [], httpx.AsyncClient - website_card = {"name": "email", "url": "/agentenv", "capabilities": {"extensions": [_CLOCK_EXT]}} - - def handle(request: httpx.Request) -> httpx.Response: - sent.append(request) - if request.url.path.endswith("/clock/state"): - return httpx.Response(200, json={"env_get_time_url": CLOCK_URL}) - if request.url.path == "/svc/email/.well-known/agent-env.json": - return httpx.Response(200, json=website_card) - return httpx.Response(200, json={}) - - monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: real(transport=httpx.MockTransport(handle))) + sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) env = SimpleNamespace(mcp_server_envs=[SimpleNamespace(environment_name="email")], website_envs=[SimpleNamespace(environment_name="email")]) - record = dataclasses.replace(_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]})), control_token="ctl") + record = dataclasses.replace(_carded(await _composed_card({"mcp-email": [_CLOCK_EXT], "email": [_CLOCK_EXT]})), control_token="ctl") context = TaskStepContext(deployed_envs=[record]) with patch("agent_env.env.env.Env.get", return_value=env): await _step().execute(context) - website_sync = [r for r in sent if str(r.url) == "http://gw/svc/email/agentenv/ext/sync_time"] + website_sync = [r for r in sent if str(r.url) == f"{_CARD}/svc/email/agentenv/ext/sync_time"] assert len(website_sync) == 1 assert {k: website_sync[0].headers.get(k) for k in record.control_headers} == record.control_headers assert [s["service"] for s in context.metadata["clock_configurations"][0]["synced"]] == ["email", "email"] diff --git a/tst/unit/task_step/test_apply_server_config.py b/tst/unit/task_step/test_apply_server_config.py index ddeb96806..94ed35986 100644 --- a/tst/unit/task_step/test_apply_server_config.py +++ b/tst/unit/task_step/test_apply_server_config.py @@ -609,18 +609,38 @@ async def get(self, url, *, timeout=None, params=None): @pytest.mark.asyncio -async def test_stored_card_broadcast_reaches_websites_and_leaves_a_shared_name_to_the_mcp_server(): +async def test_stored_card_broadcast_reaches_websites_and_a_website_sharing_an_mcp_servers_name(): + """The card lists a website and an MCP server sharing a name as two children at distinct urls, so a broadcast + reaches both, each at its own route; a directive naming the name means the MCP server.""" gw = _MockGateway({}) ext = _ext(_SET_ACTING_USER_URI, "/agentenv/ext/set_acting_user") - ctx = _carded_context(await _composed_card({"mcp-slack": [ext], "shop": [ext]})) - step = _step(directives=[{"service": "*", "uri": _SET_ACTING_USER_URI, "args": {"user_email": "a@x.com"}}]) + ctx = _carded_context(await _composed_card({"mcp-slack": [ext], "shop": [ext], "slack": [ext]})) + step = _step(directives=[{"service": "*", "uri": _SET_ACTING_USER_URI, "args": {"user_email": "a@x.com"}}, + {"service": "slack", "uri": _SET_ACTING_USER_URI, "args": {"user_email": "b@x.com"}}]) with _patch_httpx(gw), patch("agent_env.env.env.Env.get", return_value=_fake_multi_env(["slack"], ["shop", "slack"])): out = await step.execute(ctx) + assert gw.get_calls == [] + assert gw.post_calls == [(f"{_CARD}/svc/mcp-slack/agentenv/ext/set_acting_user", {"user_email": "a@x.com"}), + (f"{_CARD}/svc/shop/agentenv/ext/set_acting_user", {"user_email": "a@x.com"}), + (f"{_CARD}/svc/slack/agentenv/ext/set_acting_user", {"user_email": "a@x.com"}), + (f"{_CARD}/svc/mcp-slack/agentenv/ext/set_acting_user", {"user_email": "b@x.com"})] + assert [c["service"] for c in out.metadata["server_config_changes"]] == ["slack", "shop", "slack", "slack"] + + +@pytest.mark.asyncio +async def test_stored_card_broadcast_to_an_mcp_only_env_is_unchanged(): + gw = _MockGateway({}) + ext = _ext(_SET_ACTING_USER_URI, "/agentenv/ext/set_acting_user") + ctx = _carded_context(await _composed_card({"mcp-slack": [ext], "mcp-email": [ext]})) + step = _step(directives=[{"service": "*", "uri": _SET_ACTING_USER_URI, "args": {"user_email": "a@x.com"}}]) + with _patch_httpx(gw), patch("agent_env.env.env.Env.get", return_value=_fake_env("slack", "email")): + out = await step.execute(ctx) + assert gw.get_calls == [] assert [url for url, _ in gw.post_calls] == [f"{_CARD}/svc/mcp-slack/agentenv/ext/set_acting_user", - f"{_CARD}/svc/shop/agentenv/ext/set_acting_user"] - assert [c["service"] for c in out.metadata["server_config_changes"]] == ["slack", "shop"] + f"{_CARD}/svc/mcp-email/agentenv/ext/set_acting_user"] + assert out.metadata["server_config_skipped"] == [] @pytest.mark.asyncio diff --git a/tst/unit/task_step/test_snapshot_env.py b/tst/unit/task_step/test_snapshot_env.py index 380690fc2..cf75fb7ca 100644 --- a/tst/unit/task_step/test_snapshot_env.py +++ b/tst/unit/task_step/test_snapshot_env.py @@ -14,6 +14,7 @@ from agent_env.artifact import EnvironmentUniverseArtifact from agent_env.artifact.store import reset_artifact_store from agent_env.config import set_object_store +from agent_env.env import legacy_protocol from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, DeployedSandboxEnv from agent_env.env.gateway.constants import EXT_STATE_URI, GATEWAY_EXTENSIONS, WELL_KNOWN_PATH from agent_env.task_step.context import TaskStepContext @@ -33,7 +34,7 @@ def _deployed(env_id="env-1", instance_id="env-1-abc123", gateway="https://gw-1" sandbox_id="sb-1", instance_id=instance_id, environment_card_url=f"{gateway}{WELL_KNOWN_PATH}", - environment_card={"capabilities": {"extensions": GATEWAY_EXTENSIONS}}, + environment_card={"capabilities": {"extensions": GATEWAY_EXTENSIONS}, "children_environments": []}, ) @@ -565,7 +566,7 @@ def __init__(self, names): @pytest.mark.asyncio async def test_core_all_or_nothing_on_partial_failure(monkeypatch): - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): if name == "slack": raise ValueError("export-state returned non-JSON body (starts with b'<')") with open(tmp_path, "w") as f: @@ -609,7 +610,7 @@ def put(id, *, environment_name, file_artifact): @pytest.mark.asyncio async def test_core_success_creates_universe_with_metadata(monkeypatch): - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): with open(tmp_path, "w") as f: json.dump({"ok": name}, f) return ".json" @@ -673,7 +674,7 @@ async def test_core_zip_export_creates_zip_filename_artifact(monkeypatch): """A FilePart(zip) export flows through as a .zip-suffixed FileArtifact path, so FileArtifact.filename (= basename) ends in .zip — the load side routes a .zip file:// artifact to reset_data() for a lossless round-trip.""" - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): # Simulate the lossless-zip case: write zip bytes and report ".zip". with open(tmp_path, "wb") as f: f.write(b"PK\x03\x04zip-bytes") @@ -911,7 +912,7 @@ async def test_core_registers_a_service_upload_in_place_and_it_loads_back(local_ bundle_file.write_bytes(b"PK\x03\x04bundle") uploaded = store.put_file_at(upload_url, str(bundle_file)) - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): return uploaded monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(fake_export)) @@ -933,7 +934,7 @@ async def test_core_fails_a_service_whose_upload_the_store_cannot_find(local_sto set_object_store(FakeObjectStore()) reset_artifact_store() - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): return "fake://shared/snapshots/missing.zip" monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(fake_export)) @@ -950,7 +951,7 @@ async def test_core_snapshots_under_an_id_longer_than_a_filename_can_be(local_st local store may be the whole id.""" snapshot_id = f"task-{'x' * 250}__snapshot-snap-1" - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): with open(tmp_path, "w") as f: json.dump({"service": name}, f) return ".json" @@ -1044,14 +1045,16 @@ async def test_export_of_a_child_env_missing_from_the_card_streams_legacy_export @pytest.mark.asyncio async def test_snapshot_uses_the_stored_card_only_for_the_deployments_own_gateway(monkeypatch): + """An override's gateway has its card read live, and the record's stored card describes its own gateway only.""" seen = [] - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): - seen.append((gateway_url, deployed)) + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, server=None, **_): + seen.append((gateway_url, deployed, server.address, server.card and server.card["url"])) raise ValueError("stop after the export call") monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(fake_export)) - record = _carded({"name": "gw", "children_environments": []}) + sent = _mock_http(monkeypatch, body=json.dumps(await _card({"mcp-slack": []})).encode()) + record = _carded(await _card({"mcp-slack": []})) record.gateway_url = "https://gw-1/" for gateway_url in ("https://gw-1/", "https://elsewhere"): with pytest.raises(RuntimeError, match="not exported"): @@ -1059,7 +1062,9 @@ async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_ env=_FakeMultiEnv(["slack"]), gateway_url=gateway_url, snapshot_id="snap-x", deployed=record, ) - assert seen == [("https://gw-1", record), ("https://elsewhere", None)] + assert seen == [("https://gw-1", record, _CARD, "/svc/mcp-slack/agentenv"), + ("https://elsewhere", None, "https://elsewhere", "/svc/mcp-slack/agentenv")] + assert [str(r.url) for r in sent] == [f"https://elsewhere{WELL_KNOWN_PATH}"] @pytest.mark.asyncio @@ -1081,19 +1086,26 @@ async def test_export_with_a_record_that_has_no_stored_card_keeps_the_live_probe ] -def test_enumerate_services_single_mcp_server_env(): +@pytest.mark.asyncio +async def test_a_record_without_a_card_snapshots_the_servers_its_definition_names(): env = _FakeServiceEnv("calendar") env.id, env.version = "cal-env", 1 - assert SnapshotEnvTaskStep._enumerate_environments(env) == ["calendar"] + servers = await legacy_protocol.backing_servers(_deployed_without_a_card(), "https://gw-1", env) + assert [(s.environment_name, s.mcp, s.address) for s in SnapshotEnvTaskStep._snapshot_servers(servers, env)] == [ + ("calendar", True, None)] -def test_enumerate_services_no_services_raises(): +def test_snapshot_servers_refuses_an_env_with_none(): env = type("E", (), {"id": "x", "version": 1})() with pytest.raises(RuntimeError, match="no MCP services"): - SnapshotEnvTaskStep._enumerate_environments(env) + SnapshotEnvTaskStep._snapshot_servers(legacy_protocol.definition_servers(env), env) -# ── loaded-universe metadata carry-over ────────────────────────────────────── +def _deployed_without_a_card() -> DeployedEnv: + """An older record: no card URL, so its servers are the ones its definition names.""" + return DeployedGatewayEnv(env_id="env-1", env_version=3, gateway_url="https://gw-1", mcp_url="https://gw-1/mcp", + db_web_url=None, sandbox_id="sb-1") + def test_resolve_loaded_universe_ref_reads_store(monkeypatch): import agent_env.env.store as store_mod @@ -1139,7 +1151,7 @@ def get_environment_universe(self, instance_id): def _install_core_artifact_fakes(monkeypatch, universe_cls): """Wire the export + FileArtifact/EnvironmentArtifact fakes shared by the core carry-over tests; the caller supplies the EnvironmentUniverseArtifact fake.""" - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **_): with open(tmp_path, "w") as f: json.dump({"ok": name}, f) return ".json" @@ -1199,7 +1211,7 @@ def get_environment_universe(self, instance_id): await SnapshotEnvTaskStep.snapshot_env_state( env=_FakeMultiEnv(["slack"]), - gateway_url="https://gw", + gateway_url="https://gw-1", snapshot_id="snap-x", deployed=_deployed(instance_id="env-1-abc123"), ) @@ -1240,7 +1252,7 @@ def get_environment_universe(self, instance_id): await SnapshotEnvTaskStep.snapshot_env_state( original_universe_artifact_id="explicit-base", env=_FakeMultiEnv(["slack"]), - gateway_url="https://gw", + gateway_url="https://gw-1", snapshot_id="snap-x", deployed=_deployed(), ) @@ -1276,7 +1288,7 @@ def get_environment_universe(self, instance_id): result = await SnapshotEnvTaskStep.snapshot_env_state( env=_FakeMultiEnv(["slack"]), - gateway_url="https://gw", + gateway_url="https://gw-1", snapshot_id="snap-x", deployed=_deployed(), ) @@ -1302,7 +1314,7 @@ def get(artifact_id, version=None): await SnapshotEnvTaskStep.snapshot_env_state( original_universe_artifact_id="bad", env=_FakeMultiEnv(["slack"]), - gateway_url="https://gw", + gateway_url="https://gw-1", snapshot_id="snap-x", deployed=_deployed(), ) @@ -1376,17 +1388,103 @@ def _multi_env(mcp_names, website_names): website_envs=[_FakeServiceEnv(n) for n in website_names]) -def test_enumerate_includes_websites_and_leaves_a_shared_name_to_the_mcp_server(): - servers = SnapshotEnvTaskStep._enumerate_servers(_multi_env(["slack", "gmail"], ["shop", "slack"])) - assert [(s.environment_name, s.mcp) for s in servers] == [("slack", True), ("gmail", True), ("shop", False)] +@pytest.mark.asyncio +async def test_the_servers_snapshotted_are_the_cards_children_one_per_name_and_those_serving_no_card(caplog): + """A website sharing an MCP server's name is its own server on the card, left out of the snapshot (logged) since a + universe names an environment by its name; a server serving no card, which the card leaves off, keeps its legacy + export.""" + record = _carded(await _card({"mcp-slack": [], "mcp-gmail": [], "shop": [], "slack": []})) + env = _multi_env(["slack", "gmail", "legacy"], ["shop", "slack"]) + + listed = await legacy_protocol.backing_servers(record, "https://gw-1", env) + with caplog.at_level("WARNING"): + servers = SnapshotEnvTaskStep._snapshot_servers(listed, env) + + assert [(s.environment_name, s.mcp, s.card and s.card["url"]) for s in listed] == [ + ("slack", True, "/svc/mcp-slack/agentenv"), ("gmail", True, "/svc/mcp-gmail/agentenv"), + ("shop", False, "/svc/shop/agentenv"), ("slack", False, "/svc/slack/agentenv"), ("legacy", True, None)] + assert [(s.environment_name, s.mcp) for s in servers] == [("slack", True), ("gmail", True), ("legacy", True), ("shop", False)] + assert "website 'slack' of env-1 is not snapshotted: it shares its name with the MCP server kept" in caplog.text -def test_enumerate_a_single_website_env_and_a_multi_env_of_websites_only(): +@pytest.mark.asyncio +async def test_a_single_website_env_and_a_multi_env_of_websites_only_are_snapshotted_as_websites(): from agent_env.env.envs.website import WebsiteEnv site = WebsiteEnv(id="site-env", version=1, environment_name="shop") - assert [(s.environment_name, s.mcp) for s in SnapshotEnvTaskStep._enumerate_servers(site)] == [("shop", False)] - assert SnapshotEnvTaskStep._enumerate_environments(_multi_env([], ["shop"])) == ["shop"] + assert [(s.environment_name, s.mcp) for s in legacy_protocol.definition_servers(site)] == [("shop", False)] + record = _carded(await _card({"shop": []})) + assert [(s.environment_name, s.mcp) for s in await legacy_protocol.backing_servers(record, "https://gw-1", _multi_env([], ["shop"]))] == [ + ("shop", False)] + + +@pytest.mark.asyncio +async def test_a_card_only_website_is_left_out_as_serving_no_v1_data_plane(monkeypatch, tmp_path): + """Its card declares no data/get, so it is skipped with no request, not failed on a data/get it doesn't serve.""" + record = _carded(await _card({"mcp-slack": [], "shop": []})) + record.environment_card["children_environments"][1]["capabilities"]["operations"] = [] + (shop,) = [s for s in await legacy_protocol.backing_servers(record, "https://gw-1", _multi_env(["slack"], ["shop"])) + if s.environment_name == "shop"] + sent = _mock_http(monkeypatch) + + assert await SnapshotEnvTaskStep._export_environment_to_file("https://gw-1", "shop", str(tmp_path / "out"), 30, record, + server=shop) is None + assert sent == [] + + +@pytest.mark.asyncio +async def test_a_card_only_website_is_reported_as_serving_no_v1_data_plane(monkeypatch): + real_export = SnapshotEnvTaskStep.__dict__["_export_environment_to_file"] + _install_core_artifact_fakes(monkeypatch, EnvironmentUniverseArtifact) + monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", real_export) + record = _carded(await _card({"shop": []})) + record.environment_card["children_environments"][0]["capabilities"]["operations"] = ["data/reset"] + sent = _mock_http(monkeypatch) + + with pytest.raises(RuntimeError, match=r"exported no state: \['shop'\] serve no v1 data plane"): + await SnapshotEnvTaskStep.snapshot_env_state(env=_multi_env([], ["shop"]), gateway_url="https://gw-1", + snapshot_id="snap-x", deployed=record) + assert sent == [] + + +@pytest.mark.asyncio +async def test_an_mcp_only_env_snapshots_each_server_at_its_route_and_one_serving_no_card_by_export_state(monkeypatch): + real_export = SnapshotEnvTaskStep.__dict__["_export_environment_to_file"] + universe_puts = {} + + class _FakeUniverse: + id, version = "snap-x", 1 + + @staticmethod + def put(id, *, environment_artifacts, metadata=None): + universe_puts["names"] = [sa.environment_name for sa in environment_artifacts] + return _FakeUniverse() + + _install_core_artifact_fakes(monkeypatch, _FakeUniverse) + monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", real_export) + sent = _mock_http(monkeypatch, rpc_result={"parts": [{"kind": "data", "data": {"rows": 1}}]}, body=b'{"rows": 2}') + record = _carded(await _card({"mcp-slack": [], "mcp-gmail": []})) + + result = await SnapshotEnvTaskStep.snapshot_env_state(env=_FakeMultiEnv(["slack", "gmail", "legacy"]), gateway_url="https://gw-1", + snapshot_id="snap-x", deployed=record) + + assert sorted((r.method, str(r.url)) for r in sent) == [ + ("GET", "https://gw-1/svc/mcp-legacy/export-state"), ("POST", f"{_CARD}/svc/mcp-gmail/agentenv"), + ("POST", f"{_CARD}/svc/mcp-slack/agentenv")] + assert universe_puts["names"] == ["slack", "gmail", "legacy"] and result.total == 3 + + +@pytest.mark.asyncio +async def test_a_website_sharing_an_mcp_servers_name_is_exported_at_its_own_route(monkeypatch, tmp_path): + """The card tells the two apart by their urls, so the website's data/get goes to /svc/slack, not the MCP server.""" + record = _carded(await _card({"mcp-slack": [], "slack": []})) + servers = await legacy_protocol.backing_servers(record, "https://gw-1", _multi_env(["slack"], ["slack"])) + sent = _mock_http(monkeypatch, rpc_result={"parts": [{"kind": "data", "data": {"orders": 2}}]}) + + for server in servers: + assert await SnapshotEnvTaskStep._export_environment_to_file("https://gw-1", "slack", str(tmp_path / "out"), 30, record, + server=server) == ".json" + assert [str(r.url) for r in sent] == [f"{_CARD}/svc/mcp-slack/agentenv", f"{_CARD}/svc/slack/agentenv"] @pytest.mark.asyncio @@ -1500,7 +1598,7 @@ async def test_export_of_a_website_serving_no_data_plane_exports_nothing_and_sen async def test_core_snapshots_websites_through_their_data_plane_and_leaves_out_those_without_one(monkeypatch): calls = [] - async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, **website): + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, data_get_params=None, server=None, **website): calls.append((name, website)) if name == "legacy-site": return None @@ -1539,5 +1637,5 @@ async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None, **web _install_core_artifact_fakes(monkeypatch, EnvironmentUniverseArtifact) monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(fake_export)) - with pytest.raises(RuntimeError, match=r"exported no state: its websites \['shop'\] serve no v1 data plane"): + with pytest.raises(RuntimeError, match=r"exported no state: \['shop'\] serve no v1 data plane"): await SnapshotEnvTaskStep.snapshot_env_state(env=_multi_env([], ["shop"]), gateway_url="https://gw", snapshot_id="snap-x")