feat(multi-env): child_envs on deploy_env, each child env's options by its name (FD-4930)

src/agent_env/entity/paths.py

@@ -1,13 +1,13 @@
 """Paths to fields of an entity's dict or toml: dotted keys, each with an optional ``[]`` for every element of a list
-(``env_ids[]``, ``artifacts[].id``)."""
+(``env_ids[]``, ``artifacts[].id``) or ``{}`` for every value of a table (``child_envs{}.sandbox_type``)."""
 
 from __future__ import annotations
 
 import re
 from collections.abc import Iterator
 from typing import Any
 
-PATH = re.compile(r"[A-Za-z_]\w*(\[\])?(\.[A-Za-z_]\w*(\[\])?)*")
+PATH = re.compile(r"[A-Za-z_]\w*(\[\]|\{\})?(\.[A-Za-z_]\w*(\[\]|\{\})?)*")
 
 
 def field_sites(data: dict, path: str) -> Iterator[tuple[str, dict, str]]:
@@ -24,12 +24,18 @@ def descend(node: Any, segments: list[str], trail: tuple[str, ...]) -> Iterator[
     if not segments:
         yield trail, node
         return
-    key, each = segments[0].removesuffix("[]"), segments[0].endswith("[]")
+    segment = segments[0]
+    key = segment.removesuffix("[]").removesuffix("{}")
     child = node.get(key) if isinstance(node, dict) else None
     if child is None:
         return
-    if not each:
+    if segment.endswith("[]"):
+        if isinstance(child, list):
+            for i, item in enumerate(child):
+                yield from descend(item, segments[1:], (*trail, f"{key}[{i}]"))
+    elif segment.endswith("{}"):
+        if isinstance(child, dict):
+            for name, item in child.items():
+                yield from descend(item, segments[1:], (*trail, key, str(name)))
+    else:
         yield from descend(child, segments[1:], (*trail, key))
-    elif isinstance(child, list):
-        for i, item in enumerate(child):
-            yield from descend(item, segments[1:], (*trail, f"{key}[{i}]"))

src/agent_env/env/env.py

@@ -409,10 +409,12 @@ async def deploy(self, **kwargs) -> DeployedEnv:
         raise NotImplementedError(f"{type(self).__name__} must implement deploy()")
 
     def deploy_refusal(self, *, sandbox_type: Optional[str] = None, env_params: Optional[dict] = None,
-                       **options) -> Optional[str]:
+                       child_envs: Optional[dict] = None, **options) -> Optional[str]:
         """Why deploy() would refuse these options before building anything, or None. A type that checks its options
-        overrides this; one that doesn't takes no ``env_params``."""
-        return f"{self.type} envs take no env_params" if env_params else None
+        overrides this; one that doesn't takes no ``env_params``, and has no child envs to give ``child_envs``."""
+        if env_params:
+            return f"{self.type} envs take no env_params"
+        return f"{self.type} envs have no child envs" if child_envs else None
 
     async def run_env_params(self, env_params: Optional[dict], deployed_envs: list[DeployedEnv]) -> Optional[dict]:
         """``env_params`` as a run's deploy_env passes them to deploy(), once the run has deployed ``deployed_envs``:

src/agent_env/env/envs/multi_env.py

@@ -53,6 +53,9 @@ def _snapshot_after_load_default() -> bool:
 from agent_env.attribution import Attribution
 
 
+# A child_envs entry's options: a deploy_env step's for that env alone, less the multi env's (lifetime, gateway, state).
+CHILD_ENV_OPTIONS = ("sandbox_type", "cpu", "memory_mb", "disk_size_gb", "env_params")
+
 # The servicedb's image line in a deploy's compose file, as the local Postgres state renders it.
 _SERVICE_DB_IMAGE = re.compile(rf"^(  {re.escape(DATABASE_SERVICE_NAME)}:\n    image: ).*$", re.MULTILINE)
 
@@ -161,7 +164,10 @@ def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict:
             ctx.refuse(problems)
         return fields
 
-    async def deploy(self, ttl_seconds: int = 10800, disk_size_gb: float = 10, gateway_mode: GatewayMode = GatewayMode.PERFORMANCE, cpu: float | None = None, memory_mb: int | None = None, sandbox_type: str | None = None, env_state_type: str | None = None, env_state_instance_id: str | None = None, artifact_id: str | None = None, artifact_version: int | None = None, *, attribution: Optional[Attribution] = None) -> DeployedEnv:
+    async def deploy(self, ttl_seconds: int = 10800, disk_size_gb: float = 10, gateway_mode: GatewayMode = GatewayMode.PERFORMANCE, cpu: float | None = None, memory_mb: int | None = None, sandbox_type: str | None = None, env_state_type: str | None = None, env_state_instance_id: str | None = None, artifact_id: str | None = None, artifact_version: int | None = None, *, attribution: Optional[Attribution] = None,
+                     child_envs: Optional[dict] = None) -> DeployedEnv:
+        if refusal := self.child_envs_refusal(child_envs):
+            raise ValueError(refusal)
         deployed_env = await deploy_through_provider(
             self, environment_name=None, ttl_seconds=ttl_seconds, sandbox_type=sandbox_type,
             disk_size_gb=disk_size_gb, gateway_mode=gateway_mode, cpu=cpu, memory_mb=memory_mb,
@@ -174,10 +180,31 @@ async def deploy(self, ttl_seconds: int = 10800, disk_size_gb: float = 10, gatew
         _hand_record_to_child_envs(self, deployed_env)
         return deployed_env
 
-    def deploy_refusal(self, *, sandbox_type: str | None = None, **options) -> str | None:
+    def deploy_refusal(self, *, sandbox_type: str | None = None, child_envs: Optional[dict] = None, **options) -> str | None:
         """Why deploy() would refuse these options before building anything, or None; raises, as deploy() does, for a type this process can't find.
-        ``sandbox_type`` is deploy()'s own, which no provider takes."""
-        return deploy_refusal(self, provider_or_class(self), options)
+        ``sandbox_type`` and ``child_envs`` are deploy()'s own, which no provider takes."""
+        return deploy_refusal(self, provider_or_class(self), options) or self.child_envs_refusal(child_envs)
+
+    def child_envs_refusal(self, child_envs: Optional[dict]) -> str | None:
+        """Why a deploy would refuse ``child_envs``, each child env's options by its environment name, or None."""
+        if not child_envs:
+            return None
+        if not isinstance(child_envs, dict):
+            return f"child_envs is a table of child env names, not {type(child_envs).__name__}"
+        names = list(dict.fromkeys(env.environment_name for env in (*self.mcp_server_envs, *self.website_envs)))
+        problems = []
+        for name, entry in child_envs.items():
+            extra = sorted(map(str, set(entry) - set(CHILD_ENV_OPTIONS))) if isinstance(entry, dict) else []
+            if name not in names:
+                problems.append(f"child_envs.{name}: env '{self.id}' has no child env named {name!r}; its child envs "
+                                f"are {', '.join(names)}")
+            elif not isinstance(entry, dict) or extra:
+                problems.append(f"child_envs.{name}: an entry is a table of {', '.join(CHILD_ENV_OPTIONS)}, not " +
+                                (', '.join(extra) if extra else f"a {type(entry).__name__}"))
+            else:
+                problems.append(f"child_envs.{name}: {name} runs on the multi env's own sandbox, which the step's own "
+                                "fields set, so it takes no entry")
+        return "; ".join(problems)
 
     @classmethod
     async def from_deployed_env(cls, deployed: DeployedEnv) -> MultiEnv:

src/agent_env/task_step/task_steps/deploy_env.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import copy
 import inspect
 import logging
 from collections.abc import Mapping
@@ -26,7 +27,7 @@ class DeployEnvTaskStep(TaskStep):
         EntityRef.artifact("artifact_id", version_field="artifact_version"),
     )
     field_scopes = {**TaskStep.field_scopes, "sandbox_type": FieldScope.USER, "env_state_type": FieldScope.USER,
-                    "env_state_instance_id": FieldScope.USER}
+                    "env_state_instance_id": FieldScope.USER, "child_envs{}.sandbox_type": FieldScope.USER}
 
     def __init__(
         self,
@@ -48,6 +49,7 @@ def __init__(
         artifact_id: Optional[str] = None,
         artifact_version: Optional[int] = None,
         env_params: Optional[dict] = None,
+        child_envs: Optional[dict[str, dict]] = None,
     ):
         super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
         self.env_id = env_id
@@ -66,6 +68,8 @@ def __init__(
         self.metadata = metadata or {}
         # The env type's own deploy options, which this step passes to its deploy() without reading them.
         self.env_params = env_params
+        # Each child env's deploy options by its name, as a step deploying that env alone would give them.
+        self.child_envs = child_envs
 
     def run_state(self, user_overrides: Mapping[str, Any]) -> tuple[str | None, str | None]:
         """The env state type and attached state instance a run with ``user_overrides`` deploys on: each the run's,
@@ -89,6 +93,8 @@ def to_dict(self) -> dict:
         base["env_state_instance_id"] = self.env_state_instance_id
         base["metadata"] = self.metadata
         base["env_params"] = self.env_params
+        if self.child_envs:
+            base["child_envs"] = copy.deepcopy(self.child_envs)  # a bundle export prunes the dict in place
         return base
 
     @classmethod
@@ -109,6 +115,7 @@ def from_dict(cls, data: dict) -> DeployEnvTaskStep:
             env_state_instance_id=data.get("env_state_instance_id"),
             metadata=dict(data.get("metadata") or {}),
             env_params=data.get("env_params"),
+            child_envs=data.get("child_envs"),
         )
 
     def preflight(self) -> list[str]:
@@ -122,18 +129,23 @@ def preflight(self) -> list[str]:
             return [f"deploy_env '{self.id}': {e}"]  # the words execute() would raise
         try:
             env = Env.get(self.env_id, self.env_version)
+            # The words execute() raises, whatever the env's provider would take.
+            if self.env_params and not _takes(env.deploy, "env_params"):
+                return [f"deploy_env '{self.id}': {env.type} envs take no env_params"]
+            if self.child_envs and not _takes(env.deploy, "child_envs"):
+                return [f"deploy_env '{self.id}': {env.type} envs have no child envs"]
             # As stored: run-time overrides aren't known at save. Every run passes an attribution, so the check does too, whatever it
             # will hold.
             refusal = env.deploy_refusal(ttl_seconds=self.ttl_seconds, disk_size_gb=self.disk_size_gb, gateway_mode=self.gateway_mode, cpu=self.cpu,
                                          memory_mb=self.memory_mb, env_state_type=self.env_state_type,
                                          env_state_instance_id=self.env_state_instance_id, attribution={},
-                                         sandbox_type=self.sandbox_type, **self._env_params_option())
+                                         sandbox_type=self.sandbox_type, **self._env_options())
         except (NotFoundError, ValueError) as e:  # missing, or a type or env_provider_type this process can't load
             return [f"deploy_env '{self.id}': env '{self.env_id}' can't be loaded: {e}"]
         return [f"deploy_env '{self.id}': {refusal}"] if refusal else []
 
-    def _env_params_option(self) -> dict:
-        return {"env_params": self.env_params} if self.env_params else {}
+    def _env_options(self) -> dict:
+        return {name: value for name, value in (("env_params", self.env_params), ("child_envs", self.child_envs)) if value}
 
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
         from agent_env.env.env import Env
@@ -175,19 +187,21 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         # sidecars attribute spend to the same key as the agent. Guard on the
         # signature — most envs' deploy() takes a fixed param list and would
         # TypeError on an unexpected kwarg.
-        params = inspect.signature(env.deploy).parameters
-        accepts_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
         model_api_key = user_overrides.get("litellm_api_key")
-        if model_api_key and ("litellm_api_key" in params or accepts_kwargs):
+        if model_api_key and _takes(env.deploy, "litellm_api_key"):
             deploy_kwargs["litellm_api_key"] = get_config().resolve_reference(model_api_key)
-        if self.artifact_id is not None and ("artifact_id" in params or accepts_kwargs):
+        if self.artifact_id is not None and _takes(env.deploy, "artifact_id"):
             deploy_kwargs["artifact_id"] = self.artifact_id
-        if self.artifact_version is not None and ("artifact_version" in params or accepts_kwargs):
+        if self.artifact_version is not None and _takes(env.deploy, "artifact_version"):
             deploy_kwargs["artifact_version"] = self.artifact_version
         if self.env_params:
-            if "env_params" not in params and not accepts_kwargs:
+            if not _takes(env.deploy, "env_params"):
                 raise ValueError(f"deploy_env '{self.id}': {env.type} envs take no env_params")
             deploy_kwargs["env_params"] = await env.run_env_params(self.env_params, context.deployed_envs)
+        if self.child_envs:
+            if not _takes(env.deploy, "child_envs"):
+                raise ValueError(f"deploy_env '{self.id}': {env.type} envs have no child envs")
+            deploy_kwargs["child_envs"] = self.child_envs
         deployed_env = await env.deploy(**deploy_kwargs)
         # create_instance runs inside env.deploy(), so an annotation set after it
         # returns has to be persisted explicitly to reach the instance record.
@@ -219,3 +233,9 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 instance_id=agent_info.get("instance_id") or deployed_env.instance_id,
             ))
         return context
+
+
+def _takes(deploy, name: str) -> bool:
+    """Whether ``deploy`` takes the keyword ``name``: it names it, or takes ``**kwargs``."""
+    params = inspect.signature(deploy).parameters
+    return name in params or any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values())

tst/integration/env/envs/multi_env_child_envs_local_test.py

@@ -0,0 +1,81 @@
+"""A multi env's deploy_env with child_envs on the local backend: an entry the multi env refuses is refused at save and at
+the run before any sandbox is made, and the same env deployed without child_envs serves its MCP server through the
+gateway. Requires a docker daemon; spins up a throwaway ``registry:2`` and skips if it can't start.
+"""
+
+import re
+import uuid
+
+import pytest
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+from agent_env.env.env import Env
+from agent_env.env.envs.mcp_server import MCPServerEnv
+from agent_env.env.envs.multi_env import MultiEnv
+from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandboxProvider
+from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep
+from tst.integration.env.envs.server_env_local_test import _put_items_image
+from tst.integration.store.local_parallel_deploy_test import _docker, _seed_bootstrap_envs, local_stack as local_stack
+
+pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow]
+
+_SHARED = "runs on the multi env's own sandbox, which the step's own fields set, so it takes no entry"
+
+
+@pytest.fixture
+def made(monkeypatch) -> list[str]:
+    """The local provider's sandbox-making calls, by name, each still made."""
+    calls: list[str] = []
+    for name in ("create_sandbox", "create_vm", "create_container"):
+        real = getattr(LocalSandboxProvider, name)
+
+        async def spy(self, *args, _real=real, _name=name, **kwargs):
+            calls.append(_name)
+            return await _real(self, *args, **kwargs)
+
+        monkeypatch.setattr(LocalSandboxProvider, name, spy)
+    return calls
+
+
+@pytest.mark.asyncio
+async def test_child_envs_the_multi_env_refuses_make_no_sandbox_and_without_them_it_deploys(local_stack, made):
+    _seed_bootstrap_envs()
+    uid = uuid.uuid4().hex[:8]
+    items = MCPServerEnv.put(id=f"child-items-{uid}", environment_name="items",
+                             docker_image_artifact=_put_items_image(f"child-items-{uid}"))
+    env = MultiEnv.put(id=f"child-multi-{uid}", mcp_server_envs=[items], name="suite")
+
+    refused = [
+        ({"child_envs": {"items": {"cpu": 2}}}, f"child_envs.items: items {_SHARED}"),
+        ({"child_envs": {"desktop": {"cpu": 2}}},
+         f"child_envs.desktop: env '{env.id}' has no child env named 'desktop'; its child envs are items"),
+        ({"child_envs": {"items": {"ttl_seconds": 60}}},
+         "child_envs.items: an entry is a table of sandbox_type, cpu, memory_mb, disk_size_gb, env_params, not ttl_seconds"),
+        ({"env_params": {"form": "container"}}, "multi envs take no env_params"),
+    ]
+    for options, problem in refused:
+        step = DeployEnvTaskStep(id="env", version=None, env_id=env.id, sandbox_type="local", ttl_seconds=900, **options)
+        assert step.preflight() == [f"deploy_env 'env': {problem}"]
+        context = TaskStepContext()
+        with pytest.raises(ValueError, match=re.escape(problem)):
+            await step.execute(context)
+        assert context.deployed_envs == []
+    assert made == []
+
+    step = DeployEnvTaskStep(id="env", version=None, env_id=env.id, sandbox_type="local", ttl_seconds=900)
+    assert step.preflight() == []
+    [deployed] = (await step.execute(TaskStepContext())).deployed_envs
+    deployed_env = await Env.from_instance_id(deployed.instance_id)
+    work_dir = deployed_env._sandbox.work_dir
+    try:
+        assert made
+        async with streamable_http_client(deployed.mcp_url) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                assert "list_items" in [tool.name for tool in (await session.list_tools()).tools]
+                assert not (await session.call_tool("list_items", {})).isError
+        await deployed_env.close()
+    finally:
+        _docker("compose", "--project-directory", str(work_dir), "down", "-v", "--remove-orphans")

tst/unit/drivers/macos_harness_driver_test.py

@@ -34,7 +34,8 @@
         print("Window: \\"BenchLab\\", App: BenchLab.\\n\\t5 button Next\\n")
         print(f"Screenshot (2x1 px): {state / 'shot.jpeg'}")
     elif "slow" in sys.argv:
-        (state / "pid").write_text(str(os.getpid()))
+        (state / "pid.tmp").write_text(str(os.getpid()))
+        os.replace(state / "pid.tmp", state / "pid")  # whole, so a reader that sees the file sees the pid
         import time
         time.sleep(60)
     elif "fail" in sys.argv: