diff --git a/src/agent_env/env/legacy_protocol.py b/src/agent_env/env/legacy_protocol.py index 16899025..8790157a 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 41f82ffd..c8c8e1f0 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 48fc9d6d..9415dac9 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 76ff196e..c80cef8a 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 9508e8d2..7d107ad3 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 6ccb7ad7..d86b71d9 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 5ab7bee8..5b83c51d 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 f9badb26..ddeb9680 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 877244c3..380690fc 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")