diff --git a/packages/agentenv-protocol/README.md b/packages/agentenv-protocol/README.md index fe1176eda..8cc9fb639 100644 --- a/packages/agentenv-protocol/README.md +++ b/packages/agentenv-protocol/README.md @@ -505,8 +505,8 @@ driver has them, through the `served_tools`, `served_extensions` and `served_ope or a `transfers.ReadObject` read grant under the marker's `read`, and writes them atomically. `data/get` returns the files under the `paths` it's given, relative to the home, or under the card's `default_data_paths` (`Desktop`, `Documents`, `Downloads`) when it's given none, inline or - uploaded under a `write_namespace` grant. An environment whose `@get_data` hook takes no `paths` - refuses them. + uploaded under a `write_namespace` grant, leaving out the paths its apps keep their data in. An + environment whose `@get_data` hook takes no `paths` refuses them. - **The card** (the name from `ENVIRONMENT_NAME`, then `@computer_driver`'s), and per-role tool access (`tool_access` is on). @@ -674,8 +674,10 @@ class MailApp(ComputerAppEnvironment): naming one the app declares no install for is refused when the class is read. A step reads `self.build_settings`; a deploy's `self.settings` aren't known at build, and reading one raises `AttributeError` (`python -m agentenv_protocol.computer.run_install_step : `). -- **Its skill** is a file of its folder (`skill = "skill.md"`) or an `@agent_skill` method, not both; - its data hooks are the usual `@add_data`, `@get_data` and `@reset_data`. +- **Its skill** is a file of its folder (`skill = "skill.md"`) or an `@agent_skill` method, not both. +- **Its data** is under the `paths` its OS's decorator names, relative to the session user's home, + none inside another (`@linux(..., paths=["Mail", ".config/mail/accounts.json"])`), or what its own + `@add_data`, `@get_data` and `@reset_data` hooks serve instead (below). - **`Settings`** is typed; a computer hands each app the values a deploy gives it, as one JSON value in `APP_SETTINGS_ENV` (`AGENTENV_APP_SETTINGS`, the apps' values by name), and the base class reads its own, by `ENVIRONMENT_NAME`, as `self.settings`. A `SecretStr` field is a secret. @@ -701,6 +703,18 @@ folder: the computer is ready once every app's server answers and each `boot` ap no launch, says it's ready; a `demand` app counts once something launches it. What the server does in the session (`agentenv_protocol.computer.session`) is Linux's. +The server serves the app's data plane and its skill. With `paths` alone, `data/add` writes the files +it's given under them and refuses any other path, or one a link leads out of them; `data/get` +returns the files there, a file tree; and `data/reset` puts them back as the server found them when +it first started on this boot, which is the app as installed. Hooks serve what they mark instead. +A hook's `data/add` parts come as they were sent, inline or as a read grant, and +`part_bytes(part)` returns a part's bytes either way. Either way the base class takes data calls and +launches one at a time, runs a load and a reset with the app stopped and relaunches it after, +follows a failed load with a reset, and fails a call that runs past `data_timeout_s` (300 s); a sync +hook runs in a thread, which the reset and relaunch after it wait for. The driver fronts its apps: +its card lists each one's card as a child environment, rooted at `/svc//`, where it forwards +requests to the app's server. + `MailApp.declaration()` returns what the class declares, an `AppDeclaration`, which a `computer_app` env stores, and `python -m agentenv_protocol.computer._declare ` prints it as JSON, or a driver's `DriverDeclaration`, without the class's own dependencies installed. diff --git a/packages/agentenv-protocol/src/agentenv_protocol/agent_env_environment.py b/packages/agentenv-protocol/src/agentenv_protocol/agent_env_environment.py index f2b3b4676..c88634844 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/agent_env_environment.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/agent_env_environment.py @@ -14,7 +14,8 @@ import types import uuid from abc import ABC, abstractmethod -from typing import Annotated, Any, Callable, ClassVar, Literal, NamedTuple, Union, get_args, get_origin, get_type_hints +from typing import (Annotated, Any, Callable, ClassVar, Literal, NamedTuple, Optional, Union, get_args, get_origin, + get_type_hints) from pydantic import TypeAdapter, ValidationError from pydantic.fields import FieldInfo @@ -376,7 +377,10 @@ def _card_for(self, app: Any) -> EnvironmentCard: return self.environment_card async def _serve_card(self, request: Request) -> Response: - return _json_response(self.environment_card.model_dump()) + card = self.environment_card + if isinstance(self.handler, AgentEnvEnvironment) and (children := await self.handler.children_environments()): + card = card.model_copy(update={"children_environments": children}) + return _json_response(card.model_dump()) async def update_tool(self, name: str, *, description: str) -> None: """Change a served tool's description after the mount: on the card and in tools/list alike.""" @@ -662,6 +666,11 @@ def declared_extensions(self) -> list[EnvironmentExtension]: """Declaration-only extensions to advertise, like ``intake/v1``: a promise with params and no handler.""" return [] + async def children_environments(self) -> Optional[list[EnvironmentCard]]: + """The environments this one fronts, as each serves its card with its paths under this one's, listed on the card + each time it's served; None for none.""" + return None + def served_extensions(self, extensions: list[EnvironmentExtension]) -> list[EnvironmentExtension]: """The ``@extension`` handlers' extensions to serve, chosen at mount: one left out is not served, and of one served with ``op``, only the methods its returned descriptor still lists.""" diff --git a/packages/agentenv-protocol/src/agentenv_protocol/client.py b/packages/agentenv-protocol/src/agentenv_protocol/client.py index 0d0bfe648..33f6f2375 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/client.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/client.py @@ -147,6 +147,12 @@ def intake_declaration(card: dict) -> dict | None: return params if params else None +def operations(card: dict) -> list[str] | None: + """The data-plane methods a card declares it serves (``capabilities.operations``), or None when it declares none, + as an older card doesn't.""" + return (card.get("capabilities") or {}).get("operations") + + def find_tool(card: dict, name: str) -> dict | None: """Return the MCP tool advertised under `name`, or None. diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/__init__.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/__init__.py index 5ffb4ed01..586e962cc 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/__init__.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/__init__.py @@ -67,6 +67,7 @@ type_text, write_files, ) +from .dataplane import part_bytes from .keys import PRINTABLE_KEYS from .manifest import ManifestSegment, RecordingManifest from .platform import Platform @@ -107,7 +108,7 @@ def __getattr__(name: str) -> Any: "ComputerAppEnvironment", "computer_app", "install", "linux", "install_step", "ready_check", "Linux", "Service", "Launch", "Ready", "Packages", "AptSource", "Download", "RunScript", "AppDeclaration", "AppOs", - "APP_SETTINGS_ENV", + "APP_SETTINGS_ENV", "part_bytes", "Display", "Image", "Effect", "ExecResult", "ReadyCheck", "Platform", "ComputerSettings", "SETTINGS_ENV", "PRINTABLE_KEYS", "display", "set_display", "capture", "click", "drag", "scroll", "type_text", "press", diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py index 25308532f..6772fc087 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/app.py @@ -1,24 +1,32 @@ """ComputerAppEnvironment: software on a computer, an AgentEnvEnvironment whose class also declares what it is for a computer: the platforms it supports and the permissions it needs (``@computer_app``); for each OS, how it installs and launches there and what differs there (``@linux``, or ``@install`` for any OS type); its install steps and ready checks, -each on every OS or on those it names; its skill, its typed ``Settings`` and its ``BuildSettings``. A ``computer_app`` -env's put reads that declaration from the class; its data hooks are the base class's. Its name is its env's -``environment_name``. On the computer the class runs as the app's own server (``serve_app``), which launches the app, -adds its launcher and says whether it's ready (``computer-app/v1``), as declared for the OS it runs on.""" +each on every OS or on those it names; its skill, its typed ``Settings`` and its ``BuildSettings``; and its data, the +``paths`` it keeps it in under the session user's home or its own ``@add_data``, ``@get_data`` and ``@reset_data`` +hooks. A ``computer_app`` env's put reads that declaration from the class. Its name is its env's ``environment_name``. +On the computer the class runs as the app's own server (``serve_app``), which launches the app, adds its launcher, says +whether it's ready (``computer-app/v1``), serves its skill, and serves its data plane, stopping the app around a load +and a reset, as declared for the OS it runs on.""" from __future__ import annotations import functools import inspect +import itertools import json import os from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Annotated, Any, Callable, ClassVar, Literal, Optional, Union -from pydantic import AfterValidator, BaseModel, ConfigDict, Field, create_model, field_validator, model_validator +from pydantic import (AfterValidator, BaseModel, ConfigDict, Field, TypeAdapter, ValidationError, create_model, + field_validator, model_validator) -from ..agent_env_environment import _EXT_ATTR, _OP_ATTR, AgentEnvEnvironment, extension -from ..types import METHOD_ADD, METHOD_GET, METHOD_RESET, SKILL_EXTENSION_URI +from ..agent_env_environment import (_EXT_ATTR, _OP_ATTR, OP_ADD_DATA, OP_GET_DATA, OP_RESET_DATA, AgentEnvEnvironment, + add_data, agent_skill, extension, get_data, reset_data) +from ..transfers import RelativePath, WriteNamespaceGrant +from ..types import (METHOD_ADD, METHOD_GET, METHOD_RESET, SKILL_EXTENSION_URI, EnvironmentExtension, IntakeDeclaration, + IntakeFormat, intake_extension) +from .app_data import AppDataPlane from .app_runtime import AppRuntime from .errors import InvalidArgument from .platform import detect @@ -27,7 +35,8 @@ APP_SETTINGS_ENV = "AGENTENV_APP_SETTINGS" _INSTALL_STEP_ATTR = "_aee_install_step" _READY_CHECK_ATTR = "_aee_ready_check" -_DATA_METHODS = {"reset_data": METHOD_RESET, "add_data": METHOD_ADD, "get_data": METHOD_GET} +_DATA_METHODS = {OP_RESET_DATA: METHOD_RESET, OP_ADD_DATA: METHOD_ADD, OP_GET_DATA: METHOD_GET} +_PATHS = TypeAdapter(list[RelativePath]) #: What a launch's window can be matched by, as the session lists each window. WINDOW_KEYS = frozenset({"class", "instance", "title"}) #: The framework extension an app's own server serves: its launch as params, and its ``launch`` and ``ready`` methods. @@ -207,13 +216,15 @@ class AppReadyCheck(BaseModel): class AppOs(BaseModel): """What an app declares for one OS type: ``install``, how that OS installs it, as the OS type's install model gives - it; its ``launch`` there; and its ``skill``, ``settings`` and ``build_settings`` there when they differ from the - class's, the schemas with the class's fields added.""" + it; its ``launch`` there; the ``paths`` it keeps its data in there, relative to the session user's home; and its + ``skill``, ``settings`` and ``build_settings`` there when they differ from the class's, the schemas with the class's + fields added.""" model_config = ConfigDict(extra="forbid") install: dict[str, Any] launch: Optional[Launch] = None + paths: Optional[list[str]] = None skill: Optional[AppSkill] = None settings: Optional[dict[str, Any]] = None build_settings: Optional[dict[str, Any]] = None @@ -258,6 +269,7 @@ class _OsDeclared: install: BaseModel launch: Optional[Launch] + paths: tuple[str, ...] skill: Optional[str] settings: Optional[type[BaseModel]] build_settings: Optional[type[BaseModel]] @@ -274,17 +286,26 @@ def deco(cls: type) -> type: return deco -def install(os_type: str, spec: BaseModel, *, launch: Optional[Launch] = None, skill: Optional[str] = None, - settings: Optional[type[BaseModel]] = None, +def install(os_type: str, spec: BaseModel, *, launch: Optional[Launch] = None, paths: Optional[list[str]] = None, + skill: Optional[str] = None, settings: Optional[type[BaseModel]] = None, build_settings: Optional[type[BaseModel]] = None) -> Callable: """Declare how a computer of ``os_type`` installs the app, ``spec`` being that OS type's install model, and what is - the app's own there: its ``launch``; its ``skill``, a file of its folder, instead of the class's; and ``settings`` - and ``build_settings``, models whose fields add to the class's ``Settings`` and ``BuildSettings`` there. ``@linux`` - is this for Linux; an OS type a plugin adds is declared with it.""" + the app's own there: its ``launch``; the ``paths`` it keeps its data in, relative to the session user's home, which + serve its data plane when it has no data hooks; its ``skill``, a file of its folder, instead of the class's; and + ``settings`` and ``build_settings``, models whose fields add to the class's ``Settings`` and ``BuildSettings`` there. + ``@linux`` is this for Linux; an OS type a plugin adds is declared with it.""" + try: + owned = tuple(_PATHS.validate_python(list(paths or []))) + except ValidationError: + raise ValueError(f"paths are normalized paths relative to the session user's home, not {paths!r}") from None + if overlapping := [f"{a!r} and {b!r}" for a, b in itertools.combinations(owned, 2) + if PurePosixPath(a).is_relative_to(b) or PurePosixPath(b).is_relative_to(a)]: + raise ValueError(f"an app's paths each hold files of their own, but {'; '.join(overlapping)} overlap") + def deco(cls: type) -> type: if not (isinstance(cls, type) and issubclass(cls, ComputerAppEnvironment)): raise TypeError("@install decorates a ComputerAppEnvironment subclass") - cls._installs = {**cls._installs, os_type: _OsDeclared(spec, launch, skill, settings, build_settings)} + cls._installs = {**cls._installs, os_type: _OsDeclared(spec, launch, owned, skill, settings, build_settings)} return cls return deco @@ -292,12 +313,13 @@ def deco(cls: type) -> type: def linux(*, files: str = ".", packages: Optional[Packages] = None, python: Optional[str] = None, sources: Optional[list[AptSource]] = None, downloads: Optional[list[Download]] = None, run: Optional[list[RunScript]] = None, services: Optional[list[Service]] = None, - ports: Optional[list[int]] = None, launch: Optional[Launch] = None, skill: Optional[str] = None, - settings: Optional[type[BaseModel]] = None, build_settings: Optional[type[BaseModel]] = None) -> Callable: + ports: Optional[list[int]] = None, launch: Optional[Launch] = None, paths: Optional[list[str]] = None, + skill: Optional[str] = None, settings: Optional[type[BaseModel]] = None, + build_settings: Optional[type[BaseModel]] = None) -> Callable: """``@install`` for Linux: how a Linux computer installs the app, ``Linux``'s fields, and what is its own there.""" return install("linux", Linux(files=files, packages=packages, python=python, sources=sources, downloads=downloads, run=run, services=services or [], ports=ports), - launch=launch, skill=skill, settings=settings, build_settings=build_settings) + launch=launch, paths=paths, skill=skill, settings=settings, build_settings=build_settings) def install_step(scope: Literal["system", "session"] = "system", *, @@ -328,15 +350,41 @@ def _named(os_types: Union[str, list[str], None]) -> Optional[list[str]]: class ComputerAppEnvironment(AgentEnvEnvironment): """Subclass it, declare it with ``@computer_app`` and an OS decorator for each OS (``@linux``), and set ``skill`` as a class attribute; a deploy's settings arrive as ``self.settings``, those its env fixes at build as - ``self.build_settings``, and its launch as ``self.launch``, each as declared for the OS it runs on.""" + ``self.build_settings``, and its launch and paths as ``self.launch`` and ``self.paths``, each as declared for the OS + it runs on. + + Its data plane is its ``@add_data``, ``@get_data`` and ``@reset_data`` hooks, or, with none, defaults on its + ``paths``, which the base class serves around its contract (``app_data.AppDataPlane``): one call at a time, a load + and a reset with the app stopped and relaunched after, a load that fails followed by a reset, and each call within + ``data_timeout_s``.""" supports: ClassVar[list] = [] permissions: ClassVar[list[str]] = [] _installs: ClassVar[dict[str, _OsDeclared]] = {} + #: Its data hooks' method names, by the operation each serves. + _data_hooks: ClassVar[dict[str, str]] = {} _computer_app: Optional[AppRuntime] = None + _computer_app_data: Optional[AppDataPlane] = None launch: Optional[Launch] = None + paths: list[str] = [] #: A file of the app's folder holding its skill for agents, when no ``@agent_skill`` method writes it. skill: ClassVar[Optional[str]] = None + #: How long one of its data calls may take, a hook or its paths' default, before it fails. + data_timeout_s: ClassVar[float] = 300.0 + + def __init_subclass__(cls, **kwargs: Any) -> None: + """Take the class's data hooks over, for the base class's handlers to serve around its contract; and serve its + skill file unless it writes its skill itself.""" + super().__init_subclass__(**kwargs) + hooks = dict(cls._data_hooks) + for name, member in vars(cls).items(): + if (op := getattr(member, _OP_ATTR, None)) in _DATA_METHODS: + hooks[op] = name + delattr(member, _OP_ATTR) + cls._data_hooks = hooks + if any(getattr(getattr(member, _EXT_ATTR, None), "uri", None) == SKILL_EXTENSION_URI + for member in vars(cls).values()): + cls._computer_app_skill = None class Settings(BaseModel): """The app's settings: an app declares its own; a ``SecretStr`` field is a secret.""" @@ -350,12 +398,17 @@ def __init__(self) -> None: values = json.loads(os.environ.get(APP_SETTINGS_ENV) or "{}").get(os.environ.get("ENVIRONMENT_NAME"), {}) self.settings = cls.settings_on(on).model_validate(values) self.build_settings = _built_settings(cls, on) - self.launch = declared.launch if (declared := cls._installs.get(on)) is not None else None + declared = cls._installs.get(on) + self.launch = declared.launch if declared is not None else None + self.paths = list(declared.paths) if declared is not None else [] ready = self.launch.ready if self.launch is not None else None self._computer_app = AppRuntime( self.launch, {fn.__name__: getattr(self, fn.__name__) for fn in _methods(cls) if hasattr(fn, _READY_CHECK_ATTR) and _runs_on(getattr(fn, _READY_CHECK_ATTR), on)}, ready.timeout_s if ready is not None else DEFAULT_TIMEOUT_S) + self._computer_app_data = AppDataPlane( + self._computer_app, {op: getattr(self, name) for op, name in cls._data_hooks.items()}, self.paths, + cls.data_timeout_s) @classmethod def settings_on(cls, os_type: Optional[str]) -> type[BaseModel]: @@ -375,7 +428,7 @@ def build_settings_on(cls, os_type: Optional[str]) -> type[BaseModel]: async def _computer_app_launch(self) -> dict: if self.launch is None: raise InvalidArgument(f"{type(self).__name__} declares no launch") - return {"pid": await self._computer_app_runtime().launch_app()} + return {"pid": await self._computer_app_data_plane().launch()} @extension(COMPUTER_APP, op="ready", method="GET") async def _computer_app_ready(self) -> dict: @@ -386,11 +439,61 @@ def _computer_app_runtime(self) -> AppRuntime: raise TypeError(f"{type(self).__qualname__}.__init__ must call super().__init__()") return self._computer_app + def _computer_app_data_plane(self) -> AppDataPlane: + self._computer_app_runtime() + return self._computer_app_data + def extension_params(self, uri: str) -> dict: if uri == COMPUTER_APP: - return {"launch": self.launch.model_dump(mode="json", exclude_none=True) if self.launch else None} + return {"launch": self.launch.model_dump(mode="json", exclude_none=True) if self.launch else None, + "paths": self.paths} return super().extension_params(uri) + def served_operations(self, operations: list[str]) -> list[str]: + """Its hooks' operations, or with none, all three when it has paths.""" + served = ({_DATA_METHODS[op] for op in self._data_hooks} if self._data_hooks + else set(_DATA_METHODS.values()) if self.paths else set()) + return [op for op in operations if op in served] + + def served_extensions(self, extensions: list[EnvironmentExtension]) -> list[EnvironmentExtension]: + """Without its skill file on this OS, no skill, unless it writes one itself.""" + own_skill = getattr(type(self), "_computer_app_skill", None) is not None + if own_skill and self._computer_app_skill_file() is None: + extensions = [extension for extension in extensions if extension.uri != SKILL_EXTENSION_URI] + return super().served_extensions(extensions) + + def declared_extensions(self) -> list[EnvironmentExtension]: + """With its paths serving its data, a file tree: ``data/get`` returns a part per file.""" + declared = super().declared_extensions() + if self.paths and not self._data_hooks: + tree = IntakeDeclaration(get=[IntakeFormat(part="file", format="file-tree")], + add=[IntakeFormat(part="file", format="file-tree", load="additive")]) + declared.append(intake_extension(tree, f"data/add writes files under the app's paths, " + f"{', '.join(self.paths)}; data/get returns them; data/reset puts " + "them back as installed.")) + return declared + + @add_data + async def _computer_app_add_data(self, parts: list) -> None: + await self._computer_app_data_plane().call(OP_ADD_DATA, parts) + + @get_data + async def _computer_app_get_data(self, write_namespace: Optional[WriteNamespaceGrant] = None) -> list: + return await self._computer_app_data_plane().call(OP_GET_DATA, write_namespace) + + @reset_data + async def _computer_app_reset_data(self) -> None: + await self._computer_app_data_plane().call(OP_RESET_DATA) + + @agent_skill + def _computer_app_skill(self) -> tuple[str, str]: + return _skill_parts(Path(self._computer_app_skill_file()).read_text(), os.environ.get("ENVIRONMENT_NAME", "")) + + def _computer_app_skill_file(self) -> Optional[str]: + """The file of the app's folder holding its skill on this OS, or the class's.""" + declared = type(self)._installs.get(detect().os) + return declared.skill if declared is not None and declared.skill is not None else type(self).skill + @classmethod def declaration(cls) -> AppDeclaration: """What the class declares. Raises ValueError for a skill declared twice, a build setting that's a secret, or @@ -404,7 +507,8 @@ def declaration(cls) -> AppDeclaration: "it in Settings") methods = _methods(cls) skill_methods = [fn.__name__ for fn in methods - if getattr(getattr(fn, _EXT_ATTR, None), "uri", None) == SKILL_EXTENSION_URI] + if getattr(getattr(fn, _EXT_ATTR, None), "uri", None) == SKILL_EXTENSION_URI + and not fn.__name__.startswith("_computer_app_")] if cls.skill is not None and skill_methods: raise ValueError(f"{cls.__name__} declares its skill twice: skill = {cls.skill!r} and @agent_skill " f"{skill_methods[0]}") @@ -425,6 +529,7 @@ def declaration(cls) -> AppDeclaration: supports=cls.supports, permissions=cls.permissions, os={os_type: AppOs( install=declared.install.model_dump(mode="json", exclude_none=True), launch=declared.launch, + paths=list(declared.paths) or None, skill=AppSkill(file=declared.skill) if declared.skill is not None else None, settings=cls.settings_on(os_type).model_json_schema() if declared.settings is not None else None, build_settings=(cls.build_settings_on(os_type).model_json_schema() @@ -433,11 +538,23 @@ def declaration(cls) -> AppDeclaration: settings=cls.Settings.model_json_schema(), build_settings=build if build.get("properties") else None, skill=AppSkill(file=cls.skill) if cls.skill is not None else AppSkill(method=skill_methods[0]) if skill_methods else None, - data=sorted({_DATA_METHODS[op] for fn in methods if (op := getattr(fn, _OP_ATTR, None)) in _DATA_METHODS}), + data=sorted(_DATA_METHODS[op] for op in cls._data_hooks), install_steps=steps, ready_checks=checks, ) +def _skill_parts(text: str, name: str) -> tuple[str, str]: + """A skill file's description and body: from its frontmatter's ``description`` when it has one.""" + if text.startswith("---\n") and (end := text.find("\n---", 4)) != -1: + front, body = text[4:end], text[end + 4:].lstrip("\n") + description = next((line.partition(":")[2].strip().strip("'\"") for line in front.splitlines() + if line.startswith("description:")), "") + if description: + return description, body + text = body + return f"How to use {name} on this computer.", text + + def _secret_at(schema: dict, path: str = "") -> Optional[str]: """Where in ``schema``, a JSON schema, a secret is, a ``SecretStr``'s ``writeOnly``: in it or any of its subschemas, a field's, a list's items or a nested model's definition; None when it holds none. A default or a field's name diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/app_data.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_data.py new file mode 100644 index 000000000..3505055ef --- /dev/null +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_data.py @@ -0,0 +1,217 @@ +"""An app's data plane as its own server serves it (``AppDataPlane``), by the app's data hooks or, with none, the +defaults on the paths it keeps its data in, under the session user's home: a load writes the files it's given under +them, an export returns the files there, and a reset puts back the copy its server kept when it first started on this +boot, which is the app as installed, since every boot starts from the computer's image.""" +from __future__ import annotations + +import asyncio +import functools +import inspect +import logging +import shutil +from collections.abc import Callable +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Any, Optional + +from ..agent_env_environment import OP_ADD_DATA, OP_GET_DATA, OP_RESET_DATA +from ..transfers import WriteNamespaceGrant +from ..types import FilePart, uploaded_object_path +from .dataplane import file_parts, part_bytes + +if TYPE_CHECKING: + from .app_runtime import AppRuntime + +logger = logging.getLogger(__name__) + +#: Where an app's server keeps its paths as installed, in the app's folder. +INITIAL = Path(".agentenv/initial") + + +class AppDataPlane: + """An app's data plane, served around its contract: one data call or launch at a time; a load and a reset stop the + running app first and relaunch it after, and a load that fails is followed by a reset. A call fails once it runs + past ``timeout_s`` and is cancelled: what awaits stops at once, but a thread can't be stopped, so a sync hook's or + a default's thread is waited for, as are the reset and relaunch after it and the next call.""" + + def __init__(self, runtime: AppRuntime, hooks: dict[str, Callable], paths: list[str], timeout_s: float) -> None: + self.runtime = runtime + self.hooks = hooks + self.paths = paths + self.timeout_s = timeout_s + self.turn = asyncio.Lock() + self.changing: set[asyncio.Task] = set() + + def keep_initial(self) -> None: + """Keep the app's paths as installed, before anything changes them, when they serve its data.""" + if self.paths and not self.hooks: + keep_initial(self.paths) + + async def call(self, op: str, *args: Any) -> Any: + """``op``'s answer, by its hook or its paths' default, once the data call before has ended. The change it + makes, with the reset and relaunch after it, holds the turn until it ends, past an answer given at its + timeout.""" + served = self._served(op) + await self._wait_turn(op) + answer = asyncio.get_running_loop().create_future() + change = asyncio.ensure_future(self._change(op, served, args, answer)) + self.changing.add(change) + change.add_done_callback(self.changing.discard) + change.add_done_callback(lambda _: self.turn.release()) + return await answer + + async def launch(self) -> int: + """Launch the app once the data call before has ended; its pid.""" + await self._wait_turn("launch") + try: + return await self.runtime.launch_app() + finally: + self.turn.release() + + async def _change(self, op: str, served: Callable, args: tuple, answer: asyncio.Future) -> None: + """``op`` between the app's stop and its relaunch, answered on ``answer``: as it ends, or once it runs past + ``timeout_s``.""" + try: + running = op != OP_GET_DATA and await self.runtime.stop_app() + try: + result = await self._within(op, served, args, answer) + except Exception: + if op == OP_ADD_DATA: + await self._undo(answer) + raise + finally: + if running: + await self.runtime.launch_app() + except Exception as e: + if not answer.done(): + answer.set_exception(e) + else: + answer.set_result(result) + + async def _undo(self, answer: asyncio.Future) -> None: + """Reset the app's data after a load that failed.""" + if (reset := self._served(OP_RESET_DATA)) is None: + return + try: + await self._within(OP_RESET_DATA, reset, (), answer) + except Exception as e: + logger.warning(f"the reset after a failed data/add failed too: {type(e).__name__}: {e}") + + async def _wait_turn(self, op: str) -> None: + """Wait, within ``timeout_s``, for the data call before to end, its reset and relaunch included.""" + try: + await asyncio.wait_for(self.turn.acquire(), self.timeout_s) + except TimeoutError: + raise TimeoutError(f"{op} waited past the app's data_timeout_s, {self.timeout_s:g} s, for the data call " + "before it to end") from None + + def _served(self, op: str) -> Optional[Callable]: + """What serves ``op``: the app's hook, or its paths' default; None when nothing does.""" + if self.hooks: + hook = self.hooks.get(op) + if hook is not None and op == OP_GET_DATA and "write_namespace" not in inspect.signature(hook).parameters: + return lambda write_namespace=None: hook() + return hook + if not self.paths: + return None + return {OP_ADD_DATA: functools.partial(add, paths=self.paths), + OP_GET_DATA: functools.partial(get, self.paths), + OP_RESET_DATA: functools.partial(reset, self.paths)}[op] + + async def _within(self, op: str, served: Callable, args: tuple, answer: asyncio.Future) -> Any: + """``served``'s result within ``timeout_s``; past it, ``answer`` fails at once and the call is cancelled, and + this raises the same once it has ended: at once for what awaits, after any thread it runs.""" + work = asyncio.ensure_future(_called(served, *args)) + try: + return await asyncio.wait_for(asyncio.shield(work), self.timeout_s) + except TimeoutError: + late = TimeoutError(f"{op} ran past the app's data_timeout_s, {self.timeout_s:g} s") + if not answer.done(): + answer.set_exception(late) + work.cancel() + await asyncio.wait([work]) + if not work.cancelled(): + work.exception() + raise late from None + + +def keep_initial(paths: list[str]) -> None: + """Copy what's under each of ``paths`` of the home into ``INITIAL``, unless this boot kept a copy already.""" + if INITIAL.exists(): + return + staging = INITIAL.with_name(f"{INITIAL.name}.partial") + shutil.rmtree(staging, ignore_errors=True) + staging.mkdir(parents=True) + for path in paths: + _copy(Path.home() / path, staging / path) + staging.rename(INITIAL) + + +async def add(parts: list, paths: list[str]) -> None: + """Write each file part at the path it names, which must be under one of ``paths``, through no link that leads + out of them; nothing is written when one isn't.""" + files = {} + for part in parts: + path = uploaded_object_path(part) if isinstance(part, FilePart) else None + if path is None: + raise ValueError(f"data/add on this app takes file parts naming their path, got a {part.kind} part") + if not any(PurePosixPath(path).is_relative_to(owned) for owned in paths): + raise ValueError(f"{path!r} is under none of this app's paths: {', '.join(paths)}") + files[path] = await part_bytes(part) + await in_thread(_write, Path.home(), files, paths) + + +async def get(paths: list[str], write_namespace: Optional[WriteNamespaceGrant]) -> list: + """A part for each file under ``paths``.""" + return await file_parts(Path.home(), paths, write_namespace) + + +async def reset(paths: list[str]) -> None: + """Put each of ``paths`` back as ``keep_initial`` kept it, or remove it where it wasn't there then.""" + await in_thread(_restore, Path.home(), paths) + + +async def _called(fn: Callable, *args: Any) -> Any: + """``fn``'s result, a sync one's from a thread, so it can't hold up the server.""" + if inspect.iscoroutinefunction(fn): + return await fn(*args) + result = await in_thread(fn, *args) + return await result if inspect.isawaitable(result) else result + + +async def in_thread(fn: Callable, *args: Any) -> Any: + """``fn``'s result from a thread, which a cancelled call still waits for to the end, since a thread can't be + stopped.""" + work = asyncio.ensure_future(asyncio.to_thread(fn, *args)) + try: + return await asyncio.shield(work) + except asyncio.CancelledError: + await asyncio.wait([work]) + raise + + +def _write(home: Path, files: dict[str, bytes], paths: list[str]) -> None: + owned = [(home / path).resolve() for path in paths] + if linked := [path for path in files if not any((home / path).resolve().is_relative_to(o) for o in owned)]: + raise ValueError(f"{', '.join(map(repr, linked))} lead out of this app's paths through a link") + for path, data in files.items(): + target = home / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(data) + + +def _restore(home: Path, paths: list[str]) -> None: + for path in paths: + target = home / path + if target.is_dir() and not target.is_symlink(): + shutil.rmtree(target) + else: + target.unlink(missing_ok=True) + _copy(INITIAL / path, target) + + +def _copy(source: Path, target: Path) -> None: + if source.is_dir() and not source.is_symlink(): + shutil.copytree(source, target, symlinks=True) + elif source.exists() or source.is_symlink(): + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, target, follow_symlinks=False) diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py index d5f27370f..89c5bcd8f 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/app_runtime.py @@ -1,6 +1,6 @@ """A computer app's own server at work: it launches the app into the desktop session, marking each launch for the rest -of the boot, and says whether the app is ready, its checks run at once, each within ``CHECK_TIMEOUT_S`` -(``AppReadiness``). ``ComputerAppEnvironment`` keeps one as ``self._computer_app``.""" +of the boot, stops the app it launched, and says whether the app is ready, its checks run at once, each within +``CHECK_TIMEOUT_S`` (``AppReadiness``). ``ComputerAppEnvironment`` keeps one as ``self._computer_app``.""" from __future__ import annotations import asyncio @@ -23,6 +23,11 @@ LAUNCHED = Path("/run/agentenv/apps") #: How long one readiness check may take; an app's checks run at once. CHECK_TIMEOUT_S = 10.0 +#: Where an app's server keeps the pid of the app it launched last, in the app's folder, so a server that restarts can +#: still stop it. +PID = Path(".agentenv/app.pid") +#: How long a stopped app may take to exit before it's killed. +STOP_TIMEOUT_S = 10.0 class AppReadiness(BaseModel): @@ -68,10 +73,24 @@ async def launch_app(self) -> int: """Start the app from its folder, marked as launched; its pid.""" pid = await session.launch(self.launch.command, Path.cwd(), {"ENVIRONMENT_NAME": os.environ["ENVIRONMENT_NAME"]}) self.launch_failed = None + PID.parent.mkdir(parents=True, exist_ok=True) + PID.write_text(str(pid)) _mark().parent.mkdir(parents=True, exist_ok=True) _mark().touch() return pid + async def stop_app(self) -> bool: + """Stop the app this server launched last, with everything it started; whether any of it was running.""" + try: + pid = int(PID.read_text()) + except (FileNotFoundError, ValueError): + return False + PID.unlink() + if not session.alive(pid): + return False + await session.stop(pid, STOP_TIMEOUT_S) + return True + async def readiness(self) -> AppReadiness: ready = self.launch.ready if self.launch is not None else None checks: list[tuple[str, Awaitable[Optional[str]]]] = [] diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/apps.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/apps.py index 56c41b567..9de569ed5 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/apps.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/apps.py @@ -1,15 +1,20 @@ -"""The apps composed onto a computer, as its driver finds them, and whether each is ready. Each app's own server -answers on its folder's socket (``serve_app``): its card says how the app launches, and its ``computer-app/v1`` -``ready`` whether it's ready. Every app's server must answer before the computer is ready; an app that starts with the -computer must also be ready, and a ``demand`` app is checked once something launches it.""" +"""The apps composed onto a computer, as its driver finds them, whether each is ready, and how the driver fronts them. +Each app's own server answers on its folder's socket (``serve_app``): its card says how the app launches and which +paths it owns, and its ``computer-app/v1`` ``ready`` whether it's ready. Every app's server must answer before the +computer is ready; an app that starts with the computer must also be ready, and a ``demand`` app is checked once +something launches it. The driver lists each app as a child environment named for its folder, its card's paths under +``/svc//``, and forwards each request there to the app's server.""" from __future__ import annotations +import asyncio +import copy from pathlib import Path +from typing import Optional import httpx from .. import client -from ..types import WELL_KNOWN_PATH +from ..types import RPC_PATH, WELL_KNOWN_PATH, EnvironmentCard from .app import COMPUTER_APP, FAILURE, SOCKET from .app_runtime import CHECK_TIMEOUT_S, AppReadiness from .operations import ReadyCheck @@ -46,11 +51,68 @@ async def check(folder: Path) -> ReadyCheck: return ReadyCheck(f"app {name}", False, f"{when}: {'; '.join(answer.problems)}") +async def children(folders: list[Path]) -> list[EnvironmentCard]: + """Each app in ``folders`` as a child environment of its computer: its server's card, rooted at ``/svc//``, + or, while its server isn't answering, its name and route alone.""" + cards = await asyncio.gather(*(_card(folder) for folder in folders)) + return [EnvironmentCard.model_validate(child_card(folder.name, card)) if card is not None + else EnvironmentCard(name=folder.name, url=f"/svc/{folder.name}{RPC_PATH}") + for folder, card in zip(folders, cards)] + + +def child_card(name: str, card: dict) -> dict: + """``card`` with every path it names, its ``url`` and each extension's and method's endpoint, under + ``/svc/``, so it resolves against the computer's address; and none of its interfaces, which agents reach + through the computer.""" + card = copy.deepcopy(card) + + def rooted(path: object) -> object: + return f"/svc/{name}{path}" if isinstance(path, str) and path.startswith("/") else path + + card["url"] = rooted(card.get("url") or RPC_PATH) + card["additionalInterfaces"], card["children_environments"] = [], None + for extension in (card.get("capabilities") or {}).get("extensions") or []: + params = extension.get("params") + if not isinstance(params, dict): + continue + if "endpoint" in params: + params["endpoint"] = rooted(params["endpoint"]) + for method in (params.get("methods") or {}).values(): + if isinstance(method, dict) and "endpoint" in method: + method["endpoint"] = rooted(method["endpoint"]) + return card + + +async def owned_paths(folders: list[Path]) -> list[str]: + """The paths under the session user's home that the apps in ``folders`` keep their data in, as their cards say.""" + cards = await asyncio.gather(*(_card(folder) for folder in folders)) + return [path for card in cards if card is not None + for path in client.extension_params(card, COMPUTER_APP).get("paths") or []] + + +async def forward(folder: Path, method: str, path: str, query: str, body: bytes, + content_type: Optional[str]) -> httpx.Response: + """The app's server's answer to the request, sent on as it came; it takes as long as the app does.""" + async with httpx.AsyncClient(transport=httpx.AsyncHTTPTransport(uds=str(folder / SOCKET)), base_url="http://app", + timeout=None) as http: + return await http.request(method, f"/{path}" + (f"?{query}" if query else ""), content=body, + headers={"content-type": content_type} if content_type else None) + + def endpoint(card: dict, method: str) -> str: """The path of a ``computer-app/v1`` method, as the app's card advertises it.""" return client.find_extension_method(card, COMPUTER_APP, method)["endpoint"] +async def _card(folder: Path) -> Optional[dict]: + """The card the app in ``folder`` serves, or None while its server isn't answering.""" + try: + async with _client(folder / SOCKET) as http: + return (await http.get(WELL_KNOWN_PATH)).raise_for_status().json() + except (httpx.HTTPError, OSError, ValueError): + return None + + def _unanswered(folder: Path) -> str: try: return f"its server failed to start: {(folder / FAILURE).read_text()}" diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py index a13bff5e9..16f139d89 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py @@ -87,8 +87,8 @@ def __init__(self, driver: Driver, session: ClientSession, ready_timeout_s: floa async def basic(self) -> None: for check, section in (("P0", self._ready), ("P1", self._card), ("P2", self._frame), ("P5", self._refusals), ("X", self._display), ("X", self._spaces), ("X", self._keyboard_settings), - ("X", self._exec), ("X", self._data_get), ("X", self._tool_access), - ("X", self._pointer), ("X", self._skill)): + ("X", self._exec), ("X", self._data_get), ("X", self._app_exports), + ("X", self._tool_access), ("X", self._pointer), ("X", self._skill)): await self._guarded(check, section) self._grants() @@ -410,6 +410,18 @@ async def get() -> str | None: await self.check("X", "data/get answers its parts", get, *DATA_PLANE) + async def _app_exports(self) -> None: + for app in self.driver.card.get("children_environments") or []: + if "data/get" not in (protocol.operations(app) or []): + continue + + async def export(app: dict = app) -> str | None: + got = await self.driver.rpc("data/get", card=app) + return None if isinstance((got.get("result") or {}).get("parts"), list) else f"it answered {got}" + + await self.check("X", f"app {app.get('name')}'s data/get answers its parts through the driver", export, + *DATA_PLANE) + async def _tool_access(self) -> None: if not (self.driver.declares(DISABLE_TOOL_EXTENSION_URI) and self.driver.declares(ENABLE_TOOL_EXTENSION_URI)): return diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/client.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/client.py index 4a71ebd8b..a941e4f52 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/client.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/client.py @@ -61,9 +61,10 @@ async def invoke(self, uri: str, method: str, params: dict | None = None) -> Any return await protocol.invoke_extension(self.url, self.card, uri, params or {}, method=method, timeout=int(self.timeout_s)) - async def rpc(self, method: str, params: dict | None = None) -> dict: + async def rpc(self, method: str, params: dict | None = None, card: dict | None = None) -> dict: + """``method``'s answer from the data plane of ``card``, the driver's own or a child's it fronts.""" async with httpx.AsyncClient(timeout=self.timeout_s) as http: - response = await http.post(self.url + (self.card.get("url") or "/agentenv"), + response = await http.post(self.url + ((card or self.card).get("url") or "/agentenv"), json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params or {}}) return response.json() diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py index ba9f8dcb8..01f704429 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/dataplane.py @@ -1,13 +1,14 @@ """The data plane on a computer: ``data/add`` writes files under the session user's home, ``data/get`` returns the -files under the paths its caller names, or under ``DEFAULT_DATA_PATHS`` when it names none. Each file part names its -path, relative to the home, under the ``data-objects/v1`` marker.""" +files under the paths its caller names, or under ``DEFAULT_DATA_PATHS`` when it names none, leaving out those under a +path one of the computer's apps keeps its data in. Each file part names its path, relative to the home, under the +``data-objects/v1`` marker.""" from __future__ import annotations import base64 import binascii import mimetypes import tempfile -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Optional import httpx @@ -47,7 +48,7 @@ async def _computer_data_add(self, parts: list) -> None: if path is None: raise ValueError(f"data/add on a computer takes file parts naming their path under " f"{DATA_OBJECTS_EXTENSION_URI}, got a {part.kind} part") - files[path] = await _fetch(part) + files[path] = await part_bytes(part) await self._computer.call("write_files", files) @get_data @@ -57,23 +58,31 @@ async def _computer_data_get(self, write_namespace: Optional[WriteNamespaceGrant named = _PATHS.validate_python(paths) if paths is not None else list(DEFAULT_DATA_PATHS) except ValidationError: raise ValueError(f"data/get on a computer takes paths relative to the home, not {paths!r}") from None - files = _data_files(Path.home(), named) - if write_namespace is not None: - uploader = NamespaceUploader(write_namespace) - parts = [] - for relative, source in files: - await uploader.upload(relative, source) - parts.append(uploaded_file_part(relative, name=source.name, mime_type=_mime(source))) - return parts - return [ - FilePart(file=FileWithBytes(bytes=base64.b64encode(source.read_bytes()).decode("ascii"), - name=source.name, mimeType=_mime(source)), - metadata={DATA_OBJECTS_EXTENSION_URI: {"path": relative}}) - for relative, source in files - ] - - -async def _fetch(part: FilePart) -> bytes: + return await file_parts(Path.home(), named, write_namespace, leaving_out=await self._computer.app_paths()) + + +async def file_parts(home: Path, paths: list[str], write_namespace: Optional[WriteNamespaceGrant], + leaving_out: list[str] = ()) -> list: + """A part for each file under ``paths`` of ``home`` but under none of ``leaving_out``, naming its path relative to + ``home``: uploaded under ``write_namespace`` when there is one, else inline.""" + files = [(relative, source) for relative, source in _data_files(home, paths) + if not any(PurePosixPath(relative).is_relative_to(owned) for owned in leaving_out)] + if write_namespace is not None: + uploader = NamespaceUploader(write_namespace) + parts = [] + for relative, source in files: + await uploader.upload(relative, source) + parts.append(uploaded_file_part(relative, name=source.name, mime_type=_mime(source))) + return parts + return [ + FilePart(file=FileWithBytes(bytes=base64.b64encode(source.read_bytes()).decode("ascii"), + name=source.name, mimeType=_mime(source)), + metadata={DATA_OBJECTS_EXTENSION_URI: {"path": relative}}) + for relative, source in files + ] + + +async def part_bytes(part: FilePart) -> bytes: """A part's bytes: inline, from a read grant under the marker, or from an https URL.""" marker = part.metadata[DATA_OBJECTS_EXTENSION_URI] if "read" in marker: diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/driver.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/driver.py index 17067b566..0b660db85 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/driver.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/driver.py @@ -1,18 +1,24 @@ """ComputerDriverEnvironment: a driver writes screen operations, and the base class serves the computer contract -from them: the tools agents call, the framework extensions agentenv calls, the data plane and the card.""" +from them: the tools agents call, the framework extensions agentenv calls, the data plane and the card, which lists the +computer's apps as child environments, each reached at ``/svc//`` through the driver.""" from __future__ import annotations import asyncio from typing import Any, Callable, ClassVar +import httpx from pydantic import BaseModel, ConfigDict +from starlette.requests import Request +from starlette.responses import Response -from ..agent_env_environment import _CARD_CONFIG_ATTR, AgentEnvEnvironment, AgentEnvFastMCPApplication, environment_card +from ..agent_env_environment import (_CARD_CONFIG_ATTR, AgentEnvEnvironment, AgentEnvFastMCPApplication, + _error_response, environment_card) from ..tool_context import ToolContext from ..types import EnvironmentCard, EnvironmentExtension, EnvironmentTool -from . import _linux, card +from . import _linux, apps, card from .checks import check_features, check_names from .dataplane import DataPlane +from .errors import DeliveryFailed, TargetUnavailable from .extensions import Extensions from .operations import ReadyCheck, check_class, default_for from .platform import Platform, detect @@ -118,11 +124,32 @@ async def bundled_skill(self) -> tuple[str, str, dict[str, str]] | None: def create_app(self) -> Any: self._computer_require_init() - return super().create_app() + return self._computer_front_apps(super().create_app()) def mount(self, app: Any) -> Any: self._computer_require_init() - return super().mount(app) + return self._computer_front_apps(super().mount(app)) + + async def children_environments(self) -> list[EnvironmentCard] | None: + """The computer's apps, from the folders it composed them into.""" + return await apps.children(self._computer.apps) if self._computer.apps else None + + def _computer_front_apps(self, app: Any) -> Any: + app.custom_route("/svc/{name}/{path:path}", methods=["GET", "POST"])(self._computer_forward) + return app + + async def _computer_forward(self, request: Request) -> Response: + """A request under ``/svc//``, sent on to the app named so: its own server answers it.""" + name = request.path_params["name"] + folder = next((folder for folder in self._computer.apps if folder.name == name), None) + if folder is None: + return _error_response(TargetUnavailable(f"this computer has no app named {name!r}")) + try: + answer = await apps.forward(folder, request.method, request.path_params["path"], request.url.query, + await request.body(), request.headers.get("content-type")) + except (httpx.HTTPError, OSError) as e: + return _error_response(DeliveryFailed(f"app {name}'s server isn't answering: {type(e).__name__}: {e}")) + return Response(answer.content, status_code=answer.status_code, media_type=answer.headers.get("content-type")) # The card: what the driver's operations support. diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/runtime.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/runtime.py index 2546679f0..51888f16a 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/runtime.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/runtime.py @@ -43,6 +43,10 @@ def __init__(self, driver: ComputerDriverEnvironment) -> None: self.recordings = Recordings(self) self.apps = apps.expected() + async def app_paths(self) -> list[str]: + """The paths under the session user's home that the computer's apps keep their data in.""" + return await apps.owned_paths(self.apps) if self.apps else [] + def has(self, capability: str) -> bool: if capability in GROUPS: return all(op in self.operations for op in GROUPS[capability]) diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py index e7367c42c..4ef274fa7 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/serve_app.py @@ -1,7 +1,8 @@ """Serve a computer app's class from its file, as its computer's program starts it: ``python -m agentenv_protocol.computer.serve_app :``, run in the app's folder, serves the class's card and -``computer-app/v1`` on the Unix socket ``.agentenv/server.sock`` there. Once it serves, the app's server adds the app's -launcher and launches a boot app (``AppRuntime.start``). A server that can't start, or whose start fails once it serves, +``computer-app/v1`` on the Unix socket ``.agentenv/server.sock`` there. Before it serves, it keeps a copy of the app's +data paths as installed, which a reset puts back; once it serves, it adds the app's launcher and launches a boot app +(``AppRuntime.start``). A server that can't start, or whose start fails once it serves, writes why to ``.agentenv/server.failure``, which its computer's driver reports, and exits. Needs the ``app`` extra.""" from __future__ import annotations @@ -36,6 +37,7 @@ def _serve(argv: list[str]) -> None: file, _, name = argv[0].partition(":") app = load_class(Path(file), name, ComputerAppEnvironment)() runtime = app._computer_app_runtime() + app._computer_app_data_plane().keep_initial() asgi = app.create_app().streamable_http_app() server = uvicorn.Server(uvicorn.Config(asgi, uds=str(SOCKET), log_level="warning")) serving, failed = asgi.router.lifespan_context, [] diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py index 3102ff8a5..1c8cdb726 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/session.py @@ -1,10 +1,12 @@ """What a computer app's own server does in the desktop session, the Linux way: wait for the session, launch the app -into it, add its menu launcher and list the session's windows. Another OS type brings its own.""" +into it and stop it, add its menu launcher and list the session's windows. Another OS type brings its own.""" from __future__ import annotations import asyncio +import contextlib import os import re +import signal from pathlib import Path #: How often a starting session is looked for. @@ -32,6 +34,36 @@ async def launch(argv: list[str], cwd: Path, env: dict[str, str]) -> int: return process.pid +async def stop(pid: int, timeout_s: float) -> None: + """End the session ``launch`` started as ``pid``, every process in it: asked to exit, then killed after + ``timeout_s``.""" + try: + os.killpg(pid, signal.SIGTERM) + except ProcessLookupError: + return + try: + await asyncio.wait_for(_exited(pid), timeout_s) + except TimeoutError: + with contextlib.suppress(ProcessLookupError): + os.killpg(pid, signal.SIGKILL) + + +def alive(pid: int) -> bool: + """Whether a process of the session ``launch`` started as ``pid`` is running, its first or one it started.""" + try: + os.killpg(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +async def _exited(pid: int) -> None: + while alive(pid): + await asyncio.sleep(POLL_S) + + def add_launcher(name: str, folder: Path) -> Path: """The desktop entry that launches the app ``name`` in ``folder`` through its own server, written where the session user's menu finds it.""" diff --git a/packages/agentenv-protocol/tests/test_client.py b/packages/agentenv-protocol/tests/test_client.py index 13088bf08..edcd38798 100644 --- a/packages/agentenv-protocol/tests/test_client.py +++ b/packages/agentenv-protocol/tests/test_client.py @@ -148,6 +148,14 @@ def test_extension_params_returns_empty_when_absent_or_paramless(): } +def test_operations_are_what_a_card_declares_and_none_when_it_declares_none(): + assert protocol_v1.operations({"capabilities": {"operations": ["data/get"]}}) == ["data/get"] + assert protocol_v1.operations({"capabilities": {"operations": []}}) == [] + assert protocol_v1.operations({"capabilities": {}}) is None + assert protocol_v1.operations({"capabilities": None}) is None + assert protocol_v1.operations({}) is None + + def test_find_tool_returns_matching(): assert protocol_v1.find_tool(_CARD_WITH_TOOLS, "slack_ping")["description"] == "Ping the server." diff --git a/packages/agentenv-protocol/tests/test_computer_app_data.py b/packages/agentenv-protocol/tests/test_computer_app_data.py new file mode 100644 index 000000000..abbd71509 --- /dev/null +++ b/packages/agentenv-protocol/tests/test_computer_app_data.py @@ -0,0 +1,280 @@ +"""A computer app's data plane, served by its own server. An app that declares only paths, none inside another, gets +the defaults: a load writes files under them, through no link out of them, an export returns them, and a reset puts back +the copy its server kept before it served; its card declares a file tree. Hooks override paths, and read the parts they +get with part_bytes. Around a load and a reset the app is stopped and relaunched, a failed load is followed by a reset, +and data calls take turns. A call past data_timeout_s fails then and is cancelled, but a sync hook runs in a thread, +which its reset and relaunch, and the next call, wait for. The app serves its skill file.""" +from __future__ import annotations + +import asyncio +import base64 +import contextlib +import os +import time + +import pytest + +pytest.importorskip("mcp.server.fastmcp") + +import httpx # noqa: E402 + +from agentenv_protocol import add_data, client, reset_data # noqa: E402 +from agentenv_protocol.computer import ComputerAppEnvironment, Launch, linux, part_bytes # noqa: E402 +from agentenv_protocol.computer import app, app_runtime, session # noqa: E402 +from agentenv_protocol.computer.app import COMPUTER_APP # noqa: E402 +from agentenv_protocol.computer.platform import Platform # noqa: E402 +from agentenv_protocol.types import (DATA_OBJECTS_EXTENSION_URI, INTAKE_EXTENSION_URI, RPC_PATH, # noqa: E402 + SKILL_EXTENSION_URI, WELL_KNOWN_PATH) + +MAIL = ["Mail", ".config/mail/accounts.json"] + + +@linux(paths=MAIL, launch=Launch(command=["mail"])) +class PathsMail(ComputerAppEnvironment): + skill = "SKILL.md" + + +@linux(paths=MAIL, launch=Launch(command=["mail"])) +class HookedMail(ComputerAppEnvironment): + loaded: dict + resets = 0 + + @add_data + async def load(self, parts: list) -> None: + if any(part.metadata[DATA_OBJECTS_EXTENSION_URI]["path"] == "bad" for part in parts): + raise RuntimeError("the seed is bad") + self.loaded = {part.metadata[DATA_OBJECTS_EXTENSION_URI]["path"]: await part_bytes(part) for part in parts} + + @reset_data + def wipe(self) -> None: + type(self).resets += 1 + + +@pytest.fixture +def home(tmp_path, monkeypatch): + """The session user's home, the app's folder as the current directory, on Linux, the app named mail.""" + monkeypatch.setenv("HOME", str(tmp_path / "home")) + (tmp_path / "home").mkdir() + (tmp_path / "folder").mkdir() + monkeypatch.chdir(tmp_path / "folder") + monkeypatch.setattr(app, "detect", lambda: Platform(os="linux")) + monkeypatch.setenv("ENVIRONMENT_NAME", "mail") + monkeypatch.setattr(app_runtime, "LAUNCHED", tmp_path / "launched") + return tmp_path / "home" + + +@pytest.fixture +def desktop(monkeypatch): + """Each launch a new pid, alive until stopped; what's launched and stopped, in order.""" + events, running = [], set() + + async def launch(argv, cwd, env): + pid = 4242 + len(events) + running.add(pid) + events.append(("launch", pid)) + return pid + + async def stop(pid, timeout_s): + running.discard(pid) + events.append(("stop", pid)) + + monkeypatch.setattr(session, "launch", launch) + monkeypatch.setattr(session, "stop", stop) + monkeypatch.setattr(session, "alive", lambda pid: pid in running) + return events + + +@pytest.mark.asyncio +async def test_an_app_with_only_paths_loads_exports_and_resets_them_as_installed(home): + (home / "Mail").mkdir() + (home / "Mail/inbox.txt").write_text("installed") + mail = PathsMail() + mail._computer_app_data_plane().keep_initial() + + async with _http(mail) as http: + card = (await http.get(WELL_KNOWN_PATH)).json() + await _rpc(http, "data/add", {"parts": [_inline("Mail/new.txt", b"new"), _inline("Mail/inbox.txt", b"loaded"), + _inline(".config/mail/accounts.json", b"{}")]}) + loaded = _files(await _rpc(http, "data/get", {})) + await _rpc(http, "data/reset", {}) + reset = _files(await _rpc(http, "data/get", {})) + + assert sorted(card["capabilities"]["operations"]) == ["data/add", "data/get", "data/reset"] + assert client.extension_params(card, COMPUTER_APP)["paths"] == MAIL + assert client.find_extension(card, INTAKE_EXTENSION_URI) is not None + assert loaded == {"Mail/inbox.txt": b"loaded", "Mail/new.txt": b"new", ".config/mail/accounts.json": b"{}"} + assert reset == {"Mail/inbox.txt": b"installed"} + + +@pytest.mark.asyncio +async def test_a_file_under_none_of_its_paths_or_through_a_link_out_of_them_is_refused_and_nothing_is_written(home): + (home / "Documents").mkdir() + (home / "Mail").mkdir() + os.symlink(home / "Documents", home / "Mail" / "shared") + mail = PathsMail() + mail._computer_app_data_plane().keep_initial() + + async with _http(mail) as http: + outside = await http.post(RPC_PATH, json=_call("data/add", {"parts": [_inline("Mail/a.txt", b"a"), + _inline("Documents/b.txt", b"b")]})) + linked = await http.post(RPC_PATH, json=_call("data/add", {"parts": [_inline("Mail/a.txt", b"a"), + _inline("Mail/shared/b.txt", b"b")]})) + + assert "'Documents/b.txt' is under none of this app's paths" in outside.json()["error"]["message"] + assert "'Mail/shared/b.txt' lead out of this app's paths through a link" in linked.json()["error"]["message"] + assert sorted(p.name for p in (home / "Mail").iterdir()) == ["shared"] and not any((home / "Documents").iterdir()) + + +@pytest.mark.parametrize("paths, problem", [ + (["/etc/mail"], "paths are normalized paths relative to the session user's home"), + (["Mail", "Mail/inbox"], "an app's paths each hold files of their own, but 'Mail' and 'Mail/inbox' overlap"), + (["Mail", "Mail"], "but 'Mail' and 'Mail' overlap"), +], ids=["absolute", "nested", "twice"]) +def test_paths_that_arent_relative_or_overlap_are_refused_where_theyre_declared(paths, problem): + with pytest.raises(ValueError, match=problem): + linux(paths=paths) + + +@pytest.mark.asyncio +async def test_hooks_override_its_paths_and_read_each_part_they_get_with_part_bytes(home): + mail = HookedMail() + + async with _http(mail) as http: + card = (await http.get(WELL_KNOWN_PATH)).json() + await _rpc(http, "data/add", {"parts": [_inline("Mail/a.txt", b"a")]}) + + assert mail.loaded == {"Mail/a.txt": b"a"} and not (home / "Mail").exists() + assert sorted(card["capabilities"]["operations"]) == ["data/add", "data/reset"] + assert client.find_extension(card, INTAKE_EXTENSION_URI) is None + assert HookedMail.declaration().data == ["data/add", "data/reset"] + + +@pytest.mark.asyncio +async def test_a_load_and_a_reset_stop_the_running_app_and_relaunch_it_and_an_export_leaves_it_be(home, desktop): + mail = PathsMail() + mail._computer_app_data_plane().keep_initial() + await mail._computer_app.launch_app() + + async with _http(mail) as http: + await _rpc(http, "data/add", {"parts": [_inline("Mail/a.txt", b"a")]}) + await _rpc(http, "data/get", {}) + await _rpc(http, "data/reset", {}) + + assert desktop == [("launch", 4242), ("stop", 4242), ("launch", 4244), ("stop", 4244), ("launch", 4246)] + + +@pytest.mark.asyncio +async def test_an_app_that_isnt_running_is_left_stopped(home, desktop): + mail = PathsMail() + mail._computer_app_data_plane().keep_initial() + + async with _http(mail) as http: + await _rpc(http, "data/add", {"parts": [_inline("Mail/a.txt", b"a")]}) + + assert desktop == [] + + +@pytest.mark.asyncio +async def test_a_failed_load_is_followed_by_a_reset_and_the_app_is_relaunched(home, desktop): + mail, HookedMail.resets = HookedMail(), 0 + await mail._computer_app.launch_app() + + async with _http(mail) as http: + answer = await http.post(RPC_PATH, json=_call("data/add", {"parts": [_inline("bad", b"x")]})) + + assert answer.json()["error"]["message"] == "the seed is bad" + assert HookedMail.resets == 1 and desktop == [("launch", 4242), ("stop", 4242), ("launch", 4244)] + + +@pytest.mark.asyncio +async def test_an_async_call_past_its_timeout_fails_saying_so_and_is_cancelled_so_the_next_takes_its_turn( + home, desktop): + @linux(launch=Launch(command=["slow"])) + class Slow(ComputerAppEnvironment): + data_timeout_s = 0.2 + + @add_data + async def load(self, parts: list) -> None: + if len(parts) == 2: + await asyncio.Event().wait() + + async with _http(Slow()) as http: + answer = await http.post(RPC_PATH, json=_call("data/add", {"parts": [_inline("a.txt", b"a"), + _inline("b.txt", b"b")]})) + await _rpc(http, "data/add", {"parts": [_inline("c.txt", b"c")]}) + + assert answer.json()["error"]["message"] == "add_data ran past the app's data_timeout_s, 0.2 s" + + +@pytest.mark.asyncio +async def test_calls_take_turns_and_one_past_its_timeout_fails_then_while_what_follows_waits_for_its_thread( + home, desktop): + @linux(launch=Launch(command=["slow"])) + class Slow(ComputerAppEnvironment): + data_timeout_s = 1.0 + + @add_data + def load(self, parts: list) -> None: + desktop.append(("load", len(parts))) + time.sleep(1.4 if len(parts) == 2 else 0) + desktop.append(("loaded", len(parts))) + + @reset_data + def wipe(self) -> None: + desktop.append(("reset",)) + + slow = Slow() + await slow._computer_app.launch_app() + + async with _http(slow) as http: + late = await http.post(RPC_PATH, json=_call("data/add", {"parts": [_inline("a", b"a"), _inline("b", b"b")]})) + desktop.append(("answered",)) + card = await http.get(WELL_KNOWN_PATH) + await _rpc(http, "data/add", {"parts": [_inline("c", b"c")]}) + + assert late.json()["error"]["message"] == "add_data ran past the app's data_timeout_s, 1 s" + assert card.status_code == 200 + assert desktop == [("launch", 4242), ("stop", 4242), ("load", 2), ("answered",), ("loaded", 2), ("reset",), + ("launch", 4248), ("stop", 4248), ("load", 1), ("loaded", 1), ("launch", 4252)] + + +@pytest.mark.asyncio +async def test_it_serves_its_skill_file_and_no_skill_without_one(home, tmp_path): + (tmp_path / "folder" / "SKILL.md").write_text("---\ndescription: Reading and sending mail.\n---\n\n# Mail\nUse it.\n") + + async with _http(PathsMail()) as http: + card = (await http.get(WELL_KNOWN_PATH)).json() + skill = (await http.get(client.find_extension_method(card, SKILL_EXTENSION_URI, "get")["endpoint"], + params={"name": "mail"})).json() + async with _http(HookedMail()) as http: + unskilled = (await http.get(WELL_KNOWN_PATH)).json() + + assert skill["description"] == "Reading and sending mail." and skill["skill_md"].endswith("# Mail\nUse it.\n") + assert client.find_extension(unskilled, SKILL_EXTENSION_URI) is None + + +@contextlib.asynccontextmanager +async def _http(served: ComputerAppEnvironment): + asgi = served.create_app().streamable_http_app() + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=asgi), base_url="http://t") as http: + yield http + + +def _call(method: str, params: dict) -> dict: + return {"jsonrpc": "2.0", "id": 1, "method": method, "params": params} + + +async def _rpc(http: httpx.AsyncClient, method: str, params: dict) -> dict: + answer = (await http.post(RPC_PATH, json=_call(method, params))).json() + assert "error" not in answer, answer + return answer["result"] + + +def _inline(path: str, data: bytes) -> dict: + return {"kind": "file", "file": {"bytes": base64.b64encode(data).decode(), "name": path.rsplit("/", 1)[-1]}, + "metadata": {DATA_OBJECTS_EXTENSION_URI: {"path": path}}} + + +def _files(result: dict) -> dict[str, bytes]: + return {part["metadata"][DATA_OBJECTS_EXTENSION_URI]["path"]: base64.b64decode(part["file"]["bytes"]) + for part in result["parts"]} diff --git a/packages/agentenv-protocol/tests/test_computer_app_server.py b/packages/agentenv-protocol/tests/test_computer_app_server.py index 752afbbdf..231c281e4 100644 --- a/packages/agentenv-protocol/tests/test_computer_app_server.py +++ b/packages/agentenv-protocol/tests/test_computer_app_server.py @@ -1,5 +1,6 @@ """A computer app's own server: it adds the app's launcher, launches a boot app once per boot and says whether the app -is ready (``computer-app/v1``); its computer's driver finds it on the socket in the app's folder and asks it.""" +is ready (``computer-app/v1``); its computer's driver finds it on the socket in the app's folder and asks it, lists it +as a child environment and forwards its requests to it, and leaves the paths it owns out of its own data/get.""" from __future__ import annotations import asyncio @@ -19,6 +20,8 @@ pytest.importorskip("mcp.server.fastmcp") uvicorn = pytest.importorskip("uvicorn") +import base64 # noqa: E402 + import httpx # noqa: E402 import agentenv_protocol.computer.launch as launcher # noqa: E402 @@ -28,6 +31,7 @@ from agentenv_protocol.computer import app, app_runtime, apps, session # noqa: E402 from agentenv_protocol.computer.app import COMPUTER_APP, FAILURE, SOCKET # noqa: E402 from agentenv_protocol.computer.platform import Platform # noqa: E402 +from agentenv_protocol.types import DATA_OBJECTS_EXTENSION_URI, WELL_KNOWN_PATH # noqa: E402 @@ -268,6 +272,45 @@ async def settled(self): assert status["ready"] is True and took < 2.5 +@linux(paths=["Mail"]) +class Inbox(ComputerAppEnvironment): + pass + + +@pytest.mark.asyncio +async def test_the_driver_lists_its_apps_as_children_forwards_each_ones_requests_and_leaves_their_paths_out( + tmp_path, monkeypatch): + from computer_fake import FakeDriver, http_client + + monkeypatch.setenv("HOME", str(tmp_path)) + for path in ("Mail/inbox.txt", "Documents/note.txt"): + (tmp_path / path).parent.mkdir(parents=True, exist_ok=True) + (tmp_path / path).write_text(path) + get = {"jsonrpc": "2.0", "id": 1, "method": "data/get", "params": {}} + with tempfile.TemporaryDirectory(dir="/tmp", prefix="apps-") as root: + monkeypatch.setattr(apps, "APPS", Path(root)) + (Path(root) / "music").mkdir() + async with _serving(Inbox(), Path(root) / "mail", monkeypatch): + async with http_client(FakeDriver()) as http: + card = (await http.get(WELL_KNOWN_PATH)).json() + mail = client.find_child(card, "mail") + forwarded = (await http.post(mail["url"], json=get)).json()["result"]["parts"] + unknown = await http.post("/svc/notes/agentenv", json=get) + silent = await http.post("/svc/music/agentenv", json=get) + own = (await http.post("/agentenv", json={**get, "params": {"paths": ["Documents", "Mail"]}})).json() + + assert [(child["name"], child["url"]) for child in card["children_environments"]] == [ + ("mail", "/svc/mail/agentenv"), ("music", "/svc/music/agentenv")] + assert client.extension_params(mail, COMPUTER_APP)["paths"] == ["Mail"] + assert client.find_extension_method(mail, COMPUTER_APP, "ready")["endpoint"].startswith("/svc/mail/") + assert [(part["metadata"][DATA_OBJECTS_EXTENSION_URI]["path"], base64.b64decode(part["file"]["bytes"])) + for part in forwarded] == [("Mail/inbox.txt", b"Mail/inbox.txt")] + assert (unknown.status_code, unknown.json()["error"]["code"]) == (404, "target_unavailable") + assert (silent.status_code, silent.json()["error"]["code"]) == (502, "delivery_failed") + assert [part["metadata"][DATA_OBJECTS_EXTENSION_URI]["path"] for part in own["result"]["parts"]] == [ + "Documents/note.txt"] + + @pytest.mark.asyncio async def test_the_session_launches_an_app_in_a_session_of_its_own_from_its_folder(tmp_path, monkeypatch): monkeypatch.setenv("DISPLAY", ":99") @@ -280,6 +323,18 @@ async def test_the_session_launches_an_app_in_a_session_of_its_own_from_its_fold assert (tmp_path / "seen").read_text() == f"{os.path.realpath(tmp_path)} notes :99 True" +@pytest.mark.asyncio +async def test_a_stop_ends_what_the_app_started_though_its_first_process_exited(tmp_path): + sleeper = f"{sys.executable} -c 'import time; time.sleep(60)' & exit 0" + pid = await session.launch(["sh", "-c", sleeper], tmp_path, {}) + await asyncio.gather(*session._running) + + running = session.alive(pid) + await session.stop(pid, 5) + + assert running and not session.alive(pid) + + def test_the_launcher_runs_the_app_through_its_own_server(tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) folder = Path("/opt/agentenv/apps/notes") diff --git a/packages/agentenv-protocol/tests/test_computer_conformance.py b/packages/agentenv-protocol/tests/test_computer_conformance.py index 5f14c3cdd..154738a95 100644 --- a/packages/agentenv-protocol/tests/test_computer_conformance.py +++ b/packages/agentenv-protocol/tests/test_computer_conformance.py @@ -1,6 +1,6 @@ """The conformance probe against fake drivers on loopback: the fake driver with the fake fixture passes every check; a -driver without what full needs gets those checks unverified; a driver broken one way fails the check for that way; -the report's verdicts and the CLI.""" +driver without what full needs gets those checks unverified; a driver broken one way fails the check for that way; each +app a driver fronts that exports is asked for its export once; the report's verdicts and the CLI.""" from __future__ import annotations import json @@ -34,6 +34,7 @@ from agentenv_protocol.computer.conformance.client import Driver # noqa: E402 from agentenv_protocol.computer.conformance.__main__ import main # noqa: E402 from agentenv_protocol.tool_access import DISABLE_TOOL_EXTENSION_URI # noqa: E402 +from agentenv_protocol.types import EnvironmentCard # noqa: E402 @pytest.fixture(autouse=True) @@ -97,6 +98,26 @@ async def test_basic_needs_no_fixture(home): assert not driver.desk.started and not (home / ".agentenv-probe").exists() +class FrontsApps(FixtureDriver): + """Fronts mail, whose card routes to the driver's own data plane, notes, which no app answers, and quiet, which + doesn't export.""" + + async def children_environments(self): + return [EnvironmentCard(name=name, url=url, capabilities={"operations": operations}) + for name, url, operations in (("mail", "/agentenv", ["data/get"]), + ("notes", "/svc/notes/agentenv", ["data/get"]), + ("quiet", "/svc/quiet/agentenv", ["data/add"]))] + + +@pytest.mark.asyncio +async def test_each_app_a_driver_fronts_that_exports_is_asked_for_its_export_once(home): + report = await _probe(FrontsApps(), "basic") + + assert [(c.name, c.status) for c in report.checks if c.name.startswith("app ")] == [ + ("app mail's data/get answers its parts through the driver", "passed"), + ("app notes's data/get answers its parts through the driver", "failed")] + + @pytest.mark.asyncio async def test_a_driver_without_exec_or_data_add_passes_basic_and_leaves_full_unverified(): report = await _probe(MinimalDriver(FakeScreen(640, 400), platform=Platform(os="macos"))) diff --git a/src/agent_env/bundle/plan.py b/src/agent_env/bundle/plan.py index 0a8d867f3..bdea42432 100644 --- a/src/agent_env/bundle/plan.py +++ b/src/agent_env/bundle/plan.py @@ -968,16 +968,19 @@ def _check_loaded_environments(self, closure: set[_Key], "seeds, or leave it out") def _server_names(self, ref: Reference, store_latest: dict[tuple[EntityKind, str], int] | None) -> set[str] | None: - """The environment_names of the servers the env ``ref`` names runs: an MCP server's or a website's own, or - each of a multi's envs'. None when they aren't all known.""" + """The environment_names of the servers the env ``ref`` names runs: an MCP server's or a website's own, each + of a multi's envs', or a computer's own and each of its apps'. None when they aren't all known.""" if isinstance(ref.local, BundleEntry): cls = get_env_registry().get(ref.local.type) + children = self.nodes[_key(ref.local)].references if cls is not None and issubclass(cls, MultiEnv): - children = self.nodes[_key(ref.local)].references names = {self._environment_name(child, store_latest) for child in children if child.kind is EntityKind.ENV} else: names = {self.accepted.get(_key(ref.local), {}).get("environment_name")} + if cls is not None and issubclass(cls, ComputerEnv): + names |= {self._environment_name(child, store_latest) for child in children + if child.kind is EntityKind.ENV and child.artifact_type == ComputerAppEnv.type} return None if None in names else names if ref.local is not None or store_latest is None: return None @@ -987,6 +990,8 @@ def _server_names(self, ref: Reference, store_latest: dict[tuple[EntityKind, str return None # reported by the store ref check if isinstance(env, MultiEnv): return {child.environment_name for child in [*env.mcp_server_envs, *env.website_envs]} + if isinstance(env, ComputerEnv): + return {env.environment_name, *(app.environment_name for app in env.composed_apps())} name = getattr(env, "environment_name", None) return {name} if name else None diff --git a/src/agent_env/env/env.py b/src/agent_env/env/env.py index c92199f7f..3bdc55afc 100644 --- a/src/agent_env/env/env.py +++ b/src/agent_env/env/env.py @@ -119,11 +119,15 @@ def require(self, uri: str, method: str) -> None: raise EnvCapabilityUnsupported(uri, method, self.env_id, self.instance_id) def get_child_env_card(self, environment_name: str) -> Optional[dict]: - """The stored card's child env named `environment_name`; a leaf card is its own only child env.""" + """The stored card's child env named `environment_name`; a leaf card is its own only child env, and a composed + card serving a data plane of its own, as a computer's driver fronting its apps does, is the one its own name + reaches.""" card = self.environment_card or {} - if not isinstance(card.get("children_environments"), list): - return card if card.get("name") == environment_name else None - return protocol_v1.find_child(card, environment_name) + composed = isinstance(card.get("children_environments"), list) + own = not composed or protocol_v1.operations(card) + if own and card.get("name") == environment_name: + return card + return protocol_v1.find_child(card, environment_name) if composed else None async def invoke(self, uri: str, method: str, params: Optional[dict] = None, *, environment_name: Optional[str] = None, timeout: int = _EXTENSION_CALL_TIMEOUT_S) -> Any: diff --git a/src/agent_env/env/envs/_deployment.py b/src/agent_env/env/envs/_deployment.py index 8cd3858d0..fc097d55b 100644 --- a/src/agent_env/env/envs/_deployment.py +++ b/src/agent_env/env/envs/_deployment.py @@ -112,7 +112,7 @@ async def load_by_signed_url(env: MCPServerEnv | WebsiteEnv, environment_artifac if not env._deployed.environment_card: raise RuntimeError(f"{deployed_by}, whose record carries no env card, so '{env.environment_name}' has no data plane to load into") base_url = await legacy_protocol.v1_base_url(env._deployed, env._gateway_url, env.environment_name, mcp=True) - operations = ((env._deployed.get_child_env_card(env.environment_name) or {}).get("capabilities") or {}).get("operations") + operations = protocol_v1.operations(env._deployed.get_child_env_card(env.environment_name) or {}) if base_url is None: raise RuntimeError(f"{deployed_by}, whose card offers '{env.environment_name}' no data plane to load into") if operations is not None and (missing := [op for op in LOAD_OPERATIONS if op not in operations]): diff --git a/src/agent_env/env/envs/computer/app_env.py b/src/agent_env/env/envs/computer/app_env.py index f9d9035fe..af60f796c 100644 --- a/src/agent_env/env/envs/computer/app_env.py +++ b/src/agent_env/env/envs/computer/app_env.py @@ -114,7 +114,8 @@ def put(cls, *, id: str, environment_name: str, module: str, files: FileArtifact if (computer is None) != (driver is None): problems.append("a computer app deploys alone on a computer and a driver; name both or neither") elif computer is not None and not problems: - _, refused = gate.check(computer, driver, apps=[], display=None, network=None, computer_settings={}) + _, refused = gate.check(computer, driver, environment_name=environment_name, apps=[], display=None, + network=None, computer_settings={}) problems += refused + gate.check_app(computer, driver, app) if problems: raise ValueError(f"computer app env {id!r}: " + "; ".join(problems)) @@ -144,17 +145,18 @@ def from_toml(cls, data: dict, ctx: AuthoringContext) -> ComputerAppEnv: driver = ctx.env(fields["driver"], ComputerDriverEnv) if "driver" in fields else None universe = FileArtifactUniverse.put_bundled(derive_id(ctx.id, "files"), files=files) try: - return cls.put(id=ctx.id, environment_name=fields.get("environment_name", ctx.name), - module=fields["module"], files=universe, declaration=declaration, computer=computer, - driver=driver, build_settings=fields.get("build_settings"), composing=ctx.composing) + return cls.put(id=ctx.id, environment_name=fields["environment_name"], module=fields["module"], + files=universe, declaration=declaration, computer=computer, driver=driver, + build_settings=fields.get("build_settings"), composing=ctx.composing) except ValueError as e: ctx.refuse([str(e)]) @classmethod def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict: """The keys of ``data``, an env.toml, a computer app env takes: ``module = ":"``, naming a file of - its folder; ``environment_name``; the ``computer`` and ``driver`` it deploys alone on; and ``build_settings``, - the values of its build settings. Raises BundleError listing every problem.""" + its folder; ``environment_name``, the folder's when it sets none; the ``computer`` and ``driver`` it deploys + alone on; and ``build_settings``, the values of its build settings. Raises BundleError listing every + problem.""" fields, refused = ctx.accepted(data, environment_name=str, module=str, computer=object, driver=object, build_settings=dict) files, module, problems = ctx.files(), fields.get("module"), [] @@ -165,7 +167,7 @@ def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict: f"{', '.join(sorted(files))}") if refused or problems: ctx.refuse([*refused, *map(ctx.config_problem, problems)]) - return fields + return {"environment_name": ctx.name, **fields} def deploy_refusal(self, *, sandbox_type: str | None = None, env_params: Optional[dict] = None, **options) -> str | None: diff --git a/src/agent_env/env/envs/computer/env.py b/src/agent_env/env/envs/computer/env.py index 9270f1655..a92ccd9e7 100644 --- a/src/agent_env/env/envs/computer/env.py +++ b/src/agent_env/env/envs/computer/env.py @@ -8,6 +8,9 @@ import posixpath from typing import TYPE_CHECKING, Any, ClassVar, Optional, Self +from agentenv_protocol import METHOD_ADD, RPC_PATH +from agentenv_protocol import client as protocol_v1 + from agent_env.artifact.artifacts.computer import ComputerArtifact, SettingValue from agent_env.attribution import Attribution from agent_env.entity.refs import EntityRef @@ -95,8 +98,8 @@ def put(cls, *, id: str, computer: ComputerArtifact, driver: ComputerDriverEnv, order, once gate 1 passes, then compose it as ``composing`` says unless ``compose`` is False. Refused, naming every mismatch, before anything is stored.""" apps = list(apps) - values, problems = gate.check(computer, driver, apps=apps, display=display, network=network, - computer_settings=computer_settings or {}) + values, problems = gate.check(computer, driver, environment_name=environment_name, apps=apps, display=display, + network=network, computer_settings=computer_settings or {}) if problems: raise ValueError(f"computer env {id!r}: " + "; ".join(problems)) env = get_env_store().put_document(cls( @@ -111,22 +114,28 @@ def from_toml(cls, data: dict, ctx: AuthoringContext) -> ComputerEnv: """Write the computer env an env.toml declares, its base, driver and apps as the bundle wrote them or the ones they name, its environment_name the folder's when it sets none. An app's ``install`` is ``"build"``, the default: it's built into the computer.""" - fields, refused = ctx.accepted(data, environment_name=str, computer=object, driver=object, apps=list, - display=dict, network=dict, computer_settings=dict) - missing = [ctx.config_problem(f"{key} is required: the {what}, such as {example}") for key, what, example in ( - ("computer", "base computer", '"ubuntu-24.04"'), ("driver", "driver", '"reference"')) if key not in data] - if refused or missing: - ctx.refuse([*refused, *missing]) + fields = cls.accept_toml(data, ctx) base, driver = ctx.artifact(fields.pop("computer"), ComputerArtifact), ctx.env(fields.pop("driver"), ComputerDriverEnv) fields["apps"] = _listed_apps(fields.get("apps", []), ctx) try: - return cls.put(id=ctx.id, computer=base, driver=driver, - environment_name=fields.pop("environment_name", ctx.name), composing=ctx.composing, - **fields) + return cls.put(id=ctx.id, computer=base, driver=driver, composing=ctx.composing, **fields) except ValueError as e: ctx.refuse([str(e)]) + @classmethod + def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict: + """The keys of ``data``, an env.toml, a computer env takes: the ``computer`` and ``driver`` it's composed of, + both required, its ``apps``, ``display``, ``network`` and ``computer_settings``, and ``environment_name``, the + folder's when it sets none. Raises BundleError listing every problem.""" + fields, refused = ctx.accepted(data, environment_name=str, computer=object, driver=object, apps=list, + display=dict, network=dict, computer_settings=dict) + missing = [ctx.config_problem(f"{key} is required: the {what}, such as {example}") for key, what, example in ( + ("computer", "base computer", '"ubuntu-24.04"'), ("driver", "driver", '"reference"')) if key not in data] + if refused or missing: + ctx.refuse([*refused, *missing]) + return {"environment_name": ctx.name, **fields} + def composed_apps(self) -> list[ComputerAppEnv]: """The computer app envs composed into the env's computer, each installed under its prefix.""" return self.apps @@ -208,32 +217,25 @@ async def load_file_artifact_universe(self, file_artifact_universe: Any, return LoadFileArtifactUniverseResult(destination_path=destination, files=paths) async def load_environment_artifact(self, environment_artifact: EnvironmentArtifact) -> None: - """Write the files of ``environment_artifact``, the computer's own, at their paths relative to the session - user's home, through the driver's ``data/add``, as ``snapshot_computer_env`` kept them. Other files there - stay.""" - if environment_artifact.environment_name != self.environment_name: - raise ValueError(f"EnvironmentArtifact environment_name '{environment_artifact.environment_name}' does not " - f"match env environment_name '{self.environment_name}'") - files = environment_artifact.get_file_artifacts() - for path in files: - validate_relative_filename(path) - await inside.add_artifacts(self._deployed, files) - logger.info(f"Loaded {len(files)} file(s) of '{environment_artifact.id}' into {self.id}'s home") + """Write the files of ``environment_artifact`` at their paths through the data plane its name reaches: the + computer's own relative to the session user's home, through the driver's ``data/add``, as + ``snapshot_computer_env`` kept them; an app's through the app's, which the driver fronts. Other files stay.""" + await self._load([environment_artifact]) async def load_environment_universe_artifact(self, environment_universe_artifact: EnvironmentUniverseArtifact) -> None: - """Load the universe's environment named after the computer, ``load_environment_artifact``; any other is left - out, with a warning.""" + """Load the universe's environment named after the computer, then each named after one of its apps, so what an + app keeps ends as its own export has it; any other is left out, with a warning.""" environments = environment_universe_artifact.get_environment_artifacts() - own = [environment for environment in environments if environment.environment_name == self.environment_name] - if not own: + names = self._environment_names() + if not (known := [environment for environment in environments if environment.environment_name in names]): raise RuntimeError(f"computer env '{self.id}' can't load '{environment_universe_artifact.id}': it has no " - f"environment named '{self.environment_name}' (it has " + f"environment named any of {names} (it has " f"{[environment.environment_name for environment in environments]})") if others := [environment.environment_name for environment in environments - if environment.environment_name != self.environment_name]: - logger.warning(f"[{self.id}] loading only '{self.environment_name}' from '{environment_universe_artifact.id}'; " - f"leaving out {others}") - await self.load_environment_artifact(own[0]) + if environment.environment_name not in names]: + logger.warning(f"[{self.id}] loading only {[environment.environment_name for environment in known]} from " + f"'{environment_universe_artifact.id}'; leaving out {others}") + await self._load(sorted(known, key=lambda environment: environment.environment_name != self.environment_name)) async def reset(self, deployed: DeployedEnv) -> None: """Return the computer to boot state: its recordings ended, its container replaced with a fresh one of the same @@ -258,6 +260,35 @@ async def close(self) -> None: logger.warning(f"Failed to terminate sandbox {self._sandbox.sandbox_id}: {e}") self._sandbox = None + async def _load(self, environments: list[EnvironmentArtifact]) -> None: + """Write each environment's files through the data plane its name reaches, every one checked before any is + sent.""" + loads = [(environment, self._data_plane(environment.environment_name), environment.get_file_artifacts()) + for environment in environments] + for _, _, files in loads: + for path in files: + validate_relative_filename(path) + for environment, url, files in loads: + await inside.add_artifacts(self._deployed, files, url) + logger.info(f"Loaded {len(files)} file(s) of '{environment.id}' into {self.id}'s " + f"'{environment.environment_name}'") + + def _data_plane(self, environment_name: str) -> str: + """The address of the data plane ``environment_name`` reaches in the deployed computer, as its card says: the + driver's for the computer's own, an app's route for an app's.""" + card = self._deployed.get_child_env_card(environment_name) + if card is None: + raise ValueError(f"computer env '{self.id}' has no environment named '{environment_name}': it has " + f"{self._environment_names()}") + if METHOD_ADD not in (protocol_v1.operations(card) or []): + raise ValueError(f"'{environment_name}' in computer env '{self.id}' serves no data/add, so nothing loads " + "into it") + return self._deployed.environment_url + card.get("url", RPC_PATH).removesuffix(RPC_PATH) + + def _environment_names(self) -> list[str]: + """The environments a load reaches by name: the computer's own, then its apps'.""" + return list(dict.fromkeys([self.environment_name, *(app.environment_name for app in self.composed_apps())])) + def _app(env: Env) -> ComputerAppEnv: if refusal := _not_an_app(env): diff --git a/src/agent_env/env/envs/computer/utils/boot.py b/src/agent_env/env/envs/computer/utils/boot.py index 5e5b2bbe0..70b4ea55c 100644 --- a/src/agent_env/env/envs/computer/utils/boot.py +++ b/src/agent_env/env/envs/computer/utils/boot.py @@ -125,8 +125,9 @@ def ready_timeout(apps: list[ComputerAppEnv], os_type: str) -> float: async def ready(url: str, timeout: float) -> dict: - """The card of the driver at ``url`` once its ``computer-ready`` status says it's ready; raises TimeoutError, naming - what it last saw, when it isn't within ``timeout`` seconds.""" + """The card of the driver at ``url``, read once its ``computer-ready`` status says it's ready, so each app it fronts + is on it as the app's own server serves it; raises TimeoutError, naming what it last saw, when it isn't within + ``timeout`` seconds.""" last = "nothing answered" try: async with asyncio.timeout(timeout): @@ -136,7 +137,7 @@ async def ready(url: str, timeout: float) -> dict: status = await protocol_v1.invoke_extension(url, card, COMPUTER_READY, None, _STATUS_TIMEOUT_S, method="status") if status.get("ready"): - return card + return await protocol_v1.get_card(url, timeout=5) last = "not ready: " + "; ".join(f"{check['name']}: {check.get('detail')}" for check in status.get("checks", []) if not check.get("ok")) except Exception as e: diff --git a/src/agent_env/env/envs/computer/utils/gate.py b/src/agent_env/env/envs/computer/utils/gate.py index 5ec5cf8f2..b0b0a3126 100644 --- a/src/agent_env/env/envs/computer/utils/gate.py +++ b/src/agent_env/env/envs/computer/utils/gate.py @@ -3,11 +3,12 @@ from __future__ import annotations +import itertools +import re from collections.abc import Sequence +from pathlib import PurePosixPath from typing import TYPE_CHECKING, Any -import re - from agentenv_protocol.computer.apps import APPS from agentenv_protocol.computer.platform import Platform, supported from jsonschema import Draft202012Validator @@ -25,11 +26,11 @@ MAX_SIDE = 32767 -def check(base: ComputerArtifact, driver: ComputerDriverEnv, *, apps: list, display: Any, network: Any, - computer_settings: Any) -> tuple[dict[str, SettingValue], list[str]]: - """The env's values of the base's settings, and every way ``base``, ``driver`` and what the env declares don't - compose.""" - problems = _platform(base, driver) + _apps(base, driver, apps) +def check(base: ComputerArtifact, driver: ComputerDriverEnv, *, environment_name: str, apps: list, display: Any, + network: Any, computer_settings: Any) -> tuple[dict[str, SettingValue], list[str]]: + """The env's values of the base's settings, and every way ``base``, ``driver`` and what the env declares, named + ``environment_name``, don't compose.""" + problems = _platform(base, driver) + _apps(base, driver, apps, environment_name) problems += _image(base) problems += _display(base, display, computer_settings) problems += _network(network) @@ -80,16 +81,20 @@ def app_folder(app: ComputerAppEnv) -> str: return folder_name(app.environment_name, "app") -def _apps(base: ComputerArtifact, driver: ComputerDriverEnv, apps: list[ComputerAppEnv]) -> list[str]: - """Every way ``apps`` don't fit together on ``base`` and ``driver``: each one's own (``check_app``), no architecture - they all run, one listed twice, and two under one environment name, in one folder, with one program name or - listening on one port, each pair of apps named for the first they'd share.""" +def _apps(base: ComputerArtifact, driver: ComputerDriverEnv, apps: list[ComputerAppEnv], + environment_name: str) -> list[str]: + """Every way ``apps`` don't fit together on ``base`` and ``driver`` in a computer env named ``environment_name``: + each one's own (``check_app``), no architecture they all run, one listed twice, one named as the computer is, two + under one environment name, in one folder, with one program name or listening on one port, each pair of apps named + for the first they'd share, and two keeping their data in one path or one inside the other.""" problems = [problem for i, app in enumerate(apps) for problem in check_app(base, driver, app, f"apps[{i}]")] if not problems and apps and not arches(base, driver, apps): problems.append(f"apps: no architecture {base.id}, {driver.id} and every app all run") ids = [app.id for app in apps] problems += [f"apps: {app_id} is listed twice" for app_id in sorted({i for i in ids if ids.count(i) > 1})] distinct, named = list({app.id: app for app in apps}.values()), set() + problems += [f"apps: {app.id} is named {environment_name!r}, as the computer env is, which names the computer's " + "own files" for app in distinct if app.environment_name == environment_name] for kind, keys in (("environment name", lambda app: [app.environment_name]), ("folder", lambda app: [f"{APPS}/{app_folder(app)}"]), ("program", lambda app: programs(app, base.os.type)), @@ -102,6 +107,10 @@ def _apps(base: ComputerArtifact, driver: ComputerDriverEnv, apps: list[Computer if len(owners) > 1 and frozenset(owners) not in named: named.add(frozenset(owners)) problems.append(f"apps: {' and '.join(owners)} would share the {kind} {key!r}") + owned = [(app, PurePosixPath(path)) for app in distinct for path in _paths(app, base.os.type)] + problems += [f"apps: {one.id} keeps its data in {str(a)!r} and {other.id} in {str(b)!r}, so they'd share it" + for (one, a), (other, b) in itertools.combinations(owned, 2) + if one is not other and (a.is_relative_to(b) or b.is_relative_to(a))] return problems @@ -112,6 +121,11 @@ def programs(app: ComputerAppEnv, os_type: str) -> list[str]: return [server, *(f"{server}-{service.name}" for service in (install.services if install is not None else []))] +def _paths(app: ComputerAppEnv, os_type: str) -> list[str]: + declared = app.declaration.os.get(os_type) + return list(declared.paths or []) if declared is not None else [] + + def _ports(app: ComputerAppEnv, os_type: str) -> list[int]: install = app.install.get(os_type) return list(install.ports or []) if install is not None else [] diff --git a/src/agent_env/env/envs/computer/utils/inside.py b/src/agent_env/env/envs/computer/utils/inside.py index f1033e23c..7be9f394f 100644 --- a/src/agent_env/env/envs/computer/utils/inside.py +++ b/src/agent_env/env/envs/computer/utils/inside.py @@ -36,7 +36,7 @@ def missing(deployed: DeployedSandboxEnv) -> list[str]: """What the env's card doesn't serve of computer-exec/v1 ``run``, ``data/add`` and ``data/get``.""" - operations = ((deployed.environment_card or {}).get("capabilities") or {}).get("operations") + operations = protocol_v1.operations(deployed.environment_card or {}) absent = [] if deployed.supports(COMPUTER_EXEC, "run") else [f"{COMPUTER_EXEC} run"] return absent + [op for op in (METHOD_ADD, METHOD_GET) if operations is not None and op not in operations] @@ -52,8 +52,10 @@ def loopback(deployed: DeployedSandboxEnv) -> dict: return {"id": deployed.env_id, "url": url, "mcp_url": url + deployed.mcp_url.removeprefix(deployed.environment_url)} -async def add_artifacts(deployed: DeployedSandboxEnv, files: dict[str, FileArtifact]) -> None: - """Write each file artifact at its path.""" +async def add_artifacts(deployed: DeployedSandboxEnv, files: dict[str, FileArtifact], + url: Optional[str] = None) -> None: + """Write each file artifact at its path, through the data plane at ``url``, the driver's or an app's it fronts; the + driver's by default.""" network = (deployed.metadata or {}).get("computer", {}).get("network") granted = not (network and NetworkPolicy.from_dict(network).restricts_egress) batch: list[FilePart] = [] @@ -61,12 +63,12 @@ async def add_artifacts(deployed: DeployedSandboxEnv, files: dict[str, FileArtif for path, artifact in files.items(): part, part_size = await asyncio.to_thread(_part, path, artifact, deployed.sandbox_type, granted) if batch and size + part_size > BATCH_BYTES: - await _add(deployed, batch, size) + await _add(deployed, batch, size, url) batch, size = [], 0 batch.append(part) size += part_size if batch: - await _add(deployed, batch, size) + await _add(deployed, batch, size, url) async def add_files(deployed: DeployedSandboxEnv, files: dict[str, bytes]) -> None: @@ -98,9 +100,9 @@ async def run(deployed: DeployedSandboxEnv, argv: list[str], *, timeout_s: float raise _broken(deployed, "computer-exec/v1 run", e) from e -async def _add(deployed: DeployedSandboxEnv, parts: list[FilePart], size: int) -> None: +async def _add(deployed: DeployedSandboxEnv, parts: list[FilePart], size: int, url: Optional[str] = None) -> None: try: - await protocol_v1.add_data(deployed.environment_url, parts, timeout=data_plane_load_timeout_s(size)) + await protocol_v1.add_data(url or deployed.environment_url, parts, timeout=data_plane_load_timeout_s(size)) except httpx.TransportError as e: raise _broken(deployed, "data/add, so some of the files it carried may be written, each whole", e) from e diff --git a/src/agent_env/task_step/task_steps/add_skills.py b/src/agent_env/task_step/task_steps/add_skills.py index ccb16b8a9..1c9505431 100644 --- a/src/agent_env/task_step/task_steps/add_skills.py +++ b/src/agent_env/task_step/task_steps/add_skills.py @@ -326,9 +326,9 @@ def _stored_skill(skill_md: str, files: dict[str, str]) -> str: async def _env_skills(context: TaskStepContext, agent: DeployedAgent, env_id: str) -> list[dict]: - """The skills the env ``env_id`` serves, each under its card's environment name: its own card's, or else each of its - child envs' (a gateway's servers); refused for an env this run didn't deploy or bind to ``agent``, or that serves - none.""" + """The skills the env ``env_id`` serves, each under its card's environment name: its own card's and each of its + child envs' (a gateway's servers, a computer's apps) its name reaches; refused for an env this run didn't deploy or + bind to ``agent``, or that serves none.""" deployed = next((d for d in context.deployed_envs if d.env_id == env_id), None) if deployed is None: raise RuntimeError(f"env '{env_id}' isn't deployed in this run") @@ -336,12 +336,12 @@ async def _env_skills(context: TaskStepContext, agent: DeployedAgent, env_id: st raise RuntimeError(f"env '{env_id}' isn't bound to agent '{agent.agent_name}': name it in the env_ids of its " "deploy_agent") card = deployed.environment_card or {} - if find_extension_method(card, SKILL_EXTENSION_URI, "get") is not None: - serving = [(None, _env_skill_name(card.get("name") or env_id, env_id))] - else: - serving = [(child["name"], _env_skill_name(child["name"], f"{env_id}/{child['name']}")) - for child in card.get("children_environments") or [] - if find_extension_method(child, SKILL_EXTENSION_URI, "get") is not None] + serving = [(None, _env_skill_name(card.get("name") or env_id, env_id))] if find_extension_method( + card, SKILL_EXTENSION_URI, "get") is not None else [] + serving += [(child["name"], _env_skill_name(child["name"], f"{env_id}/{child['name']}")) + for child in card.get("children_environments") or [] + if find_extension_method(child, SKILL_EXTENSION_URI, "get") is not None + and deployed.get_child_env_card(child["name"]) is child] if not serving: raise RuntimeError(f"env '{env_id}' serves no skill: neither its card nor a child env's declares " f"{SKILL_EXTENSION_URI}") diff --git a/src/agent_env/task_step/task_steps/computer/snapshot_computer_env.py b/src/agent_env/task_step/task_steps/computer/snapshot_computer_env.py index f28c08bb3..ffa6489c1 100644 --- a/src/agent_env/task_step/task_steps/computer/snapshot_computer_env.py +++ b/src/agent_env/task_step/task_steps/computer/snapshot_computer_env.py @@ -1,5 +1,5 @@ -"""snapshot_computer_env: a computer env's files, kept by their paths in its environment artifact, which load_artifact -loads back into a computer.""" +"""snapshot_computer_env: a computer env's files, kept by their paths in its environment artifact, and each of its apps' +exports in theirs, which load_artifact loads back into a computer.""" from __future__ import annotations @@ -24,7 +24,8 @@ class SnapshotComputerEnvTaskStep(SnapshotEnvTaskStep): """Snapshot a deployed computer env's files into a new environment universe, as ``snapshot_env`` snapshots an env, each file under its path relative to the session user's home in the computer's environment artifact: those under - ``file_paths``, by default the folders its card declares. ``load_artifact`` loads the snapshot back into a + ``file_paths``, by default the folders its card declares, leaving out each app's paths. Each app that exports is + kept beside it in an environment artifact under the app's name. ``load_artifact`` loads the snapshot back into a computer, through ``data/add``. Output: ``context.metadata["env_snapshotted_universes"][self.id]`` = ``{"id", "version"}``.""" @@ -97,7 +98,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: result = await self.snapshot_env_state( env=env, gateway_url=None, snapshot_id=snapshot_id, deployed=deployed, export_timeout_seconds=self.export_timeout_seconds, - data_get_params=None if self.file_paths is None else {"paths": self.file_paths}) + data_get_params=None if self.file_paths is None else {env.environment_name: {"paths": self.file_paths}}) context.metadata.setdefault("env_snapshotted_universes", {})[self.id] = { "id": result.environment_universe_artifact_id, "version": result.environment_universe_artifact_version} return context diff --git a/src/agent_env/task_step/task_steps/env_card_validator/verify_env_core_protocol.py b/src/agent_env/task_step/task_steps/env_card_validator/verify_env_core_protocol.py index 0f5f02144..16adf5090 100644 --- a/src/agent_env/task_step/task_steps/env_card_validator/verify_env_core_protocol.py +++ b/src/agent_env/task_step/task_steps/env_card_validator/verify_env_core_protocol.py @@ -7,6 +7,7 @@ import httpx from agentenv_protocol import METHOD_ADD, METHOD_GET, METHOD_RESET, RPC_PATH +from agentenv_protocol import client as protocol_v1 from agent_env.task_step.context import TaskStepContext from agent_env.entity.refs import EntityRef @@ -49,7 +50,6 @@ def from_dict(cls, data: dict) -> VerifyCoreEnvironmentProtocolStep: return cls(**cls._base_from_dict(data), env_id=data["env_id"]) async def execute(self, context: TaskStepContext) -> TaskStepContext: - from agentenv_protocol import client as protocol_v1 from agent_env.env.env import Env deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None) @@ -87,7 +87,7 @@ def _advertised_support(card: Optional[dict], method: str) -> Optional[bool]: backing = children[0] if len(children) == 1 else (card if not children else None) if backing is None: return None - advertised = (backing.get("capabilities") or {}).get("operations") + advertised = protocol_v1.operations(backing) return True if advertised is None else method in advertised async def _probe_operation(self, environment_url: str, method: str, params: dict) -> Optional[bool]: 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 6bd8d2055..4b082f3ef 100644 --- a/src/agent_env/task_step/task_steps/snapshot_env.py +++ b/src/agent_env/task_step/task_steps/snapshot_env.py @@ -18,7 +18,7 @@ import httpx -from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI, FilePart, uploaded_object_path +from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI, METHOD_GET, FilePart, uploaded_object_path from agentenv_protocol import client as protocol_v1 from agentenv_protocol.client import GetDataResponse, find_extension, find_extension_method, intake_declaration from agentenv_protocol.transfers import WriteNamespaceGrant @@ -169,8 +169,10 @@ class SnapshotEnvTaskStep(TaskStep): ``include_env_changes`` also stores who changed what (urn:agentenv:changes/v1) under ``metadata["env_changes"]``. An env deployed without a gateway is reached through the card its deployment recorded, its environments the - card's children, or itself. An environment whose data/get returns a file tree, a computer's, is kept file by file, - each by its path, in its environment artifact. + card's children, and itself first when it exports beside them, as a computer fronting its apps does, or itself + alone. An + environment whose data/get returns a file tree, a computer's, is kept file by file, each by its path, in its + environment artifact. """ type: ClassVar[str] = "snapshot_env" @@ -588,13 +590,14 @@ async def snapshot_env_state( original_universe_artifact_id: Optional[str] = None, export_timeout_seconds: int = DEFAULT_EXPORT_TIMEOUT_SECONDS, concurrency: int = DEFAULT_EXPORT_CONCURRENCY, - data_get_params: Optional[dict] = None, + data_get_params: Optional[dict[str, dict]] = None, ) -> EnvSnapshotResult: """Export all services via the gateway, or without one (``gateway_url`` None) through ``deployed``'s recorded card, into a new EnvironmentUniverseArtifact ``snapshot_id``, - handing each service's data/get ``data_get_params``. + handing each service's data/get its ``data_get_params``, by its name. All-or-nothing; the raised error carries public per-service summaries (verbose details - go to the log). + go to the log). A file tree with no files, an app that keeps nothing yet, is left out, + unless every environment's is. A classmethod so callers that have no step instance can reuse it.""" from agent_env.artifact import FileArtifact, EnvironmentArtifact, EnvironmentUniverseArtifact @@ -645,7 +648,7 @@ async def export_one(environment_name: str): try: result = await cls._export_environment_to_file( gateway, environment_name, tmp_path, export_timeout_seconds, - deployed=card_record, data_get_params=data_get_params, + deployed=card_record, data_get_params=(data_get_params or {}).get(environment_name), ) except httpx.HTTPStatusError as e: body_snippet = "" @@ -665,6 +668,8 @@ async def export_one(environment_name: str): try: files = await _file_tree_files(*result, environment_name, f"{snapshot_id}-{environment_name}", f"state snapshot of {environment_name} from env {env.id}") + if not files: + return None return await asyncio.to_thread( EnvironmentArtifact.put, id=f"{snapshot_id}-{environment_name}", environment_name=environment_name, file_artifacts=files) @@ -738,7 +743,7 @@ def _register_uploaded_bundle(): for name, result in zip(services, results): if isinstance(result, tuple): skipped[name] = result - else: + elif result is not None: environment_artifacts.append(result) if skipped: @@ -754,6 +759,10 @@ 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") + logger.info(f"snapshot_env: {empty} returned no files to keep, so {snapshot_id} leaves them out") universe = await asyncio.to_thread( EnvironmentUniverseArtifact.put, @@ -816,10 +825,7 @@ async def written(part) -> tuple[str, FileArtifact]: return path, await asyncio.to_thread(registered_upload, upload.object_url(path), file_id, filename, content_type, about) - files = dict(await asyncio.gather(*(written(part) for part in resp.parts))) - if not files: - raise ValueError(f"'{environment_name}' returned no files to snapshot") - return files + return dict(await asyncio.gather(*(written(part) for part in resp.parts))) def registered_upload(object_url: str, file_id: str, filename: str, content_type: str, description: str): @@ -836,11 +842,19 @@ def registered_upload(object_url: str, file_id: str, filename: str, content_type def _card_environments(deployed) -> list[str]: - """The environments a deployment without a gateway serves, as the card it recorded names them: its children, or - the env itself.""" + """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") - names = [child.get("name") for child in children] if isinstance(children, list) else [card.get("name")] + 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 diff --git a/tst/integration/computer/app/artifacts/desk_seed/Desktop/todo.txt b/tst/integration/computer/app/artifacts/desk_seed/Desktop/todo.txt new file mode 100644 index 000000000..b9feddc5f --- /dev/null +++ b/tst/integration/computer/app/artifacts/desk_seed/Desktop/todo.txt @@ -0,0 +1 @@ +water the plants diff --git a/tst/integration/computer/app/artifacts/desk_seed/Documents/seed.txt b/tst/integration/computer/app/artifacts/desk_seed/Documents/seed.txt new file mode 100644 index 000000000..c19e887d6 --- /dev/null +++ b/tst/integration/computer/app/artifacts/desk_seed/Documents/seed.txt @@ -0,0 +1 @@ +seeded diff --git a/tst/integration/computer/app/artifacts/desk_seed/artifact.toml b/tst/integration/computer/app/artifacts/desk_seed/artifact.toml new file mode 100644 index 000000000..10e9b04f5 --- /dev/null +++ b/tst/integration/computer/app/artifacts/desk_seed/artifact.toml @@ -0,0 +1,2 @@ +type = "environment" +environment_name = "desk" diff --git a/tst/integration/computer/app/artifacts/notes_seed/Notes/ideas.md b/tst/integration/computer/app/artifacts/notes_seed/Notes/ideas.md new file mode 100644 index 000000000..a8707ac1e --- /dev/null +++ b/tst/integration/computer/app/artifacts/notes_seed/Notes/ideas.md @@ -0,0 +1 @@ +# Ideas diff --git a/tst/integration/computer/app/artifacts/notes_seed/Notes/today.md b/tst/integration/computer/app/artifacts/notes_seed/Notes/today.md new file mode 100644 index 000000000..6af58f06a --- /dev/null +++ b/tst/integration/computer/app/artifacts/notes_seed/Notes/today.md @@ -0,0 +1 @@ +# Today diff --git a/tst/integration/computer/app/artifacts/notes_seed/artifact.toml b/tst/integration/computer/app/artifacts/notes_seed/artifact.toml new file mode 100644 index 000000000..c2da269f8 --- /dev/null +++ b/tst/integration/computer/app/artifacts/notes_seed/artifact.toml @@ -0,0 +1,2 @@ +type = "environment" +environment_name = "notes" diff --git a/tst/integration/computer/app/artifacts/reset_apps/run.py b/tst/integration/computer/app/artifacts/reset_apps/run.py new file mode 100644 index 000000000..fedff288f --- /dev/null +++ b/tst/integration/computer/app/artifacts/reset_apps/run.py @@ -0,0 +1,17 @@ +"""Each app's data/reset, sent from inside the computer through the driver's route to it, and what the apps keep +after: the notes in ~/Notes, and whether the clock's alarms are there.""" +import json +import urllib.request +from pathlib import Path + + +def run(input): + answers = {} + for app in ("notes", "clock"): + request = urllib.request.Request(f"{input['env']['url']}/svc/{app}/agentenv", data=json.dumps( + {"jsonrpc": "2.0", "id": 1, "method": "data/reset", "params": {}}).encode(), + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(request, timeout=60) as response: + answers[app] = "error" not in json.load(response) + return {"reset": answers, "notes": sorted(p.name for p in (Path.home() / "Notes").iterdir()), + "alarms": (Path.home() / ".clock-alarms.json").exists()} diff --git a/tst/integration/computer/app/artifacts/seed/artifact.toml b/tst/integration/computer/app/artifacts/seed/artifact.toml new file mode 100644 index 000000000..dd144e242 --- /dev/null +++ b/tst/integration/computer/app/artifacts/seed/artifact.toml @@ -0,0 +1,2 @@ +type = "environment_universe" +environment_artifacts = ["desk_seed", "notes_seed"] diff --git a/tst/integration/computer/app/artifacts/seed/clock/alarms.json b/tst/integration/computer/app/artifacts/seed/clock/alarms.json new file mode 100644 index 000000000..aa196a675 --- /dev/null +++ b/tst/integration/computer/app/artifacts/seed/clock/alarms.json @@ -0,0 +1 @@ +{"alarms": ["07:00"]} diff --git a/tst/integration/computer/app/envs/clock/app.py b/tst/integration/computer/app/envs/clock/app.py index 0aec9d802..54289b02f 100644 --- a/tst/integration/computer/app/envs/clock/app.py +++ b/tst/integration/computer/app/envs/clock/app.py @@ -1,11 +1,17 @@ """A clock app for the computer env tests, composed beside notes: its service answers on a port of its own with the time zone its settings name, and writes it to ~/.clock.json. It installs tzdata, which its run script lists the zones -of, and declares no window.""" +of, and declares no window. Its data plane is its own hooks over its alarms, kept in ~/.clock-alarms.json.""" +import json +from pathlib import Path + from pydantic import BaseModel -from agentenv_protocol.computer import ComputerAppEnvironment, Packages, RunScript, Service, computer_app, linux +from agentenv_protocol import DataPart, add_data, get_data, reset_data +from agentenv_protocol.computer import (ComputerAppEnvironment, Packages, RunScript, Service, computer_app, linux, + part_bytes) PORT = 8137 +ALARMS = ".clock-alarms.json" @computer_app(supports=[{"os": {"type": "linux", "display": ["x11"]}, "arch": ["amd64", "arm64"]}]) @@ -14,3 +20,17 @@ class Clock(ComputerAppEnvironment): class Settings(BaseModel): zone: str = "UTC" + + @add_data + async def load(self, parts: list) -> None: + for part in parts: + (Path.home() / ALARMS).write_bytes(await part_bytes(part)) + + @get_data + def alarms(self) -> list: + path = Path.home() / ALARMS + return [DataPart(data=json.loads(path.read_text()) if path.exists() else {"alarms": []})] + + @reset_data + def clear(self) -> None: + (Path.home() / ALARMS).unlink(missing_ok=True) diff --git a/tst/integration/computer/app/envs/notes/app.py b/tst/integration/computer/app/envs/notes/app.py index 11933a1b2..67ab9e55a 100644 --- a/tst/integration/computer/app/envs/notes/app.py +++ b/tst/integration/computer/app/envs/notes/app.py @@ -2,8 +2,9 @@ own server launches its window once the desktop session is up; it's ready once both have happened. It requires no Python of its own, so its service and window run in the venv every app gets for its server. Its install takes an apt package from Ubuntu's archive and one only a Launchpad PPA has, which its run script uses, a download checked by sha256, -and an install step run as the session user on the build setting its env.toml fixes. On Linux its settings gain a font, -and its install step and ready check are Linux's.""" +and an install step run as the session user on the build setting its env.toml fixes, which also leaves a welcome note +in ~/Notes, where it keeps its data, so its server serves that folder as its data plane. On Linux its settings gain a +font, and its install step and ready check are Linux's.""" from pathlib import Path from pydantic import BaseModel, SecretStr @@ -28,7 +29,7 @@ class LinuxSettings(BaseModel): services=[Service(name="notes", command="venv/bin/python serve.py")], launch=Launch(at="boot", command=["venv/bin/python", "window.py"], window={"class": "Notes"}, ready=Ready(window=True)), - settings=LinuxSettings) + paths=["Notes"], settings=LinuxSettings) class Notes(ComputerAppEnvironment): skill = "skill.md" @@ -42,6 +43,8 @@ class BuildSettings(BaseModel): @install_step(scope="session", os="linux") def remember(self) -> None: (Path.home() / ".notes-installed").write_text(self.build_settings.edition) + (Path.home() / "Notes").mkdir(exist_ok=True) + (Path.home() / "Notes" / "welcome.md").write_text("# Welcome\n") @ready_check(os="linux") def written(self) -> bool: diff --git a/tst/integration/computer/app/tasks/data.json b/tst/integration/computer/app/tasks/data.json new file mode 100644 index 000000000..167a26bd3 --- /dev/null +++ b/tst/integration/computer/app/tasks/data.json @@ -0,0 +1,9 @@ +[ + {"id": "env", "type": "deploy_env", "env_id": "desk", + "env_params": {"app_settings": {"notes": {"signature": "-- Ada", "token": "env:NOTES_TOKEN"}}}}, + {"id": "load", "type": "load_artifact", "env_id": "desk", "artifact_id": "seed"}, + {"id": "loaded", "type": "snapshot_computer_env", "env_id": "desk", "file_paths": ["Documents", "Desktop", "Notes"]}, + {"id": "reset", "type": "run_code", "env_id": "desk", "script_artifact_id": "reset_apps"}, + {"id": "after_reset", "type": "snapshot_computer_env", "env_id": "desk", + "file_paths": ["Documents", "Desktop", "Notes"]} +] diff --git a/tst/integration/computer/computer_app_test.py b/tst/integration/computer/computer_app_test.py index 6f50fab65..fefb9fb0d 100644 --- a/tst/integration/computer/computer_app_test.py +++ b/tst/integration/computer/computer_app_test.py @@ -5,15 +5,21 @@ one from a PPA, its download, run script and install step on its build setting, fetched at put, installed offline. And two apps composed into one computer env, each running on its own settings from its own folder, the files their run scripts wrote compared before the push of this machine's arch, or the image of the same key reused, as one an earlier -run pushed is. Needs Docker and the local registry on localhost:5000.""" +run pushed is. And the two apps' data planes through the computer: a snapshot loaded into it reaches each app by its +name, notes's paths and clock's hooks, the computer's snapshot keeps each app's export beside its own files, leaving the +apps' paths out of its own, and each app's data/reset, sent through the driver's route to it, puts the app back as +installed. Needs Docker and the local registry on localhost:5000.""" +import json import platform from pathlib import Path import pytest from click.testing import CliRunner +from agent_env.artifact import EnvironmentUniverseArtifact from agent_env.cli import cli +from agent_env.store.routing import namespace_routing from tst.integration.computer.conftest import run_context pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] @@ -58,3 +64,35 @@ def test_two_apps_composed_into_one_computer_env_each_run_on_their_own_settings( said = [line for line in ran.output.splitlines() if line.startswith(f"envs/desk: container/{ARCH}: ")] assert any(line.endswith(("comparing the files its parts' run scripts and install steps wrote", "reusing the image of the same key, built earlier")) for line in said), said + + +def test_a_computers_apps_load_export_and_reset_each_through_its_own_data_plane(state, monkeypatch): + monkeypatch.setenv("NOTES_TOKEN", "t0ken") + + ran = CliRunner().invoke(cli, ["run", str(APP), "--task", "data"]) + + assert ran.exit_code == 0, ran.output + context = run_context(ran.output) + (deployed,) = context["deployed_envs"] + assert sorted(child["name"] for child in deployed["environment_card"]["children_environments"]) == [ + "clock", "notes"] + snapshots = context["metadata"]["env_snapshotted_universes"] + desk = {"Documents/seed.txt": b"seeded\n", "Desktop/todo.txt": b"water the plants\n"} + assert _kept(snapshots["loaded"]) == {"desk": desk, "notes": { + "Notes/welcome.md": b"# Welcome\n", "Notes/today.md": b"# Today\n", "Notes/ideas.md": b"# Ideas\n"}, + "clock": {"alarms": ["07:00"]}} + assert context["metadata"]["script_results"]["reset"] == {"reset": {"notes": True, "clock": True}, + "notes": ["welcome.md"], "alarms": False} + assert _kept(snapshots["after_reset"]) == {"desk": desk, "notes": {"Notes/welcome.md": b"# Welcome\n"}, + "clock": {"alarms": []}} + + +def _kept(snapshot: dict) -> dict: + """Each environment of a snapshot: the computer's and notes's files by their paths, and the clock's export.""" + with namespace_routing(): + universe = EnvironmentUniverseArtifact.get(snapshot["id"], version=snapshot["version"]) + kept = {environment.environment_name: {path: file.load() for path, file in + environment.get_file_artifacts().items()} + for environment in universe.get_environment_artifacts()} + (clock,) = kept.pop("clock").values() + return {**kept, "clock": json.loads(clock)} diff --git a/tst/unit/bundle/computer_env_folder_test.py b/tst/unit/bundle/computer_env_folder_test.py index d01249337..33efd2499 100644 --- a/tst/unit/bundle/computer_env_folder_test.py +++ b/tst/unit/bundle/computer_env_folder_test.py @@ -157,6 +157,32 @@ def test_a_computer_env_folder_names_its_apps_in_order_each_written_before_it(tm assert composed == [(env.id, UBUNTU, REFERENCE, None)] +@pytest.mark.parametrize("stored", [False, True], ids=["in-the-bundle", "in-the-store"]) +def test_a_universe_loaded_into_a_computer_env_seeds_its_files_and_each_of_its_apps_by_their_folders_names( + tmp_path, composed, stored): + if stored: + plan = _plan(_bundle(tmp_path / "desk", WITH_APPS, APPS)) + materialize(plan, build_here=local_builds(plan)) + env_id, folders = f"{plan.bundle.bundle.id_root}/desktop", {} + into = f"env '{env_id}'" + else: + env_id, folders, into = "desktop", {"envs/desktop/env.toml": WITH_APPS, **APPS}, "envs/desktop" + + def loading(root, *seeded): + return layout(root, {**folders, "artifacts/world/artifact.toml": 'type = "environment_universe"\n', + **{f"artifacts/world/{name}/a.json": "{}" for name in ("desktop", "mail", *seeded)}, + "tasks/t.json": json.dumps([{"id": "load", "type": "load_artifact", "env_id": env_id, + "artifact_id": "world"}])}) + + _plan(loading(tmp_path / "seeds")) + with pytest.raises(BundleError) as refused: + _plan(loading(tmp_path / "typo", "mial")) + + assert refused.value.problems == ( + f"tasks/t.json: step 'load': artifacts/world/mial seeds an environment named 'mial', and {into} has no server " + "by that name ('desktop', 'mail', 'notes'); name it after the server it seeds, or leave it out",) + + @pytest.mark.parametrize("toml, problem", [ ('type = "computer"\ndriver = "reference"\n', 'computer is required: the base computer, such as "ubuntu-24.04"'), ('type = "computer"\ncomputer = "ubuntu-24.04"\n', 'driver is required: the driver, such as "reference"'), diff --git a/tst/unit/computer_home.py b/tst/unit/computer_home.py index ae4348d5d..1d9d8e3dc 100644 --- a/tst/unit/computer_home.py +++ b/tst/unit/computer_home.py @@ -1,5 +1,6 @@ """A computer at its driver's boundary, for steps that reach inside one: its home a folder on this machine, ``data/add`` -writing there, ``data/get`` reading there, and computer-exec/v1 ``run`` running commands there; each call recorded. +writing there, ``data/get`` reading there, and computer-exec/v1 ``run`` running commands there; each call recorded, +and the address each ``data/add`` is sent to. ``pkill`` matches nothing, so one test never ends another's processes on this machine.""" from __future__ import annotations @@ -27,6 +28,7 @@ class Computer: def __init__(self, home: Path): self.home = home self.calls: list[tuple] = [] + self.added_at: list[str] = [] self.add_fails: Exception | None = None def deployed(self, env_id: str = "desk", card: dict | None = None, **computer) -> DeployedSandboxEnv: @@ -43,6 +45,7 @@ def install(self, monkeypatch) -> Computer: async def add_data(self, base_url, parts, timeout=120, verify=True): self.calls.append(("data/add", [p.metadata[DATA_OBJECTS_EXTENSION_URI] for p in parts], timeout)) + self.added_at.append(base_url) if self.add_fails is not None: raise self.add_fails for part in parts: diff --git a/tst/unit/env/envs/_computer.py b/tst/unit/env/envs/_computer.py index 06c305cad..374d76484 100644 --- a/tst/unit/env/envs/_computer.py +++ b/tst/unit/env/envs/_computer.py @@ -325,11 +325,11 @@ async def create_vm(self, **size): def _app(base, driver, folder, compose=True, supports=None, install=None, settings=MAIL_SETTINGS, launch=None, - extra=None, steps=(), build=MAIL_BUILD, linux_settings=None, name="mail", **changes): + extra=None, steps=(), build=MAIL_BUILD, linux_settings=None, name="mail", paths=None, **changes): """A mail app on ``base`` and ``driver``, which serves IMAP from a venv of its requirements; ``extra`` files beside them, by name, ``steps``, its install steps by name, scope and the OS types they run on, ``build``, the schema of - its build settings, ``linux_settings``, that of its settings on Linux, and ``name``, its id's last part and its - environment name.""" + its build settings, ``linux_settings``, that of its settings on Linux, ``name``, its id's last part and its + environment name, and ``paths``, where it keeps its data on Linux.""" folder.mkdir(parents=True, exist_ok=True) (folder / "app.py").write_text("class Mail: ...\n") (folder / "requirements.txt").write_text("agentenv-framework-protocol\n") @@ -342,8 +342,9 @@ def _app(base, driver, folder, compose=True, supports=None, install=None, settin "supports": supports or [{"os": {"type": "linux", "display": ["x11"]}, "arch": ["amd64", "arm64"]}], "permissions": [], "settings": settings, "build_settings": build, "skill": None, "data": [], "install_steps": [{"name": name, "scope": scope, "os": os or None} for name, scope, *os in steps], - "ready_checks": [], "os": {"linux": {"launch": launch, "settings": linux_settings, "install": install or { - "python": "requirements.txt", "services": [{"name": "imap", "command": "venv/bin/python imap.py"}]}}}}) + "ready_checks": [], "os": {"linux": { + "launch": launch, "settings": linux_settings, "paths": paths, "install": install or { + "python": "requirements.txt", "services": [{"name": "imap", "command": "venv/bin/python imap.py"}]}}}}) return ComputerAppEnv.put(**{"id": f"@local/~/kit/{name}", "environment_name": name, "module": "app.py:Mail", "files": files, "declaration": declaration, "computer": base, "driver": driver, "compose": compose, **changes}) diff --git a/tst/unit/env/envs/test_computer_apps.py b/tst/unit/env/envs/test_computer_apps.py index 4306f6f6a..247573789 100644 --- a/tst/unit/env/envs/test_computer_apps.py +++ b/tst/unit/env/envs/test_computer_apps.py @@ -59,7 +59,13 @@ def test_a_computer_env_builds_its_apps_after_its_driver_in_the_order_it_lists_t (lambda mail, other: [other(name="mail", install={"services": [{"name": "sync", "command": "venv/bin/sync"}]}), other(name="mail-sync", install={"services": []})], "apps: @local/~/kit/mail and @local/~/kit/mail-sync would share the program 'app-mail-sync'"), -], ids=["twice", "environment-name", "folder", "app-port", "driver-port", "arch", "no-shared-arch", "program"]) + (lambda mail, other: [mail, other(name="desk-app", environment_name="desktop")], + "apps: @local/~/kit/desk-app is named 'desktop', as the computer env is, which names the computer's own files"), + (lambda mail, other: [other(name="a", paths=["Mail", ".config/a"]), other(name="b", paths=["Mail/inbox"]), + other(name="c", paths=[".config/c"])], + "apps: @local/~/kit/a keeps its data in 'Mail' and @local/~/kit/b in 'Mail/inbox', so they'd share it"), +], ids=["twice", "environment-name", "folder", "app-port", "driver-port", "arch", "no-shared-arch", "program", + "computer-name", "paths"]) def test_apps_that_dont_fit_together_or_on_the_computer_are_refused_before_anything_is_stored(base, driver, docker, tmp_path, apps, problem): def other(name="other", **changes): diff --git a/tst/unit/env/envs/test_computer_env.py b/tst/unit/env/envs/test_computer_env.py index 36c807f52..d1095a050 100644 --- a/tst/unit/env/envs/test_computer_env.py +++ b/tst/unit/env/envs/test_computer_env.py @@ -54,7 +54,7 @@ from agent_env.store.image_store.local_registry_image_store import LocalRegistryImageStore from agent_env.store.image_store.oci_registry_credentials import RegistryAuth from agent_env.store.object_store.local.store import LocalFilesystemObjectStore -from tst.unit.computer_home import Computer +from tst.unit.computer_home import CARD as HOME_CARD, Computer from tst.unit.env.envs._computer import (CONFIG, SERVE, CARD_EXTENSION, CARD, MAIL_SETTINGS, MAIL_BUILD, MAIL_SERVER, _app, _base, _driver, _put, FakeVmProvider) @@ -673,6 +673,25 @@ async def test_a_boot_that_never_gets_ready_takes_its_sandbox_and_work_folder_do assert local_sandbox.LocalSandbox.find_work_dir("local-unready") is None +@pytest.mark.asyncio +async def test_a_boot_is_ready_with_the_card_read_once_its_status_says_so_each_app_on_it(monkeypatch): + cards = iter([{"name": "desk"}, {"name": "desk"}, {"name": "desk", "children_environments": [{"name": "mail"}]}]) + statuses = iter([{"ready": False, "checks": [{"name": "app mail", "ok": False, "detail": "starting"}]}, + {"ready": True}]) + + async def card(url, timeout): + return next(cards) + + async def status(url, card, uri, params, timeout, method): + return next(statuses) + + monkeypatch.setattr(boot.protocol_v1, "get_card", card) + monkeypatch.setattr(boot.protocol_v1, "invoke_extension", status) + monkeypatch.setattr(boot, "_POLL_S", 0) + + assert await boot.ready("http://desk", 5) == {"name": "desk", "children_environments": [{"name": "mail"}]} + + def test_a_boot_mints_a_generated_setting_resolves_a_secret_its_record_withholds_both_and_a_secret_is_a_reference( base, monkeypatch): monkeypatch.setenv("DESK_KEY", "s3cret") @@ -718,12 +737,20 @@ def computer(tmp_path, monkeypatch): return Computer(tmp_path / "home").install(monkeypatch) -def _deployed_env(base, driver, computer, **record): - env = _put(base, driver, compose=False) - env._deployed = computer.deployed(env_id=env.id, **record) +def _deployed_env(base, driver, computer, apps=(), children=(), **record): + """The env of ``apps``, deployed on ``computer``, its card named after it and listing ``children``.""" + env = _put(base, driver, compose=False, apps=apps) + card = {**HOME_CARD, "name": env.environment_name, + **({"children_environments": list(children)} if children else {})} + env._deployed = computer.deployed(env_id=env.id, card=card, **record) return env +def _child(name: str, *operations: str) -> dict: + """An app's card on its computer's, as the driver fronts it.""" + return {"name": name, "url": f"/svc/{name}/agentenv", "capabilities": {"operations": list(operations)}} + + def _files(tmp_path, **contents): for name, data in contents.items(): (tmp_path / "files" / name).parent.mkdir(parents=True, exist_ok=True) @@ -793,33 +820,41 @@ def _snapshot(environment_name: str, contents: dict[str, bytes]) -> EnvironmentA @pytest.mark.asyncio -async def test_a_snapshot_of_its_files_loads_back_at_their_paths_leaving_out_other_environments( +async def test_a_snapshot_loads_back_its_files_then_each_apps_through_the_apps_route_leaving_out_other_environments( base, driver, docker, computer, tmp_path, monkeypatch, caplog): monkeypatch.setattr(inside, "issues_grants_to", lambda store, sandbox_type: False) - env = _deployed_env(base, driver, computer) + mail = _app(base, driver, tmp_path / "mail", compose=False) + env = _deployed_env(base, driver, computer, apps=[mail], children=[_child("mail", "data/add", "data/get")]) contents = {"Desktop/a.txt": b"a", "Documents/r/b.txt": b"b"} universe = EnvironmentUniverseArtifact.put("snap", environment_artifacts=[ - _snapshot(env.environment_name, contents), _snapshot("mail", {"inbox/1.json": b"{}"})]) + _snapshot("mail", {"Mail/inbox.txt": b"m"}), _snapshot(env.environment_name, contents), + _snapshot("notes", {"inbox/1.json": b"{}"})]) with caplog.at_level(logging.WARNING): await env.load_environment_universe_artifact(universe) assert {path: (computer.home / path).read_bytes() for path in contents} == contents + assert computer.added_at == ["http://127.0.0.1:20001", "http://127.0.0.1:20001/svc/mail"] + assert [markers for _, markers, _ in computer.calls][1] == [{"path": "Mail/inbox.txt"}] assert not (computer.home / "inbox").exists() - assert "leaving out ['mail']" in caplog.text + assert "leaving out ['notes']" in caplog.text @pytest.mark.asyncio -async def test_an_environment_not_named_after_the_computer_is_refused_before_anything_is_sent(base, driver, docker, - computer): - env = _deployed_env(base, driver, computer) - mail = _snapshot("mail", {"inbox/1.json": b"{}"}) - - with pytest.raises(ValueError, match="does not match env environment_name"): - await env.load_environment_artifact(mail) - with pytest.raises(RuntimeError, match=f"has no environment named '{env.environment_name}'"): - await env.load_environment_universe_artifact(EnvironmentUniverseArtifact.put("mail-only", - environment_artifacts=[mail])) +async def test_an_environment_named_after_neither_the_computer_nor_an_app_that_loads_is_refused_before_anything_is_sent( + base, driver, docker, computer, tmp_path): + mail = _app(base, driver, tmp_path / "mail", compose=False) + env = _deployed_env(base, driver, computer, apps=[mail], children=[_child("mail", "data/get")]) + notes, mailbox = _snapshot("notes", {"inbox/1.json": b"{}"}), _snapshot("mail", {"Mail/inbox.txt": b"m"}) + + with pytest.raises(ValueError, match="has no environment named 'notes': it has \\['desktop', 'mail'\\]"): + await env.load_environment_artifact(notes) + with pytest.raises(ValueError, match="'mail' in computer env '@local/~/kit/desktop' serves no data/add"): + await env.load_environment_universe_artifact(EnvironmentUniverseArtifact.put("both", environment_artifacts=[ + _snapshot(env.environment_name, {"Desktop/a.txt": b"a"}), mailbox])) + with pytest.raises(RuntimeError, match="has no environment named any of \\['desktop', 'mail'\\]"): + await env.load_environment_universe_artifact(EnvironmentUniverseArtifact.put("notes-only", + environment_artifacts=[notes])) assert computer.calls == [] diff --git a/tst/unit/env/test_env.py b/tst/unit/env/test_env.py index 57d8982aa..0f40fe177 100644 --- a/tst/unit/env/test_env.py +++ b/tst/unit/env/test_env.py @@ -120,9 +120,14 @@ def test_child_env_cards_come_from_the_card_by_name(composed): assert _record({**composed, "name": "slack", "children_environments": []}).get_child_env_card("slack") is None # composed, not a leaf -def test_a_leaf_card_is_its_own_only_child_env(): +def test_a_leaf_card_is_its_own_only_child_env_and_a_composed_one_with_a_data_plane_is_the_one_its_name_reaches(): assert _record(_LEAF).get_child_env_card("notes") is _LEAF assert _record(_LEAF).get_child_env_card("slack") is None + mail = {"name": "mail", "url": "/svc/mail/agentenv"} + desk = {"name": "desk", "capabilities": {"operations": ["data/get"]}, "children_environments": [mail]} + assert _record(desk).get_child_env_card("desk") is desk and _record(desk).get_child_env_card("mail") is mail + alone = {**desk, "children_environments": [{**mail, "name": "desk"}]} + assert _record(alone).get_child_env_card("desk") is alone @pytest.mark.asyncio diff --git a/tst/unit/task_step/add_skills_env_test.py b/tst/unit/task_step/add_skills_env_test.py index 8f059dd32..3b8001ea5 100644 --- a/tst/unit/task_step/add_skills_env_test.py +++ b/tst/unit/task_step/add_skills_env_test.py @@ -1,6 +1,7 @@ -"""``add_skills`` ``env_ids``: each env's skill, or each of its child envs' behind a gateway, written under that card's -environment name, installed on the agent and recorded in the run, a skill with files beside its SKILL.md sent as a -stored bundle; refused for an env the run didn't deploy or bind to the agent, or that serves no skill.""" +"""``add_skills`` ``env_ids``: each env's skill and each of its child envs' (a gateway's servers, a computer's apps) its +name reaches, written under that card's environment name, installed on the agent and recorded in the run, a skill with +files beside its SKILL.md sent as a stored bundle; refused for an env the run didn't deploy or bind to the agent, or +that serves no skill.""" import hashlib from types import SimpleNamespace @@ -107,6 +108,20 @@ async def test_an_env_behind_a_gateway_installs_each_child_envs_skill_under_the_ assert set(context.metadata[ENV_SKILLS_KEY]["solver"]["office"]) == {"mail", f"chat-room-{digest}"} +@pytest.mark.asyncio +async def test_a_computer_installs_its_own_skill_then_each_apps_and_an_app_alone_the_computers_its_name_reaches(agent): + desk = _env("desk", name="desktop", children=(("mail", (_SKILL,)), ("notes", ()))) + alone = _env("mail-alone", name="mail", children=(("mail", (_SKILL,)),)) + for computer in (desk, alone): + computer.environment_card["capabilities"]["operations"] = ["data/add", "data/get"] + + context = await _step("desk", "mail-alone").execute(_context(desk, alone, bound=("desk", "mail-alone"))) + + assert [(base_url, card) for base_url, card, *_ in agent.calls] == [ + ("http://desk", "desktop"), ("http://desk", "mail"), ("http://mail-alone", "mail")] + assert set(context.metadata[ENV_SKILLS_KEY]["solver"]["mail-alone"]) == {"mail"} + + @pytest.mark.asyncio async def test_envs_whose_names_come_out_alike_get_skills_of_their_own(agent): await _step("shop", "other").execute(_context(_env(name="my-shop"), _env("other", name="My_Shop"), diff --git a/tst/unit/task_step/snapshot_computer_env_test.py b/tst/unit/task_step/snapshot_computer_env_test.py index 0fb423c77..6c6085cd0 100644 --- a/tst/unit/task_step/snapshot_computer_env_test.py +++ b/tst/unit/task_step/snapshot_computer_env_test.py @@ -1,7 +1,8 @@ """snapshot_computer_env: a computer env's files kept by their paths in its environment artifact, in a new environment -universe, returned inline or uploaded under the grant and written where they were put, the computer asked again without -the grant when it fails under it, and file_paths sent to its data/get; refused for an env that isn't a computer, for -paths that aren't relative, and with no files to keep.""" +universe beside each of its apps' exports, returned inline or uploaded under the grant and written where they were put, +the computer asked again without the grant when it fails under it, file_paths sent to its data/get alone, and an app +that keeps nothing yet left out; refused for an env that isn't a computer, for paths that aren't relative, and with no +files to keep.""" from __future__ import annotations @@ -15,8 +16,8 @@ from agent_env.env.envs.computer.env import ComputerEnv from agent_env.task_step.registry import get_task_step_registry from agent_env.task_step.task_steps.computer.snapshot_computer_env import SnapshotComputerEnvTaskStep -from tst.unit.task_step.snapshot_env_without_gateway_test import (DATA_OBJECTS, UNREACHABLE, _computer, _inline, _record, - _serve) +from tst.unit.task_step.snapshot_env_without_gateway_test import (DATA_OBJECTS, FILE_TREE, UNREACHABLE, _computer, + _inline, _record, _serve) from tst.unit.task_step.test_snapshot_env import _ctx from tst.util.granting_object_store import GrantingObjectStore @@ -31,7 +32,6 @@ def __init__(self): class _Mail: id, version, type, environment_name = "mail", 1, "mcp_server", "mail" - async def _snapshot(monkeypatch, card: dict, env=None, **fields) -> EnvironmentArtifact: """The computer's environment artifact in the universe the step snapshots it into.""" monkeypatch.setattr("agent_env.env.env.Env.get", classmethod(lambda cls, id, version=None: env or _Desk())) @@ -78,6 +78,30 @@ def test_an_env_that_isnt_a_computer_is_refused_before_a_run(monkeypatch, stored if refused else []) +@pytest.mark.asyncio +async def test_a_computer_with_apps_keeps_its_files_beside_each_apps_export_sending_file_paths_to_itself_alone( + local_stores, monkeypatch): + card = {**_computer(), "children_environments": [ + {"name": name, "url": f"/svc/{name}/agentenv", "capabilities": { + "operations": ["data/add", "data/get"], "extensions": [FILE_TREE]}} for name in ("mail", "notes")]} + card["capabilities"] = {**card["capabilities"], "operations": ["data/add", "data/get"]} + sent = _serve(monkeypatch, {"/agentenv": lambda params: [_inline("Documents/a.txt", b"a")], + "/svc/mail/agentenv": lambda params: [_inline("Mail/inbox.txt", b"m")], + "/svc/notes/agentenv": lambda params: []}) + monkeypatch.setattr("agent_env.env.env.Env.get", classmethod(lambda cls, id, version=None: _Desk())) + context = _ctx(_record(card)) + + await SnapshotComputerEnvTaskStep(id="snap", version=None, env_id="desk", file_paths=["Documents"]).execute(context) + + snapshotted = context.metadata["env_snapshotted_universes"]["snap"] + universe = EnvironmentUniverseArtifact.get(snapshotted["id"], version=snapshotted["version"]) + assert {environment.environment_name: _kept(environment) for environment in universe.get_environment_artifacts() + } == {"desktop": {"Documents/a.txt": b"a"}, "mail": {"Mail/inbox.txt": b"m"}} + assert sorted((path, sorted(params)) for path, params in sent) == [("/agentenv", ["paths"]), + ("/svc/mail/agentenv", []), + ("/svc/notes/agentenv", [])] + + @pytest.mark.asyncio @pytest.mark.parametrize("file_paths, params", [(["Documents"], {"paths": ["Documents"]}), (None, {})], ids=["file-paths", "the-cards-folders"]) diff --git a/tst/unit/task_step/snapshot_env_without_gateway_test.py b/tst/unit/task_step/snapshot_env_without_gateway_test.py index d201c1114..a262d6407 100644 --- a/tst/unit/task_step/snapshot_env_without_gateway_test.py +++ b/tst/unit/task_step/snapshot_env_without_gateway_test.py @@ -1,7 +1,8 @@ """snapshot_env on an env deployed without a gateway: reached through the card its deployment recorded, its -environments the card's children or the env itself, each kept as one file as through a gateway; an environment whose -data/get returns a file tree, a computer's, kept file by file by its paths under a grant sized for a file tree; and an -env that fails under the upload grant asked again without it.""" +environments the card's children its names reach that export, and the env itself when it exports beside them, or the +env alone, each kept as one file as through a gateway and sent its own data/get params; an environment +whose data/get returns a file tree, a computer's, kept file by file by its paths under a grant sized for a file tree; +and an env that fails under the upload grant asked again without it.""" from __future__ import annotations @@ -69,9 +70,9 @@ def handle(request: httpx.Request) -> httpx.Response: return sent -async def _snapshot(card: dict) -> EnvironmentUniverseArtifact: +async def _snapshot(card: dict, **params) -> EnvironmentUniverseArtifact: result = await SnapshotEnvTaskStep.snapshot_env_state(env=_Desk(), gateway_url=None, snapshot_id="snap-d", - deployed=_record(card)) + deployed=_record(card), **params) return EnvironmentUniverseArtifact.get(result.environment_universe_artifact_id) @@ -94,6 +95,25 @@ async def test_each_child_on_the_recorded_card_is_reached_and_kept_as_one_file(l assert sorted(path for path, _ in sent) == ["/svc/mail/agentenv", "/svc/notes/agentenv"] +@pytest.mark.asyncio +async def test_a_card_exporting_beside_its_children_is_kept_first_with_its_own_params_leaving_out_what_doesnt_export( + local_stores, monkeypatch): + card = {"name": "desktop", "capabilities": {"operations": ["data/add", "data/get"], "extensions": []}, + "children_environments": [ + {"name": "mail", "url": "/svc/mail/agentenv", "capabilities": {"operations": ["data/get"]}}, + {"name": "notes", "url": "/svc/notes/agentenv", "capabilities": {"operations": ["data/add"]}}]} + sent = _serve(monkeypatch, {"/agentenv": lambda params: [{"kind": "data", "data": {"home": 1}}], + "/svc/mail/agentenv": lambda params: [{"kind": "data", "data": {"inbox": 2}}]}) + + universe = await _snapshot(card, data_get_params={"desktop": {"paths": ["Desktop"]}}) + alone = await _snapshot({**card, "children_environments": [ + {"name": "desktop", "url": "/svc/desktop/agentenv", "capabilities": {"operations": ["data/get"]}}]}) + + assert _kept(universe) == {"desktop": b'{"home": 1}', "mail": b'{"inbox": 2}'} + assert _kept(alone) == {"desktop": b'{"home": 1}'} + assert sent == [("/agentenv", {"paths": ["Desktop"]}), ("/svc/mail/agentenv", {}), ("/agentenv", {})] + + @pytest.mark.asyncio async def test_the_env_itself_is_reached_when_its_card_has_no_children(local_stores, monkeypatch): sent = _serve(monkeypatch, {"/agentenv": lambda params: [{"kind": "data", "data": {"rows": 1}}]})