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 16899025b..f5ca7a002 100644 --- a/src/agent_env/env/legacy_protocol.py +++ b/src/agent_env/env/legacy_protocol.py @@ -8,11 +8,15 @@ from __future__ import annotations import asyncio +import dataclasses 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 +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 @@ -33,20 +37,132 @@ 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 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 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): + 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 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]: - """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..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 @@ -1,22 +1,24 @@ """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; -only this harness step invokes them. +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. 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__) @@ -76,7 +81,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,43 +136,39 @@ def from_dict(cls, data: dict) -> "ApplyServerConfigStep": tolerate_unadvertised=data.get("tolerate_unadvertised", False), ) - def _expand_directives(self, deployed) -> list[ConfigDirective]: - """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.""" + 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 + return [(d, None, False) for d in self.directives] - from agent_env.env.env import Env + from agent_env.env import legacy_protocol + from agent_env.env.env import Env, gateway_url_of - 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] + 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 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"environment_name); nothing to apply, likely a task authoring error" + f"apply_server_config: broadcast directive matched no MCP servers or websites " + 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] = [] + 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 svc in environment_names: - expanded.append(ConfigDirective(service=svc, uri=d.uri, args=d.args)) + for server in servers: + 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: @@ -180,7 +181,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 +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 = self._expand_directives(deployed) + directives = await self._expand_directives(deployed) except ApplyServerConfigError: raise except Exception as e: @@ -203,16 +204,19 @@ 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 directive in directives: + 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 - ) - 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 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 +224,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 +237,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..abb5573f3 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"]``. @@ -470,27 +476,42 @@ def _resolve_loaded_universe_ref(deployed) -> Optional[tuple[str, Optional[int]] return (su["id"], su.get("version")) if su and su.get("id") else None @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 _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 " + "(its card lists no child envs, and its definition names no mcp_server_envs, website_envs or environment_name)" + ) + 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, - ) -> 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, 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 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`` → @@ -499,10 +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=True) + 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) 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) @@ -609,10 +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 - services = cls._enumerate_environments(env) if gateway_url else _card_environments(deployed) 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} " @@ -643,9 +678,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 service 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}-") @@ -657,6 +695,7 @@ async def export_one(environment_name: str): 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), + server=server, **website, ) except httpx.HTTPStatusError as e: body_snippet = "" @@ -671,6 +710,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 +786,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 +809,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: {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"{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( @@ -849,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 76ff196e8..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,8 +1,9 @@ """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 +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. """ @@ -10,7 +11,7 @@ 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 +23,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,14 +111,13 @@ 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).""" - 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 + 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, 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.env import gateway_url_of @@ -132,9 +135,12 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: env_get_time_url = state.get("env_get_time_url") 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 await self._backing_servers(deployed): + service = server.environment_name + 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) configuration = { "step_id": self.id, "env_id": self.env_id, @@ -151,17 +157,27 @@ 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, 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. ``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) + 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 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) 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..ef8fee3f8 100644 --- a/tst/unit/env/test_legacy_protocol.py +++ b/tst/unit/env/test_legacy_protocol.py @@ -177,3 +177,104 @@ 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_definition_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.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/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/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 5ab7bee8e..a9a109e2a 100644 --- a/tst/unit/task_step/sync_env_clock_test.py +++ b/tst/unit/task_step/sync_env_clock_test.py @@ -1,5 +1,6 @@ """Unit tests for SyncEnvClockTaskStep — round-trip + the per-server sync_time fan-out (mocked).""" import json +from types import SimpleNamespace import pytest @@ -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("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) @@ -124,6 +125,139 @@ 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_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], "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"), ("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 +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 +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, "_service_names", lambda self, deployed: ["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) @@ -241,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, "_service_names", lambda self, deployed: ["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"): @@ -264,6 +398,47 @@ 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): + import dataclasses + 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")]) + 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) == 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"] + + 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..94ed35986 100644 --- a/tst/unit/task_step/test_apply_server_config.py +++ b/tst/unit/task_step/test_apply_server_config.py @@ -589,6 +589,117 @@ 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_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], "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/mcp-email/agentenv/ext/set_acting_user"] + assert out.metadata["server_config_skipped"] == [] + + +@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..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) + +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") -# ── loaded-universe metadata carry-over ────────────────────────────────────── 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(), ) @@ -1367,3 +1379,263 @@ 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]) + + +@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 + + +@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 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 +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, server=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: \['shop'\] serve no v1 data plane"): + await SnapshotEnvTaskStep.snapshot_env_state(env=_multi_env([], ["shop"]), gateway_url="https://gw", snapshot_id="snap-x")