From 863dcd0286744d9871ccba56a3f7962c481617b1 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Fri, 6 Nov 2026 16:05:04 -0800
Subject: [PATCH 1/5] feat(multi-env): child_envs on deploy_env, each child
 env's options by its name (FD-4930)

A deploy_env step on a multi env takes child_envs: each child env's options by its
environment name, as a step deploying that env alone would give them (its sandbox
fields and its env_params). The multi env checks them at save and at deploy: an
unknown name, an entry with other keys, and an entry for an MCP server or a website,
which run on the multi env's own sandbox, are refused, each problem under
child_envs.<name>. An env with no child envs refuses child_envs.

A bundle export drops each entry's sandbox_type as it drops the step's own, through a
field path that reaches every value of a table (child_envs{}.sandbox_type).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/agent_env/entity/paths.py                 | 20 +++--
 src/agent_env/env/env.py                      |  8 +-
 src/agent_env/env/envs/multi_env.py           | 35 +++++++-
 .../task_step/task_steps/deploy_env.py        | 18 +++-
 tst/unit/task_step/deploy_env_test.py         | 84 +++++++++++++++++++
 tst/unit/task_step/test_field_scopes.py       |  2 +-
 6 files changed, 148 insertions(+), 19 deletions(-)

diff --git a/src/agent_env/entity/paths.py b/src/agent_env/entity/paths.py
index 0f329bc9e..f10a72376 100644
--- a/src/agent_env/entity/paths.py
+++ b/src/agent_env/entity/paths.py
@@ -1,5 +1,5 @@
 """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
 
@@ -7,7 +7,7 @@
 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}]"))
diff --git a/src/agent_env/env/env.py b/src/agent_env/env/env.py
index 9c8446099..2afe8ee84 100644
--- a/src/agent_env/env/env.py
+++ b/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``:
diff --git a/src/agent_env/env/envs/multi_env.py b/src/agent_env/env/envs/multi_env.py
index b962101f4..668dc7bf7 100644
--- a/src/agent_env/env/envs/multi_env.py
+++ b/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(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:
diff --git a/src/agent_env/task_step/task_steps/deploy_env.py b/src/agent_env/task_step/task_steps/deploy_env.py
index 1f8047b01..a53a47d8b 100644
--- a/src/agent_env/task_step/task_steps/deploy_env.py
+++ b/src/agent_env/task_step/task_steps/deploy_env.py
@@ -26,7 +26,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 +48,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 +67,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 +92,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"] = self.child_envs
         return base
 
     @classmethod
@@ -109,6 +114,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]:
@@ -127,13 +133,13 @@ def preflight(self) -> list[str]:
             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
@@ -188,6 +194,10 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             if "env_params" not in params and not accepts_kwargs:
                 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 "child_envs" not in params and not accepts_kwargs:
+                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.
diff --git a/tst/unit/task_step/deploy_env_test.py b/tst/unit/task_step/deploy_env_test.py
index 874832e4a..4a5fd6c17 100644
--- a/tst/unit/task_step/deploy_env_test.py
+++ b/tst/unit/task_step/deploy_env_test.py
@@ -448,3 +448,87 @@ def test_deployment_settings_keep_their_positions():
     they did: the fifth positional is still ttl_seconds."""
     step = DeployEnvTaskStep("t-1.deploy_env", None, "art-1", 2, 60)
     assert (step.ttl_seconds, step.artifact_id, step.artifact_version) == (60, None, None)
+
+
+# --- child_envs: each child env's options by its name, which only an env with child envs takes ---
+
+
+def _multi_env():
+    from agent_env.env.envs.multi_env import MultiEnv
+    from agent_env.env.envs.website import WebsiteEnv
+
+    shop = WebsiteEnv(id="web-shop", version=1, backend_docker_image_artifact=MagicMock(),
+                      frontend_docker_image_artifact=MagicMock(), environment_name="shop")
+    return MultiEnv(id="office", version=1, mcp_server_envs=[_mcp_env("gateway")], website_envs=[shop])
+
+
+def test_child_envs_round_trip_and_are_written_only_when_set():
+    entries = {"desktop": {"cpu": 4, "env_params": {"form": "container"}}}
+    step = DeployEnvTaskStep(id="d", version=None, env_id="office", child_envs=entries)
+    assert DeployEnvTaskStep.from_dict(step.to_dict()).child_envs == entries
+    assert "child_envs" not in DeployEnvTaskStep(id="d", version=None, env_id="office").to_dict()
+
+
+def test_writing_a_step_out_for_elsewhere_drops_each_child_envs_sandbox_type_as_it_drops_the_steps():
+    from agent_env.entity.scopes import FieldScope, prune
+
+    data = DeployEnvTaskStep(id="d", version=None, env_id="office", sandbox_type="modal",
+                             child_envs={"desktop": {"sandbox_type": "modal_vm", "cpu": 4}}).to_dict()
+    left = prune(data, DeployEnvTaskStep.field_scopes, frozenset({FieldScope.TASK}))
+    assert "sandbox_type" not in data and data["child_envs"] == {"desktop": {"cpu": 4}}
+    assert {where for where, _ in left} == {"sandbox_type", "child_envs.desktop.sandbox_type"}
+    assert prune({"child_envs": ["desktop"]}, DeployEnvTaskStep.field_scopes, frozenset({FieldScope.TASK})) == []
+
+
+@pytest.mark.parametrize("kind, refusal", [
+    ("server", "env 'mcp-email' has env_provider_type 'server', which doesn't take child_envs"),
+    ("gateway", "env 'mcp-email' has env_provider_type 'gateway', which doesn't take child_envs"),
+    ("other", "other envs have no child envs"),
+])
+def test_preflight_refuses_child_envs_for_an_env_with_no_child_envs(kind, refusal):
+    env = _OtherEnv("other", 1) if kind == "other" else _mcp_env(kind)
+    assert _preflight(env, child_envs={"desktop": {"cpu": 4}}) == [f"deploy_env 'd': {refusal}"]
+
+
+_ENTRY = "an entry is a table of sandbox_type, cpu, memory_mb, disk_size_gb, env_params, not"
+_SHARED = "runs on the multi env's own sandbox, which the step's own fields set, so it takes no entry"
+
+
+@pytest.mark.parametrize("child_envs, problem", [
+    ({"desktop": {"cpu": 4}}, "child_envs.desktop: env 'office' has no child env named 'desktop'; its child envs are email, shop"),
+    ({"shop": {"cpu": 4}}, f"child_envs.shop: shop {_SHARED}"),
+    ({"email": {"env_params": {"form": "container"}}}, f"child_envs.email: email {_SHARED}"),
+    ({"email": {"ttl_seconds": 60, "gateway_mode": "consistent"}}, f"child_envs.email: {_ENTRY} gateway_mode, ttl_seconds"),
+    ({"email": "big"}, f"child_envs.email: {_ENTRY} a str"),
+    ({"email": {}, "shop": {}}, f"child_envs.email: email {_SHARED}; child_envs.shop: shop {_SHARED}"),
+    (["email"], "child_envs is a table of child env names, not list"),
+], ids=["unknown-name", "website", "mcp-server-params", "whole-env-fields", "not-a-table", "each-named", "not-a-table-of-names"])
+def test_preflight_names_the_child_env_each_problem_is_with(child_envs, problem):
+    assert _preflight(_multi_env(), child_envs=child_envs) == [f"deploy_env 'd': {problem}"]
+
+
+@pytest.mark.asyncio
+async def test_a_multi_env_deploy_refuses_child_envs_before_it_deploys_anything():
+    with patch("agent_env.env.envs.multi_env.deploy_through_provider") as deploy, \
+            pytest.raises(ValueError, match=f"child_envs.shop: shop {_SHARED}"):
+        await _multi_env().deploy(child_envs={"shop": {"cpu": 4}})
+    deploy.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_execute_passes_child_envs_to_a_deploy_that_takes_them():
+    entries = {"desktop": {"cpu": 4, "env_params": {"form": "container"}}}
+    step = DeployEnvTaskStep(id="d", version=None, env_id="art-1", env_version=2, child_envs=entries)
+    with patch("agent_env.env.env.Env") as Env, patch(_UPDATE_TARGET):
+        Env.get.return_value = _fake_env(_deployed_env(None))
+        await step.execute(TaskStepContext())
+    assert Env.get.return_value.deploy.call_args.kwargs["child_envs"] == entries
+
+
+@pytest.mark.asyncio
+async def test_execute_refuses_child_envs_for_an_env_whose_deploy_takes_none():
+    step = DeployEnvTaskStep(id="d", version=None, env_id="art-1", child_envs={"desktop": {"cpu": 4}})
+    env = _env_with_fixed_deploy(_deployed_env(None))
+    with patch("agent_env.env.env.Env") as Env, pytest.raises(ValueError, match="have no child envs"):
+        Env.get.return_value = env
+        await step.execute(TaskStepContext())
diff --git a/tst/unit/task_step/test_field_scopes.py b/tst/unit/task_step/test_field_scopes.py
index 54562a0eb..fb6f71c83 100644
--- a/tst/unit/task_step/test_field_scopes.py
+++ b/tst/unit/task_step/test_field_scopes.py
@@ -37,7 +37,7 @@
 def test_each_scoped_field_is_an_optional_parameter_of_its_step_type_or_one_the_base_scopes_for_every_step(cls):
     parameters = inspect.signature(cls.__init__).parameters
     for path in cls.field_scopes:
-        top = path.split(".")[0].removesuffix("[]")
+        top = path.split(".")[0].removesuffix("[]").removesuffix("{}")
         if top in parameters:
             assert parameters[top].default is not inspect.Parameter.empty, f"{cls.type} scopes {top}, which it requires"
         else:

From c4d66da7cf7561b086e257633ffd0eec49556496 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Fri, 6 Nov 2026 16:19:47 -0800
Subject: [PATCH 2/5] fix(multi-env): copy child_envs in to_dict and refuse
 them at save as a run does

to_dict() handed out the step's own child_envs, which a bundle export prunes in place,
so exporting stripped each child env's sandbox_type from the loaded step. It now
returns a copy.

preflight() asked the env's provider about child_envs, and a provider that takes every
option passed them, while execute() refused them for an env whose deploy() takes none.
preflight() now checks the env's deploy() as execute() does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../task_step/task_steps/deploy_env.py        | 13 +++++++--
 tst/unit/task_step/deploy_env_test.py         | 27 ++++++++++++++-----
 2 files changed, 31 insertions(+), 9 deletions(-)

diff --git a/src/agent_env/task_step/task_steps/deploy_env.py b/src/agent_env/task_step/task_steps/deploy_env.py
index a53a47d8b..94c93fedf 100644
--- a/src/agent_env/task_step/task_steps/deploy_env.py
+++ b/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
@@ -93,7 +94,7 @@ def to_dict(self) -> dict:
         base["metadata"] = self.metadata
         base["env_params"] = self.env_params
         if self.child_envs:
-            base["child_envs"] = self.child_envs
+            base["child_envs"] = copy.deepcopy(self.child_envs)  # a bundle export prunes the dict in place
         return base
 
     @classmethod
@@ -128,6 +129,8 @@ 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)
+            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,
@@ -195,7 +198,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 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 "child_envs" not in params and not accepts_kwargs:
+            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)
@@ -229,3 +232,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())
diff --git a/tst/unit/task_step/deploy_env_test.py b/tst/unit/task_step/deploy_env_test.py
index 4a5fd6c17..45046c9d0 100644
--- a/tst/unit/task_step/deploy_env_test.py
+++ b/tst/unit/task_step/deploy_env_test.py
@@ -480,14 +480,27 @@ def test_writing_a_step_out_for_elsewhere_drops_each_child_envs_sandbox_type_as_
     assert prune({"child_envs": ["desktop"]}, DeployEnvTaskStep.field_scopes, frozenset({FieldScope.TASK})) == []
 
 
-@pytest.mark.parametrize("kind, refusal", [
-    ("server", "env 'mcp-email' has env_provider_type 'server', which doesn't take child_envs"),
-    ("gateway", "env 'mcp-email' has env_provider_type 'gateway', which doesn't take child_envs"),
-    ("other", "other envs have no child envs"),
-])
-def test_preflight_refuses_child_envs_for_an_env_with_no_child_envs(kind, refusal):
+@pytest.mark.parametrize("kind", ["server", "gateway", "other"])
+def test_preflight_refuses_child_envs_for_an_env_with_no_child_envs(kind):
     env = _OtherEnv("other", 1) if kind == "other" else _mcp_env(kind)
-    assert _preflight(env, child_envs={"desktop": {"cpu": 4}}) == [f"deploy_env 'd': {refusal}"]
+    assert _preflight(env, child_envs={"desktop": {"cpu": 4}}) == [f"deploy_env 'd': {env.type} envs have no child envs"]
+
+
+def test_preflight_refuses_child_envs_at_save_behind_a_provider_that_takes_every_option():
+    class _TakesEverything:
+        async def deploy(self, env, sandbox_provider, **options): ...
+
+    with patch("agent_env.env.envs.mcp_server.provider_or_class", return_value=_TakesEverything):
+        assert _preflight(_mcp_env("plugin_kw"), child_envs={"desktop": {"cpu": 4}}) == [
+            "deploy_env 'd': mcp_server envs have no child envs"]
+
+
+def test_exporting_a_step_leaves_the_steps_own_child_envs_as_they_were():
+    from agent_env.entity.scopes import FieldScope, prune
+
+    step = DeployEnvTaskStep(id="d", version=None, env_id="office", child_envs={"desktop": {"sandbox_type": "modal_vm"}})
+    prune(step.to_dict(), DeployEnvTaskStep.field_scopes, frozenset({FieldScope.TASK}))
+    assert step.child_envs == {"desktop": {"sandbox_type": "modal_vm"}}
 
 
 _ENTRY = "an entry is a table of sandbox_type, cpu, memory_mb, disk_size_gb, env_params, not"

From 3f71a2b3e2f23d6908c7386133560e31085de905 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Fri, 6 Nov 2026 16:25:30 -0800
Subject: [PATCH 3/5] fix(deploy-env): refuse env_params at save when the env's
 deploy() takes none, as a run does

preflight() asked the env's provider about env_params, and a provider that takes every
option passed them, while execute() refused them for an env whose deploy() takes none:
a task saved and then failed at its run. preflight() now checks the env's deploy() for
env_params as it does for child_envs, with the words execute() raises.

execute() asks _takes() for every optional keyword it forwards, in place of the inline
signature checks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../task_step/task_steps/deploy_env.py        | 13 ++++---
 tst/unit/task_step/deploy_env_test.py         | 38 +++++++++++--------
 2 files changed, 30 insertions(+), 21 deletions(-)

diff --git a/src/agent_env/task_step/task_steps/deploy_env.py b/src/agent_env/task_step/task_steps/deploy_env.py
index 94c93fedf..a598a7343 100644
--- a/src/agent_env/task_step/task_steps/deploy_env.py
+++ b/src/agent_env/task_step/task_steps/deploy_env.py
@@ -129,6 +129,9 @@ 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
@@ -184,17 +187,15 @@ 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:
diff --git a/tst/unit/task_step/deploy_env_test.py b/tst/unit/task_step/deploy_env_test.py
index 45046c9d0..fa4bc2bf0 100644
--- a/tst/unit/task_step/deploy_env_test.py
+++ b/tst/unit/task_step/deploy_env_test.py
@@ -267,13 +267,30 @@ def test_preflight_passes_what_the_env_takes(kind, options):
     assert _preflight(env, **options) == []
 
 
-@pytest.mark.parametrize("kind, refusal", [
-    ("server", "env 'mcp-email' has env_provider_type 'server', which doesn't take env_params"),
-    ("other", "other envs take no env_params"),
-])
-def test_preflight_refuses_env_params_an_env_doesnt_take(kind, refusal):
+@pytest.mark.parametrize("kind", ["server", "gateway", "other"])
+def test_preflight_refuses_env_params_an_env_doesnt_take(kind):
     env = _OtherEnv("other", 1) if kind == "other" else _mcp_env(kind)
-    assert _preflight(env, env_params={"form": "container"}) == [f"deploy_env 'd': {refusal}"]
+    assert _preflight(env, env_params={"form": "container"}) == [f"deploy_env 'd': {env.type} envs take no env_params"]
+
+
+class _TakesEverything:
+    """A plugin provider whose deploy() takes every option."""
+
+    async def deploy(self, env, sandbox_provider, **options): ...
+
+
+@pytest.mark.parametrize("option, refusal", [
+    ({"env_params": {"form": "container"}}, "mcp_server envs take no env_params"),
+    ({"child_envs": {"desktop": {"cpu": 4}}}, "mcp_server envs have no child envs"),
+], ids=["env-params", "child-envs"])
+@pytest.mark.asyncio
+async def test_what_a_run_refuses_behind_a_provider_that_takes_every_option_is_refused_at_save_too(option, refusal):
+    step = DeployEnvTaskStep(id="d", version=None, env_id="mcp-email", **option)
+    with patch("agent_env.env.envs.mcp_server.provider_or_class", return_value=_TakesEverything), \
+            patch("agent_env.env.env.Env.get", return_value=_mcp_env("plugin_kw")):
+        assert step.preflight() == [f"deploy_env 'd': {refusal}"]
+        with pytest.raises(ValueError, match=refusal):
+            await step.execute(TaskStepContext())
 
 
 @pytest.mark.parametrize("kind", ["server", "gateway", "other"])
@@ -486,15 +503,6 @@ def test_preflight_refuses_child_envs_for_an_env_with_no_child_envs(kind):
     assert _preflight(env, child_envs={"desktop": {"cpu": 4}}) == [f"deploy_env 'd': {env.type} envs have no child envs"]
 
 
-def test_preflight_refuses_child_envs_at_save_behind_a_provider_that_takes_every_option():
-    class _TakesEverything:
-        async def deploy(self, env, sandbox_provider, **options): ...
-
-    with patch("agent_env.env.envs.mcp_server.provider_or_class", return_value=_TakesEverything):
-        assert _preflight(_mcp_env("plugin_kw"), child_envs={"desktop": {"cpu": 4}}) == [
-            "deploy_env 'd': mcp_server envs have no child envs"]
-
-
 def test_exporting_a_step_leaves_the_steps_own_child_envs_as_they_were():
     from agent_env.entity.scopes import FieldScope, prune
 

From ba6f25d0bc2eaeaa88f3176921fb0503a13276d6 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Fri, 6 Nov 2026 16:30:43 -0800
Subject: [PATCH 4/5] test(multi-env): an integration test for child_envs on
 the local backend

On a real multi env holding an MCP server, behind the local gateway: each entry the multi
env refuses (an MCP server's, an unknown name, another key) and env_params on the multi
env are refused at save and at the run with the same words, and the local provider makes
no sandbox for any of them; without them the same env deploys and serves its server's
list_items through the gateway's /mcp.

child_envs_refusal names the other keys of an entry as strings, so an entry whose keys are
of two types is refused rather than raising.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/agent_env/env/envs/multi_env.py           |  2 +-
 .../envs/multi_env_child_envs_local_test.py   | 81 +++++++++++++++++++
 tst/unit/task_step/deploy_env_test.py         |  4 +-
 3 files changed, 85 insertions(+), 2 deletions(-)
 create mode 100644 tst/integration/env/envs/multi_env_child_envs_local_test.py

diff --git a/src/agent_env/env/envs/multi_env.py b/src/agent_env/env/envs/multi_env.py
index 668dc7bf7..5b303c644 100644
--- a/src/agent_env/env/envs/multi_env.py
+++ b/src/agent_env/env/envs/multi_env.py
@@ -194,7 +194,7 @@ def child_envs_refusal(self, child_envs: Optional[dict]) -> str | None:
         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(set(entry) - set(CHILD_ENV_OPTIONS)) if isinstance(entry, dict) else []
+            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)}")
diff --git a/tst/integration/env/envs/multi_env_child_envs_local_test.py b/tst/integration/env/envs/multi_env_child_envs_local_test.py
new file mode 100644
index 000000000..6eff71625
--- /dev/null
+++ b/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")
diff --git a/tst/unit/task_step/deploy_env_test.py b/tst/unit/task_step/deploy_env_test.py
index fa4bc2bf0..f3b916f22 100644
--- a/tst/unit/task_step/deploy_env_test.py
+++ b/tst/unit/task_step/deploy_env_test.py
@@ -521,9 +521,11 @@ def test_exporting_a_step_leaves_the_steps_own_child_envs_as_they_were():
     ({"email": {"env_params": {"form": "container"}}}, f"child_envs.email: email {_SHARED}"),
     ({"email": {"ttl_seconds": 60, "gateway_mode": "consistent"}}, f"child_envs.email: {_ENTRY} gateway_mode, ttl_seconds"),
     ({"email": "big"}, f"child_envs.email: {_ENTRY} a str"),
+    ({"email": {1: "a", "b": 2}}, f"child_envs.email: {_ENTRY} 1, b"),
     ({"email": {}, "shop": {}}, f"child_envs.email: email {_SHARED}; child_envs.shop: shop {_SHARED}"),
     (["email"], "child_envs is a table of child env names, not list"),
-], ids=["unknown-name", "website", "mcp-server-params", "whole-env-fields", "not-a-table", "each-named", "not-a-table-of-names"])
+], ids=["unknown-name", "website", "mcp-server-params", "whole-env-fields", "not-a-table", "keys-of-two-types", "each-named",
+     "not-a-table-of-names"])
 def test_preflight_names_the_child_env_each_problem_is_with(child_envs, problem):
     assert _preflight(_multi_env(), child_envs=child_envs) == [f"deploy_env 'd': {problem}"]
 

From 8b3865441785054aa41aeca812a7d66e675c63bc Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Fri, 6 Nov 2026 18:40:58 -0800
Subject: [PATCH 5/5] test(drivers): the macos-harness fake writes its pid
 whole, so a cancelled call's test never reads it empty

test_a_cancelled_call_stops_the_command_it_was_running waits for the fake's pid file to
exist, then reads the pid. The fake wrote it with write_text, which creates the file
empty and then writes it, so on a busy runner the test could read '' (it failed #522's
unit shard with "invalid literal for int() with base 10: ''"). The fake now writes
pid.tmp and renames it into place, so the file exists only with the pid in it.

With the CPU saturated, polling the file as the test does read it empty 16 times in 600
with the old write and 0 in 600 with the new one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tst/unit/drivers/macos_harness_driver_test.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tst/unit/drivers/macos_harness_driver_test.py b/tst/unit/drivers/macos_harness_driver_test.py
index 081d411eb..f08861c49 100644
--- a/tst/unit/drivers/macos_harness_driver_test.py
+++ b/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: