diff --git a/docs/v6/guides/running-an-eval.mdx b/docs/v6/guides/running-an-eval.mdx
index 15c309f3c..c080693a7 100644
--- a/docs/v6/guides/running-an-eval.mdx
+++ b/docs/v6/guides/running-an-eval.mdx
@@ -130,7 +130,7 @@ change to `env.py` or the tasks:
 
 | Runtime | Where the env runs |
 | --- | --- |
-| `LocalRuntime("env.py")` | A child process on your machine |
+| `LocalRuntime("env.py")` | In this process, on your machine |
 | `DockerRuntime("my-env")` | A fresh local container per rollout |
 | `ModalRuntime("my-env")` | A fresh [Modal](https://modal.com) sandbox per rollout |
 | `DaytonaRuntime("my-env")` | A fresh [Daytona](https://daytona.io) sandbox per rollout |
diff --git a/docs/v6/reference/cli.mdx b/docs/v6/reference/cli.mdx
index 6a5f51bbd..ca853a8a7 100644
--- a/docs/v6/reference/cli.mdx
+++ b/docs/v6/reference/cli.mdx
@@ -105,7 +105,7 @@ For a platform taskset, pass its name or id directly: `hud eval "My Tasks" claud
 | `--config`, `-c` | Agent config `key=value` (repeatable). |
 | `--verbose`, `-v` | Show agent logs (step progress, tool calls) for batch runs too. |
 | `--very-verbose`, `-vv` | Debug-level logs. |
-| `--runtime` | Placement: `local`, `hud` (HUD runtime tunnel), or `tcp://host:port`. Defaults to `local` for a tasks file; platform tasksets default to remote hosted execution. |
+| `--runtime` | Placement: `local` (a child process per rollout, serving the env source — `SubprocessRuntime`), `hud` (HUD runtime tunnel), or `tcp://host:port`. Defaults to `local` for a tasks file; platform tasksets default to remote hosted execution. |
 | `--remote` | Run the whole rollout remotely on the HUD platform. |
 | `--yes`, `-y` | Skip confirmation prompt. |
 
diff --git a/docs/v6/reference/environment.mdx b/docs/v6/reference/environment.mdx
index 6a38f8c93..ad1a60418 100644
--- a/docs/v6/reference/environment.mdx
+++ b/docs/v6/reference/environment.mdx
@@ -59,7 +59,7 @@ and [Tasks & Tasksets](/v6/reference/tasks).
 You rarely serve by hand - `hud eval`, [`task.run()`](/v6/reference/tasks), and `Taskset.run()` bring
 the environment up for you, and the [runtime](/v6/reference/runtime) you pass decides where. Serving
 itself belongs to `hud.environment.server`, the entry point every substrate runs: a
-[`LocalRuntime`](/v6/reference/runtime#localruntime) child process, a container CMD, or `hud serve`.
+[`SubprocessRuntime`](/v6/reference/runtime#subprocessruntime) child process, a container CMD, or `hud serve`.
 
 | Function | Description |
 |----------|-------------|
diff --git a/docs/v6/reference/runtime.mdx b/docs/v6/reference/runtime.mdx
index 1db44baa1..102760d83 100644
--- a/docs/v6/reference/runtime.mdx
+++ b/docs/v6/reference/runtime.mdx
@@ -18,7 +18,8 @@ await taskset.run(agent, runtime=LocalRuntime("env.py"))   # serve env.py locall
 
 | Runtime | Where the env runs | When to reach for it |
 |---------|--------------------|----------------------|
-| `LocalRuntime("env.py")` | A child process from your source | Fastest iteration; local development |
+| `LocalRuntime("env.py")` | This process, loaded fresh from your source per rollout | Fastest iteration; local development |
+| `SubprocessRuntime("env.py")` | A child process from your source | Local runs isolated from your process |
 | `DockerRuntime("my-env")` | A fresh local container per rollout | Reproducibility and parity with production |
 | `ModalRuntime("my-env")` | A fresh [Modal](https://modal.com/) sandbox per rollout | Cloud scale, no infra to manage |
 | `DaytonaRuntime("my-env")` | A fresh [Daytona](https://www.daytona.io/) sandbox per rollout | Cloud scale on Daytona |
@@ -30,10 +31,17 @@ Most runtimes are on the top-level package (`from hud import LocalRuntime, Docke
 HostedRuntime, Runtime`); `ModalRuntime` and `DaytonaRuntime` import from `hud.eval`.
 
 <Note>
-**Omit `runtime=` and it's inferred** from each task's `_source`, the file its template was defined
-in. When every task shares one `_source`, that source is served locally as `LocalRuntime(source)`;
-otherwise (mixed sources, or rows loaded from a file or the platform with no source) it falls back to
-`HUDRuntime()`. Pass a runtime explicitly the moment you want something else.
+**You can usually omit `runtime=`.** A run without one uses what HUD already knows:
+
+- a taskset loaded from Python source (`Taskset.from_file` / `from_module`) runs against that source
+- a platform taskset (`Taskset.from_api`) runs on the platform
+- otherwise, if the envs your tasks name are defined in files you've imported, each rollout gets a
+  fresh env served from its file — so `my_task().run(agent)` just works in the project that defines
+  the env
+
+When none of these apply, `run` raises and lists the runtimes you can pass — it never silently picks
+one. If two imported files define an env with the same name, that's also an error; disambiguate by
+passing the instance you mean: `runtime=LocalRuntime(env)`.
 </Note>
 
 To deploy an environment to the platform and run against it, see
@@ -74,7 +82,30 @@ The constructor for each built-in runtime:
 ### `LocalRuntime`
 
 ```python
-LocalRuntime(path, *, env=None, ready_timeout=120.0)
+LocalRuntime(source, *, env=None, ready_timeout=120.0)
+```
+
+Serves a fresh env per rollout, in this process, over the same control channel as every placement.
+`source` is any pointer to the env:
+
+- **a `.py` file or directory** that declares it, imported fresh per rollout (sibling imports
+  resolve). **`env`** pins one name when the source declares several; it defaults to the placed
+  task's env.
+- **a live `Environment`** declared at module level - its declaring module's file is the recipe;
+  the instance itself is never served, so every rollout is still fresh.
+- **a `(task) -> Environment` constructor** for envs built in code (integrations, parameterized
+  envs), called fresh per rollout with the placed row.
+
+`ready_timeout` bounds `@env.initialize` startup. The freshness boundary is the env's own source:
+it is re-imported per rollout, while modules it imports follow normal Python import caching and are
+shared process-wide - state kept in helper modules persists across rollouts. Env hooks run in this
+process and share its event loop - keep envs async, or use `SubprocessRuntime` / `DockerRuntime`
+when rollouts need whole-process isolation.
+
+### `SubprocessRuntime`
+
+```python
+SubprocessRuntime(path, *, env=None, ready_timeout=120.0)
 ```
 
 - **`path`** - `.py` file (or directory) that declares the env. The child's working directory is the source's directory, so sibling imports and relative data paths resolve.
diff --git a/docs/v6/start/overview.mdx b/docs/v6/start/overview.mdx
index 35ef95bbd..3b5818278 100644
--- a/docs/v6/start/overview.mdx
+++ b/docs/v6/start/overview.mdx
@@ -157,7 +157,7 @@ hud eval env.py claude --runtime hud    # same env, executed on HUD's hosted inf
 ```python
 from hud.eval import LocalRuntime, DockerRuntime, ModalRuntime, HUDRuntime
 
-LocalRuntime("env.py")     # local child process - fastest iteration
+LocalRuntime("env.py")     # in this process - fastest iteration
 DockerRuntime("my-env")    # a fresh container per rollout
 ModalRuntime("my-env")     # a Modal cloud sandbox per rollout
 HUDRuntime()               # HUD's hosted infra (after `hud deploy`)
diff --git a/hud/__init__.py b/hud/__init__.py
index 268724cae..76f4b5e54 100644
--- a/hud/__init__.py
+++ b/hud/__init__.py
@@ -24,6 +24,7 @@
     RuntimeGPU,
     RuntimeLimits,
     RuntimeResources,
+    SubprocessRuntime,
     SyncPlan,
     Task,
     Taskset,
@@ -49,6 +50,7 @@
     "RuntimeGPU",
     "RuntimeLimits",
     "RuntimeResources",
+    "SubprocessRuntime",
     "SyncPlan",
     "Task",
     "Taskset",
diff --git a/hud/cli/eval.py b/hud/cli/eval.py
index 39afd6edf..0a071b53b 100644
--- a/hud/cli/eval.py
+++ b/hud/cli/eval.py
@@ -708,7 +708,7 @@ def _python_defines_environment(path: Path) -> bool:
 
 
 def _spawn_target(source: Path) -> Path:
-    """The path the ``LocalRuntime`` provider serves.
+    """The path the ``SubprocessRuntime`` provider serves.
 
     Directories and env-defining ``.py`` files are served as-is. Task-only
     sources (``tasks.py`` importing from ``env.py``) resolve to a sibling
@@ -736,7 +736,7 @@ def _resolve_placement(cfg: EvalConfig, source_path: Path | None) -> Any:
     ``--remote`` submits every rollout for platform-hosted execution; a
     ``tcp://`` url attaches to an env served elsewhere.
     """
-    from hud.eval import HostedRuntime, HUDRuntime, LocalRuntime, Runtime
+    from hud.eval import HostedRuntime, HUDRuntime, Runtime, SubprocessRuntime
 
     if cfg.remote:
         require_api_key("run remote hosted evals")
@@ -744,7 +744,7 @@ def _resolve_placement(cfg: EvalConfig, source_path: Path | None) -> Any:
     if cfg.runtime == "local":
         if source_path is None:
             raise ValueError("local placement requires a local source path")
-        return LocalRuntime(_spawn_target(source_path))
+        return SubprocessRuntime(_spawn_target(source_path))
     if cfg.runtime == "hud":
         require_api_key("run HUD runtime tunnel evals")
         return HUDRuntime()
diff --git a/hud/cli/task.py b/hud/cli/task.py
index 59a0637e4..3e15470b2 100644
--- a/hud/cli/task.py
+++ b/hud/cli/task.py
@@ -94,7 +94,7 @@ def _resolve(
     """
     from contextlib import nullcontext
 
-    from hud.eval.runtime import LocalRuntime, Runtime
+    from hud.eval.runtime import Runtime, SubprocessRuntime
 
     attach = url
     if attach is None and source is None:
@@ -118,7 +118,7 @@ def _resolve(
         hud_console.error(f"No task matching {task!r} (available: {available})")
         raise typer.Exit(1)
     selected = matches[0]
-    placement = LocalRuntime(_spawn_target(source or "."))(selected)
+    placement = SubprocessRuntime(_spawn_target(source or "."))(selected)
     return selected.id, args or selected.args, placement
 
 
diff --git a/hud/environment/env.py b/hud/environment/env.py
index 07e1afd80..b3072e78c 100644
--- a/hud/environment/env.py
+++ b/hud/environment/env.py
@@ -120,13 +120,7 @@ def __call__(self, *args: P.args, **kwargs: P.kwargs) -> EvalTask:
         from hud.eval.task import Task
 
         bound = self.sig.bind(*args, **kwargs)
-        task = Task(env=self.env.name, id=self.id, args=dict(bound.arguments))
-        # Record where this template was defined so ``task.run()`` can default to
-        # serving that source locally (in-process only; never crosses the wire).
-        source = inspect.getsourcefile(self.func)
-        if source is not None:
-            task._source = source
-        return task
+        return Task(env=self.env.name, id=self.id, args=dict(bound.arguments))
 
 
 class Environment(LegacyEnvMixin):
diff --git a/hud/environment/server.py b/hud/environment/server.py
index e18f64a43..f4e264c22 100644
--- a/hud/environment/server.py
+++ b/hud/environment/server.py
@@ -6,7 +6,7 @@
 (:func:`bind`), and the full serving lifecycle (:func:`serve`) — backing
 daemons up, control channel bound (announcing the port on stdout as
 ``HUD_SERVE_PORT=<port>``), daemons down. Every substrate shape runs it: the
-:class:`~hud.eval.runtime.LocalRuntime` child process, a container CMD, and
+:class:`~hud.eval.runtime.SubprocessRuntime` child process, a container CMD, and
 ``hud serve``.
 """
 
diff --git a/hud/eval/__init__.py b/hud/eval/__init__.py
index 83f7970ab..2824645c7 100644
--- a/hud/eval/__init__.py
+++ b/hud/eval/__init__.py
@@ -14,9 +14,9 @@
 row.)
 
 Placement is passed at execution time (see :mod:`.runtime`): ``LocalRuntime`` a
-local source, ``DockerRuntime`` an image, ``Runtime(url)`` an env served
-elsewhere, ``HUDRuntime`` a HUD runtime tunnel, or ``HostedRuntime`` to run the
-whole rollout remotely on the platform::
+local source served in-process, ``DockerRuntime`` an image,
+``Runtime(url)`` an env served elsewhere, ``HUDRuntime`` a HUD runtime tunnel,
+or ``HostedRuntime`` to run the whole rollout remotely on the platform::
 
     from hud.eval import LocalRuntime, Taskset
 
@@ -46,6 +46,7 @@
     RuntimeGPU,
     RuntimeLimits,
     RuntimeResources,
+    SubprocessRuntime,
 )
 from .sync import SyncPlan
 from .task import Task
@@ -68,6 +69,7 @@
     "RuntimeGPU",
     "RuntimeLimits",
     "RuntimeResources",
+    "SubprocessRuntime",
     "SyncPlan",
     "Task",
     "Taskset",
diff --git a/hud/eval/chat.py b/hud/eval/chat.py
index 836f41713..17fee982d 100644
--- a/hud/eval/chat.py
+++ b/hud/eval/chat.py
@@ -132,8 +132,9 @@ async def send(self, message: MessageContent) -> Trace:
         )
         if self._runtime is None:
             raise RuntimeError(
-                "Chat needs a runtime to converse against — pass an env placement, "
-                'e.g. runtime=Runtime("tcp://...") or runtime=LocalRuntime("env.py").'
+                "Chat needs a runtime to converse against — pass an env placement: "
+                'LocalRuntime("env.py") (a source file), LocalRuntime(env) (a live env), '
+                "Runtime(url) (a served substrate), or HUDRuntime() (your deployed env)."
             )
         if self.job is None:  # one job spans the whole conversation
             self.job = await Job.start(self._task.id)
diff --git a/hud/eval/runtime.py b/hud/eval/runtime.py
index eea8bba34..a05e16ab1 100644
--- a/hud/eval/runtime.py
+++ b/hud/eval/runtime.py
@@ -6,8 +6,12 @@
 transparent, so "co-located" (loopback) and "split" (agent here, env
 elsewhere) are the same code, differing only in the url.
 
-- :class:`LocalRuntime` — runs a subprocess serving the row's env from a ``.py``
-  source (the path is always given, never recovered from a live object).
+- :class:`LocalRuntime` — serve a fresh env per rollout, in this process,
+  from any pointer to it: a ``.py`` source path, a live module-level
+  :class:`Environment` (its declaring file is the recipe), or a
+  ``(task) -> Environment`` constructor.
+- :class:`SubprocessRuntime` — serve the row's env from a ``.py`` source in a
+  child process, when the env should not share the orchestrator's fate.
 - :class:`DockerRuntime` — ``docker run``s an image whose CMD serves the channel.
 - ``Runtime(url)`` — the ``nullcontext`` of providers: yields itself, a
   *borrowed, shared* substrate provisioned elsewhere (env served anywhere —
@@ -49,7 +53,7 @@
 from .run import Grade, Run, rollout
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Mapping, Sequence
+    from collections.abc import AsyncIterator, Callable, Iterator, Mapping, Sequence
 
     from hud.agents.base import Agent
     from hud.environment.env import Environment
@@ -154,7 +158,204 @@ def _modal_image_from_uri(modal: Any, image_uri: str) -> Any:
 
 
 class LocalRuntime:
-    """The local provider: serve the placed row's env from *path* in a child process.
+    """The local provider: serve a fresh env per rollout, in this process.
+
+    *source* points at the env in whatever form you have:
+
+    - a ``.py`` file or directory — imported fresh per acquisition (sibling
+      imports resolve); *env* pins one name when several are declared,
+      defaulting to the placed task's env
+    - a live :class:`~hud.environment.Environment` — shorthand for its
+      declaring file; the instance itself is never served
+    - a ``(task) -> Environment`` callable — called per acquisition with the
+      placed row
+
+    ::
+
+        runtime = LocalRuntime("env.py")
+        runtime = LocalRuntime(env)
+        runtime = LocalRuntime(lambda task: build_env(task.env))
+
+    ``ready_timeout`` bounds ``@env.initialize`` startup. Freshness covers
+    the env's own source; modules it imports are cached as usual and shared
+    across rollouts. Hooks share this process's event loop, so blocking env
+    code stalls concurrent rollouts — use :class:`SubprocessRuntime` or
+    :class:`DockerRuntime` for process isolation, and ``Runtime(url)`` to
+    attach to a substrate served elsewhere.
+    """
+
+    def __init__(
+        self,
+        source: str | Path | Environment | Callable[[Task], Environment],
+        *,
+        env: str | None = None,
+        ready_timeout: float = 120.0,
+    ) -> None:
+        from hud.environment.env import Environment as _Environment
+
+        self.ready_timeout = ready_timeout
+        # A live instance may have been mutated since its module was imported;
+        # verify the fresh copy still declares its templates, so drift fails
+        # at acquisition with the cause named instead of "unknown task" later.
+        expected_templates: frozenset[str] = frozenset()
+        if isinstance(source, _Environment):
+            file = _declaring_file(source, env or source.name)
+            if file is None:
+                raise TypeError(
+                    f"LocalRuntime: env {source.name!r} is not rebuilt by importing "
+                    "any file this process has loaded (constructed in a function or "
+                    "notebook cell, or declared inside a package using relative "
+                    "imports); pass its constructor instead: "
+                    "LocalRuntime(lambda task: <build the env>)"
+                )
+            expected_templates = frozenset(source.tasks)
+            source, env = file, env or source.name
+        self._source_dir: Path | None = None
+        if isinstance(source, (str, Path)):
+            path, pinned = Path(source).resolve(), env
+            self._source_dir = path if path.is_dir() else path.parent
+            from hud.environment import load_environment
+
+            def _load(task: Task) -> _Environment:
+                loaded = load_environment(path, name=pinned or task.env)
+                missing = expected_templates - loaded.tasks.keys()
+                if missing:
+                    raise ValueError(
+                        f"env {loaded.name!r} loaded from {path} lacks template(s) "
+                        f"{sorted(missing)} present on the live instance — it was "
+                        "modified after import; pass a constructor instead: "
+                        "LocalRuntime(lambda task: <build the env>)"
+                    )
+                return loaded
+
+            self._build: Callable[[Task], _Environment] = _load
+        elif callable(source):
+            if env is not None:
+                raise TypeError("LocalRuntime: env= applies only to source paths")
+            self._build = source
+        else:
+            raise TypeError(
+                f"LocalRuntime: expected a source path, a live Environment, or a "
+                f"(task) -> Environment constructor; got {source!r}"
+            )
+
+    @asynccontextmanager
+    async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
+        from hud.environment.env import Environment as _Environment
+
+        if task.runtime_config is not None:
+            raise ValueError("LocalRuntime does not support task runtime_config")
+        # The source dir stays importable for the whole acquisition, not just
+        # the initial import, so a template can lazily import a sibling
+        # module at run time (as it could under the child-process runtime).
+        # Always insert-and-remove one entry: balanced under concurrency.
+        if self._source_dir is not None:
+            sys.path.insert(0, str(self._source_dir))
+        try:
+            try:
+                env = self._build(task)
+            except RuntimeError as e:
+                # The source ran an event loop at import — usually an unguarded
+                # top-level run call; name the actual mistake.
+                if "running event loop" not in str(e):
+                    raise
+                raise RuntimeError(
+                    "the env source ran async code while being imported to place a "
+                    'rollout — guard top-level run calls with `if __name__ == "__main__":`'
+                ) from e
+            if not isinstance(env, _Environment):
+                raise TypeError(f"LocalRuntime: constructor returned {env!r}, not an Environment")
+            async with _local(env, ready_timeout=self.ready_timeout) as runtime:
+                yield runtime
+        finally:
+            if self._source_dir is not None:
+                with contextlib.suppress(ValueError):
+                    sys.path.remove(str(self._source_dir))
+
+
+def _live_envs() -> Iterator[tuple[Environment, str]]:
+    """Envs declared in loaded, file-backed modules' globals, with their files.
+
+    The in-memory counterpart of scanning ``.py`` sources on disk
+    (:func:`~hud.environment.load_environment`): an env found here can be
+    served fresh by re-importing its file. Envs in modules without a file
+    (a notebook ``__main__``) are not yielded — re-import could not
+    reconstruct them.
+    """
+    from hud.environment.env import Environment as _Environment
+
+    for module in list(sys.modules.values()):
+        module_file = getattr(module, "__file__", None)
+        module_vars = getattr(module, "__dict__", None)
+        if not module_file or not isinstance(module_vars, dict):
+            continue
+        for value in list(module_vars.values()):
+            if isinstance(value, _Environment):
+                yield value, module_file
+
+
+def _declaring_file(env: Environment, name: str) -> Path | None:
+    """A file whose fresh import re-declares *env*, else None.
+
+    Candidate files hold the instance in their module globals, but a holder
+    may be a re-exporter (``from .env import env`` in a package
+    ``__init__``, a tasks file re-exporting its env): validate each by
+    loading it fresh — a declarer yields a *new* instance under *name*, a
+    re-exporter yields the same live one (or fails to import standalone).
+    ``__init__.py`` holders are tried last.
+    """
+    from hud.environment import load_environment
+
+    candidates = dict.fromkeys(Path(file) for live, file in _live_envs() if live is env)
+    for file in sorted(candidates, key=lambda f: f.name == "__init__.py"):
+        try:
+            probe = load_environment(file, name=name)
+        except Exception as e:
+            logger.debug("candidate %s does not rebuild env %r: %s", file, name, e)
+            continue
+        if probe is not env:
+            return file
+    return None
+
+
+def _declared_env(name: str) -> Environment | None:
+    """The one live env named *name*, else None; two distinct ones raise.
+
+    The same instance re-exported across modules is one match; distinct envs
+    claiming one name are ambiguous.
+    """
+    matches = {id(env): env for env, _ in _live_envs() if env.name == name}
+    if len(matches) > 1:
+        files = sorted({file for env, file in _live_envs() if env.name == name})
+        raise ValueError(
+            f"env name {name!r} is declared by multiple live environments "
+            f"({', '.join(files)}); pass runtime= explicitly — the exact "
+            "instance disambiguates: runtime=LocalRuntime(env)"
+        )
+    return next(iter(matches.values()), None)
+
+
+def _declared_names(source: Path) -> set[str]:
+    """Env names a ``.py`` source (file or directory) itself declares.
+
+    A fresh execution of the source yields *new* instances for envs it
+    declares; an env it merely imports is the already-live one and does not
+    count — importing the source again could not rebuild it.
+    """
+    from hud.environment.env import Environment as _Environment
+    from hud.utils.modules import iter_modules
+
+    live = {id(env) for env, _ in _live_envs()}
+    return {
+        value.name
+        for module in iter_modules(source)
+        for value in vars(module).values()
+        if isinstance(value, _Environment) and id(value) not in live
+    }
+
+
+class SubprocessRuntime:
+    """The child-process provider: serve the placed row's env from *path*.
 
     Each acquisition runs ``python -m hud.environment.server <path> --env
     name`` — the same serving entry point a container CMD runs — on an
@@ -167,8 +368,8 @@ class LocalRuntime:
     The child's working directory is the source's directory, so sibling
     imports and relative data paths resolve; ``@env.initialize`` daemons start
     in the child and die with it. Because the source is re-imported in the
-    child, a script spawning itself (``LocalRuntime(__file__)``) must keep top-level
-    run calls under ``if __name__ == "__main__":``.
+    child, a script spawning itself (``SubprocessRuntime(__file__)``) must keep
+    top-level run calls under ``if __name__ == "__main__":``.
     """
 
     def __init__(
@@ -185,9 +386,9 @@ def __init__(
     @asynccontextmanager
     async def __call__(self, task: Task) -> AsyncIterator[Runtime]:
         if task.runtime_config is not None:
-            raise ValueError("LocalRuntime does not support task runtime_config")
+            raise ValueError("SubprocessRuntime does not support task runtime_config")
         if not self.source.exists():
-            raise FileNotFoundError(f"LocalRuntime: source not found: {self.source}")
+            raise FileNotFoundError(f"SubprocessRuntime: source not found: {self.source}")
         cmd = [sys.executable, "-m", "hud.environment.server", str(self.source)]
         cmd += ["--env", self.env or task.env]
         proc = await create_process_group_exec(
@@ -603,30 +804,34 @@ async def _docker(*args: str, check: bool = True) -> tuple[str, str]:
 
 
 @asynccontextmanager
-async def _local(env: Environment) -> AsyncIterator[Runtime]:
+async def _local(env: Environment, *, ready_timeout: float | None = None) -> AsyncIterator[Runtime]:
     """Substrate-side serving: a live env owned by *this* process, as a runtime.
 
-    Not a placement the engine offers (the orchestrator never serves an env
-    in-process), so deliberately not a ``Provider`` — it serves a live object,
-    not a placed row. Code already running *inside* a placed substrate adapts
-    it (``AgentTool`` sub-rollouts: ``runtime=lambda _: _local(env)``); test
-    harnesses enter it directly.
+    One env lifecycle (start → serve → stop) around one bound control
+    channel; ``ready_timeout`` bounds ``env.start()`` (initialize
+    hooks/daemons). ``LocalRuntime`` enters this per acquisition with the
+    fresh env it built; test harnesses enter it directly with a live one.
     """
     from hud.environment.server import bind
 
-    await env.start()
-    server = await bind(env, "127.0.0.1", 0)
-    host, port = server.sockets[0].getsockname()[:2]
-    serve_task = asyncio.create_task(server.serve_forever())
+    # start() inside the try: a failed or timed-out initialize hook still gets
+    # its already-started daemons torn down by stop() (best-effort per hook).
     try:
-        yield Runtime(f"tcp://{host}:{port}")
+        started = env.start()
+        await (asyncio.wait_for(started, ready_timeout) if ready_timeout is not None else started)
+        server = await bind(env, "127.0.0.1", 0)
+        host, port = server.sockets[0].getsockname()[:2]
+        serve_task = asyncio.create_task(server.serve_forever())
+        try:
+            yield Runtime(f"tcp://{host}:{port}")
+        finally:
+            serve_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await serve_task
+            server.close()
+            with contextlib.suppress(Exception):
+                await server.wait_closed()
     finally:
-        serve_task.cancel()
-        with contextlib.suppress(asyncio.CancelledError):
-            await serve_task
-        server.close()
-        with contextlib.suppress(Exception):
-            await server.wait_closed()
         await env.stop()
 
 
@@ -1011,4 +1216,5 @@ async def ws_to_tcp() -> None:
     "RuntimeGPU",
     "RuntimeLimits",
     "RuntimeResources",
+    "SubprocessRuntime",
 ]
diff --git a/hud/eval/task.py b/hud/eval/task.py
index ab3363ae2..2f6eb2814 100644
--- a/hud/eval/task.py
+++ b/hud/eval/task.py
@@ -24,7 +24,7 @@
 import json
 from typing import TYPE_CHECKING, Any
 
-from pydantic import BaseModel, Field, PrivateAttr
+from pydantic import BaseModel, Field
 
 from .runtime import RuntimeConfig
 
@@ -40,8 +40,8 @@ class Task(BaseModel):
 
     Pure data — holds no execution state, so one ``Task`` can drive many
     concurrent rollouts. ``run`` it for a graded :class:`~hud.eval.job.Job`;
-    placement comes from ``runtime=`` (a provider), else the source the task was
-    minted from (local), else the HUD runtime tunnel by ``env`` name.
+    placement comes from ``runtime=`` (a provider), else the HUD runtime
+    tunnel by ``env`` name.
     """
 
     env: str = Field(min_length=1)
@@ -57,12 +57,6 @@ class Task(BaseModel):
     #: supported subset into their native launch shape or reject it.
     runtime_config: RuntimeConfig | None = None
 
-    #: In-process only: the source file the template was defined in, captured
-    #: when a template factory mints the task. Lets ``run`` default to serving
-    #: that source locally. Excluded from the wire (a row loaded from JSON has
-    #: none, and falls back to HUD runtime tunnel placement).
-    _source: str | None = PrivateAttr(default=None)
-
     def default_slug(self) -> str:
         """A stable slug from the task id, disambiguated by an args hash when present."""
         if not self.args:
@@ -90,8 +84,8 @@ async def run(
         open ``job`` from :meth:`Job.start` to accumulate into), ``group``
         repeats sharing a group_id, ``max_concurrent`` capping parallelism —
         over a taskset of one. ``runtime`` is the placement; left unset it
-        serves the task's source locally when minted in-process, else falls
-        back to the HUD runtime tunnel by ``env`` name.
+        falls back to the HUD runtime tunnel by ``env`` name. For a local
+        run, pass one explicitly (``runtime=LocalRuntime("env.py")``).
         """
         from .taskset import Taskset  # circular: taskset -> sync -> task
 
diff --git a/hud/eval/taskset.py b/hud/eval/taskset.py
index a7f7e9cea..2ef0d223a 100644
--- a/hud/eval/taskset.py
+++ b/hud/eval/taskset.py
@@ -23,7 +23,7 @@
 
 from .job import Job, job_enter
 from .run import rollout
-from .runtime import HostedRuntime, HUDRuntime, LocalRuntime
+from .runtime import HostedRuntime, HUDRuntime, LocalRuntime, _declared_env, _declared_names
 from .sync import fetch_taskset_tasks, resolve_taskset_id
 
 if TYPE_CHECKING:
@@ -208,6 +208,37 @@ def environment_names(self) -> set[str]:
         """Return env names referenced by tasks in this taskset."""
         return {task.env for task in self}
 
+    def _resolve_placement(self) -> Provider | HUDRuntime:
+        if self.origin and self.origin.startswith("module:"):
+            # The origin claims the rows only if it actually declares their
+            # envs (a tasks-only module importing its envs from elsewhere
+            # does not) — and it serves as the exact path, so a same-named
+            # variant in a sibling file is never dragged in.
+            source = Path(self.origin[len("module:") :])
+            if self.environment_names() <= _declared_names(source):
+                return LocalRuntime(source)
+        if self.origin and self.origin.startswith("api:"):
+            return HUDRuntime()
+        declared = {name: _declared_env(name) for name in self.environment_names()}
+        if declared and all(declared.values()):
+            providers = {
+                name: LocalRuntime(env) for name, env in declared.items() if env is not None
+            }
+            logger.info(
+                "no runtime given: serving %s fresh from their declaring modules",
+                ", ".join(sorted(providers)),
+            )
+            return lambda task: providers[task.env](task)
+        missing = sorted(name for name, env in declared.items() if env is None)
+        raise ValueError(
+            f"no placement for env(s) {', '.join(missing) or '<none>'}: pass runtime= — "
+            'LocalRuntime("env.py") (a source file), LocalRuntime(env) (a live env), '
+            "LocalRuntime(build) (a (task) -> Environment constructor), Runtime(url) "
+            "(a served substrate), or HUDRuntime() (your deployed env). A row taken "
+            "from a loaded taskset keeps its placement when run through it: "
+            'taskset.filter(["slug"]).run(...)'
+        )
+
     async def run(
         self,
         agent: Agent,
@@ -224,7 +255,12 @@ async def run(
         placement: a :class:`~hud.eval.runtime.Provider` (the env served
         somewhere, the agent loop driven here by :func:`~hud.eval.run.rollout`),
         or :class:`~hud.eval.runtime.HostedRuntime` to run each rollout remotely
-        on the platform (left unset: HUD tunnel by env name). One provider serves a mixed-env
+        on the platform. Left unset, what is already known decides: a
+        taskset loaded from local ``.py`` source serves that source's
+        directory; a platform taskset runs on the platform; rows naming envs
+        declared in imported modules serve each fresh from its file; anything
+        else raises, naming the forms to pass. One provider serves a
+        mixed-env
         taskset and can size each substrate per row. Registers one HUD job as
         the platform receipt and reports each run's trace under it — or, given
         an open ``job`` (:meth:`Job.start`), accumulates this batch into it
@@ -264,17 +300,17 @@ async def run(
 
         # Placement is chosen once for the batch: HostedRuntime delegates the
         # whole rollout to the platform, anything else is a Provider driven
-        # locally by rollout().
-        # No runtime: serve the tasks' shared source locally if they were minted
-        # in-process from one file (the common authoring case); otherwise (mixed
-        # or wire-loaded rows with no source) default to the HUD runtime tunnel.
-        if runtime is None:
-            sources = {t._source for t in task_list if t._source is not None}
-            runtime = LocalRuntime(next(iter(sources))) if len(sources) == 1 else None
-        placement = runtime if runtime is not None else HUDRuntime()
+        # locally by rollout(). No runtime: what the taskset or this process
+        # already knows decides (rows never carry placement) — a loaded
+        # taskset runs where it came from; rows naming envs declared in
+        # imported modules serve each fresh from its file; anything else is
+        # an error naming the forms to pass.
+        # An empty taskset schedules nothing, so it needs no placement.
+        placement = runtime if runtime is not None or not task_list else self._resolve_placement()
         sem = asyncio.Semaphore(max_concurrent) if max_concurrent else None
 
         async def _run(task: Task, group_id: str) -> Run:
+            assert placement is not None  # only reached when tasks were expanded
             if isinstance(placement, HostedRuntime):
                 return await placement.run(task, agent, job_id=job_id, group_id=group_id)
             return await rollout(
diff --git a/hud/eval/tests/test_chat.py b/hud/eval/tests/test_chat.py
index 68b6bdaf0..47220ccfb 100644
--- a/hud/eval/tests/test_chat.py
+++ b/hud/eval/tests/test_chat.py
@@ -1,6 +1,6 @@
 """``Chat`` — multi-turn conversation runner over a task.
 
-Turn tests place each turn's rollout with ``runtime=LocalRuntime(env_file)`` — a pure-data
+Turn tests place each turn's rollout with ``runtime=SubprocessRuntime(env_file)`` — a pure-data
 ``Task`` row against a chat-style env served from a child process.
 """
 
@@ -13,7 +13,7 @@
 from mcp.types import TextContent
 
 from hud.agents.base import Agent
-from hud.eval import LocalRuntime, Task
+from hud.eval import SubprocessRuntime, Task
 from hud.eval.chat import Chat, _content_to_blocks
 
 if TYPE_CHECKING:
@@ -89,7 +89,7 @@ class TestSend:
     async def test_send_runs_a_turn_and_stores_prompt_message_format(
         self, chat_env_file: Path
     ) -> None:
-        chat = Chat(_chat_task(), _EchoAgent(), runtime=LocalRuntime(chat_env_file))
+        chat = Chat(_chat_task(), _EchoAgent(), runtime=SubprocessRuntime(chat_env_file))
 
         trace = await chat.send("hello")
 
@@ -107,7 +107,7 @@ async def test_send_runs_a_turn_and_stores_prompt_message_format(
         assert assistant_msg["content"]["text"] == "echo:hello"
 
     async def test_one_job_spans_the_conversation(self, chat_env_file: Path) -> None:
-        chat = Chat(_chat_task(), _EchoAgent(), runtime=LocalRuntime(chat_env_file))
+        chat = Chat(_chat_task(), _EchoAgent(), runtime=SubprocessRuntime(chat_env_file))
 
         await chat.send("hello")
         await chat.send("again")
@@ -125,7 +125,7 @@ class _Boom(Agent):
             async def __call__(self, run: Any) -> None:
                 raise RuntimeError("agent exploded")
 
-        chat = Chat(_chat_task(), _Boom(), runtime=LocalRuntime(chat_env_file))
+        chat = Chat(_chat_task(), _Boom(), runtime=SubprocessRuntime(chat_env_file))
 
         with pytest.raises(RuntimeError, match="agent exploded"):
             await chat.send("hello")
diff --git a/hud/eval/tests/test_local_runtime.py b/hud/eval/tests/test_local_runtime.py
new file mode 100644
index 000000000..fd8ec4240
--- /dev/null
+++ b/hud/eval/tests/test_local_runtime.py
@@ -0,0 +1,398 @@
+"""Local placement: LocalRuntime and the no-runtime resolution ladder.
+
+LocalRuntime serves a fresh env per rollout from any pointer to it — a source
+path (throwaway import), a live module-level env (its declaring file is the
+recipe), or a ``(task) -> Environment`` constructor. With no runtime, a run
+uses what is already known: taskset origin, then envs declared in imported
+modules, else a loud error. Everything crosses the real control channel —
+these tests drive the rollout engine end to end.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from collections.abc import AsyncGenerator  # noqa: TC003 - env.template resolves at runtime
+from typing import Any, cast
+
+import pytest
+
+from hud.agents.base import Agent
+from hud.environment import Environment
+from hud.eval import LocalRuntime, Task, Taskset
+from hud.eval.run import rollout
+
+_SUMS_ENV = """\
+from hud import Environment
+
+env = Environment("{name}")
+
+
+@env.template(id="add")
+async def add(a: int, b: int):
+    answer = yield f"add:{{a}}:{{b}}"
+    yield 1.0 if answer == str(a + b) else 0.0
+"""
+
+
+@pytest.fixture
+def imported_env(tmp_path, request):
+    """Write an env module, import it for real, and clean it up after.
+
+    The module stays in ``sys.modules`` for the test's duration — the state a
+    user's ``from env import add`` leaves behind.
+    """
+    module_name = f"_sums_mod_{request.node.name}"
+    file = tmp_path / f"{module_name}.py"
+    file.write_text(_SUMS_ENV.format(name="sums"), encoding="utf-8")
+    spec = importlib.util.spec_from_file_location(module_name, file)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
+    yield module
+    del sys.modules[module_name]
+
+
+def _sums_env(name: str = "sums") -> Environment:
+    env = Environment(name)
+
+    @env.template(id="add")
+    async def add(a: int, b: int) -> AsyncGenerator[Any, Any]:
+        answer = yield f"add:{a}:{b}"
+        yield 1.0 if answer == str(a + b) else 0.0
+
+    return env
+
+
+class _FnAgent(Agent):
+    """Stateless agent: answers each run by applying ``fn`` to ``run.prompt``."""
+
+    def __init__(self, fn: Any) -> None:
+        self._fn = fn
+
+    async def __call__(self, run: Any) -> None:
+        run.trace.content = self._fn(run.prompt)
+
+
+def _solve_add(prompt: str) -> str:
+    _, a, b = prompt.split(":")
+    return str(int(a) + int(b))
+
+
+# ─── LocalRuntime: the three pointer forms ─────────────────────────────
+
+
+async def test_source_path_serves_a_fresh_env_per_rollout(tmp_path) -> None:
+    # LOADS is module state: a fresh throwaway import per acquisition means
+    # every rollout sees exactly one load.
+    (tmp_path / "env.py").write_text(
+        "from hud import Environment\n\n"
+        "LOADS = []\n"
+        'env = Environment("sums")\n\n\n'
+        '@env.template(id="add")\nasync def add(a: int, b: int):\n'
+        "    LOADS.append(1)\n"
+        '    answer = yield f"add:{a}:{b}:{len(LOADS)}"\n'
+        "    yield 1.0 if answer == str(a + b) else 0.0\n",
+        encoding="utf-8",
+    )
+
+    def _solve(prompt: str) -> str:
+        _, a, b, loads = prompt.split(":")
+        assert loads == "1"
+        return str(int(a) + int(b))
+
+    job = await Task(env="sums", id="add", args={"a": 2, "b": 3}).run(
+        _FnAgent(_solve),
+        runtime=LocalRuntime(tmp_path / "env.py"),
+        group=2,
+    )
+
+    assert [run.reward for run in job.runs] == [1.0, 1.0]
+
+
+async def test_live_env_pointer_resolves_to_its_declaring_file(imported_env) -> None:
+    run = await rollout(
+        Task(env="sums", id="add", args={"a": 2, "b": 3}),
+        _FnAgent(_solve_add),
+        runtime=LocalRuntime(imported_env.env),
+    )
+
+    assert run.reward == 1.0
+
+
+def test_live_env_without_a_declaring_file_is_rejected() -> None:
+    with pytest.raises(TypeError, match="constructor instead"):
+        LocalRuntime(_sums_env())
+
+
+async def test_constructor_builds_fresh_per_rollout_from_the_row() -> None:
+    built: list[str] = []
+
+    def env_for(task: Task) -> Environment:
+        built.append(task.env)
+        return _sums_env(task.env)
+
+    job = await Task(env="sums", id="add", args={"a": 1, "b": 2}).run(
+        _FnAgent(_solve_add),
+        runtime=LocalRuntime(env_for),
+        group=3,
+        max_concurrent=3,
+    )
+
+    assert all(run.reward == 1.0 for run in job.runs)
+    assert built == ["sums", "sums", "sums"]
+
+
+async def test_source_missing_env_name_fails_loudly(tmp_path) -> None:
+    (tmp_path / "env.py").write_text(
+        'from hud import Environment\n\nenv = Environment("sums")\n',
+        encoding="utf-8",
+    )
+    provider = LocalRuntime(tmp_path / "env.py")
+
+    with pytest.raises(ValueError, match="no Environment named 'other'"):
+        async with provider(Task(env="other", id="add")):
+            pass
+
+
+# ─── the no-runtime resolution ladder ──────────────────────────────────
+
+
+async def test_module_loaded_taskset_serves_its_source_by_default(tmp_path, request) -> None:
+    (tmp_path / "env.py").write_text(_SUMS_ENV.format(name="sums"), encoding="utf-8")
+    (tmp_path / "tasks.py").write_text(
+        "from env import add\n\ntasks = [add(a=2, b=3), add(a=4, b=5)]\n",
+        encoding="utf-8",
+    )
+    # tasks.py's `from env import add` imports env.py normally, so it outlives
+    # the throwaway tasks module.
+    request.addfinalizer(lambda: sys.modules.pop("env", None))
+
+    taskset = Taskset.from_module(tmp_path / "tasks.py")
+    job = await taskset.run(_FnAgent(_solve_add))
+
+    assert len(job.runs) == 2
+    assert all(run.reward == 1.0 for run in job.runs)
+
+
+async def test_minted_tasks_resolve_a_declared_env_by_name(imported_env) -> None:
+    job = await Taskset("sums", [imported_env.add(a=2, b=3), imported_env.add(a=4, b=5)]).run(
+        _FnAgent(_solve_add)
+    )
+
+    assert len(job.runs) == 2
+    assert all(run.reward == 1.0 for run in job.runs)
+
+
+async def test_no_placement_fails_with_the_forms_to_pass() -> None:
+    with pytest.raises(ValueError, match="no placement for env"):
+        await Task(env="ghost", id="add").run(_FnAgent(_solve_add))
+
+
+async def test_ambiguous_env_name_fails_loudly(imported_env, tmp_path, request) -> None:
+    module_name = f"_sums_rival_{request.node.name}"
+    file = tmp_path / f"{module_name}.py"
+    file.write_text(_SUMS_ENV.format(name="sums"), encoding="utf-8")
+    spec = importlib.util.spec_from_file_location(module_name, file)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    try:
+        spec.loader.exec_module(module)
+        with pytest.raises(ValueError, match="LocalRuntime\\(env\\)"):
+            await Task(env="sums", id="add").run(_FnAgent(_solve_add))
+    finally:
+        del sys.modules[module_name]
+
+
+def test_rejects_a_non_pointer_argument() -> None:
+    with pytest.raises(TypeError, match="expected a source path"):
+        LocalRuntime(cast("Any", 42))
+
+
+async def test_failed_startup_still_runs_shutdown_hooks() -> None:
+    from hud.eval.runtime import _local
+
+    env = _sums_env()
+    lifecycle: list[str] = []
+
+    @env.initialize
+    async def _up() -> None:
+        lifecycle.append("up")
+
+    @env.shutdown
+    async def _down() -> None:
+        lifecycle.append("down")
+
+    @env.initialize
+    async def _boom() -> None:
+        raise RuntimeError("daemon failed to start")
+
+    with pytest.raises(RuntimeError, match="daemon failed to start"):
+        async with _local(env):
+            pass
+
+    assert lifecycle == ["up", "down"]
+
+
+async def test_tasks_only_module_resolves_envs_imported_from_elsewhere(
+    tmp_path, monkeypatch, request
+) -> None:
+    # The env lives in a separate importable package, not next to tasks.py:
+    # the origin declares no envs, so resolution falls through to the live
+    # env the tasks module imported.
+    env_dir = tmp_path / "pkg"
+    env_dir.mkdir()
+    (env_dir / "sums_envmod.py").write_text(_SUMS_ENV.format(name="sums"), encoding="utf-8")
+    tasks_dir = tmp_path / "tasks"
+    tasks_dir.mkdir()
+    (tasks_dir / "tasks.py").write_text(
+        "from sums_envmod import add\n\ntasks = [add(a=2, b=3)]\n",
+        encoding="utf-8",
+    )
+    monkeypatch.syspath_prepend(str(env_dir))
+    request.addfinalizer(lambda: sys.modules.pop("sums_envmod", None))
+
+    taskset = Taskset.from_module(tasks_dir / "tasks.py")
+    job = await taskset.run(_FnAgent(_solve_add))
+
+    assert [run.reward for run in job.runs] == [1.0]
+
+
+async def test_single_file_taskset_never_drags_in_a_same_named_sibling(tmp_path) -> None:
+    (tmp_path / "env_a.py").write_text(
+        _SUMS_ENV.format(name="sums") + "\ntasks = [add(a=2, b=3)]\n", encoding="utf-8"
+    )
+    (tmp_path / "env_b.py").write_text(_SUMS_ENV.format(name="sums"), encoding="utf-8")
+
+    taskset = Taskset.from_module(tmp_path / "env_a.py")
+    job = await taskset.run(_FnAgent(_solve_add))
+
+    assert [run.reward for run in job.runs] == [1.0]
+
+
+async def test_empty_taskset_runs_without_a_placement() -> None:
+    job = await Taskset("empty", []).run(_FnAgent(_solve_add))
+
+    assert job.runs == []
+
+
+async def test_reexporting_tasks_module_does_not_claim_the_env(
+    tmp_path, monkeypatch, request
+) -> None:
+    # tasks.py re-exports the env object alongside the factory: the origin
+    # must not claim it (re-import of tasks.py would reuse the cached env
+    # module) — each rollout rebuilds the env from its real file instead.
+    env_dir = tmp_path / "pkg"
+    env_dir.mkdir()
+    (env_dir / "sums_reexp_envmod.py").write_text(
+        "from hud import Environment\n\n"
+        "LOADS = []\n"
+        'env = Environment("sums")\n\n\n'
+        '@env.template(id="add")\nasync def add(a: int, b: int):\n'
+        "    LOADS.append(1)\n"
+        '    answer = yield f"add:{a}:{b}:{len(LOADS)}"\n'
+        "    yield 1.0 if answer == str(a + b) else 0.0\n",
+        encoding="utf-8",
+    )
+    tasks_dir = tmp_path / "tasks"
+    tasks_dir.mkdir()
+    (tasks_dir / "tasks.py").write_text(
+        "from sums_reexp_envmod import add, env\n\ntasks = [add(a=2, b=3)]\n",
+        encoding="utf-8",
+    )
+    monkeypatch.syspath_prepend(str(env_dir))
+    request.addfinalizer(lambda: sys.modules.pop("sums_reexp_envmod", None))
+
+    def _solve(prompt: str) -> str:
+        _, a, b, loads = prompt.split(":")
+        assert loads == "1"  # a fresh env module per rollout, not the cached one
+        return str(int(a) + int(b))
+
+    taskset = Taskset.from_module(tasks_dir / "tasks.py")
+    job = await taskset.run(_FnAgent(_solve), group=2)
+
+    assert [run.reward for run in job.runs] == [1.0, 1.0]
+
+
+async def test_package_reexported_env_pointer_uses_the_declaring_submodule(
+    tmp_path, monkeypatch, request
+) -> None:
+    pkg = tmp_path / "sums_pkg"
+    pkg.mkdir()
+    (pkg / "__init__.py").write_text("from .env_mod import env\n", encoding="utf-8")
+    (pkg / "env_mod.py").write_text(_SUMS_ENV.format(name="sums"), encoding="utf-8")
+    monkeypatch.syspath_prepend(str(tmp_path))
+    request.addfinalizer(
+        lambda: [sys.modules.pop(m, None) for m in ("sums_pkg", "sums_pkg.env_mod")]
+    )
+    import importlib
+
+    sums_pkg = importlib.import_module("sums_pkg")
+
+    run = await rollout(
+        Task(env="sums", id="add", args={"a": 2, "b": 3}),
+        _FnAgent(_solve_add),
+        runtime=LocalRuntime(sums_pkg.env),
+    )
+
+    assert run.reward == 1.0
+
+
+async def test_template_can_lazily_import_a_sibling_module(tmp_path) -> None:
+    (tmp_path / "lazy_helper.py").write_text("ANSWER_SUFFIX = ':ok'\n", encoding="utf-8")
+    (tmp_path / "env.py").write_text(
+        "from hud import Environment\n\n"
+        'env = Environment("sums")\n\n\n'
+        '@env.template(id="add")\nasync def add(a: int, b: int):\n'
+        "    import lazy_helper\n"
+        '    answer = yield f"add:{a}:{b}{lazy_helper.ANSWER_SUFFIX}"\n'
+        "    yield 1.0 if answer == str(a + b) else 0.0\n",
+        encoding="utf-8",
+    )
+
+    def _solve(prompt: str) -> str:
+        _, a, b, ok = prompt.split(":")
+        assert ok == "ok"
+        return str(int(a) + int(b))
+
+    job = await Task(env="sums", id="add", args={"a": 2, "b": 3}).run(
+        _FnAgent(_solve),
+        runtime=LocalRuntime(tmp_path / "env.py"),
+        group=2,
+        max_concurrent=2,
+    )
+
+    assert [run.reward for run in job.runs] == [1.0, 1.0]
+
+
+# ─── seam defenses ─────────────────────────────────────────────────────
+
+
+@pytest.mark.filterwarnings("ignore::RuntimeWarning")  # the broken source leaks its coroutine
+async def test_unguarded_run_call_in_source_names_the_mistake(tmp_path) -> None:
+    (tmp_path / "env.py").write_text(
+        "import asyncio\n\nfrom hud import Environment\n\n"
+        'env = Environment("sums")\n\n'
+        "asyncio.run(asyncio.sleep(0))\n",
+        encoding="utf-8",
+    )
+    provider = LocalRuntime(tmp_path / "env.py")
+
+    with pytest.raises(RuntimeError, match='if __name__ == "__main__"'):
+        async with provider(Task(env="sums", id="add")):
+            pass
+
+
+async def test_live_env_mutated_after_import_fails_with_the_cause(imported_env) -> None:
+    @imported_env.env.template(id="patched")
+    async def patched() -> Any:
+        answer = yield "noop"
+        yield 1.0 if answer else 0.0
+
+    provider = LocalRuntime(imported_env.env)
+
+    with pytest.raises(ValueError, match="modified after import"):
+        async with provider(Task(env="sums", id="patched")):
+            pass
diff --git a/hud/eval/tests/test_rollout.py b/hud/eval/tests/test_rollout.py
index 0d8bb5166..3473d5a1c 100644
--- a/hud/eval/tests/test_rollout.py
+++ b/hud/eval/tests/test_rollout.py
@@ -1,7 +1,7 @@
 """The rollout engine: ``rollout(task, agent)`` and its schedulers.
 
 These drive the engine end-to-end through the real placement path: a pure-data
-``Task`` row plus ``runtime=LocalRuntime(env_file)`` — a child process serves the env, the
+``Task`` row plus ``runtime=SubprocessRuntime(env_file)`` — a child process serves the env, the
 engine connects over the wire, the agent answers, grading comes back. The
 engine contract is a graded :class:`Run` with a trace id (always under a job —
 there are no standalone traces), and failure isolation that never raises: a
@@ -31,7 +31,7 @@
 from hud.agents.openai_compatible import OpenAIChatAgent
 from hud.agents.types import OpenAIChatConfig
 from hud.environment import Environment
-from hud.eval import Job, LocalRuntime, Task, Taskset
+from hud.eval import Job, SubprocessRuntime, Task, Taskset
 from hud.eval.run import Run, rollout
 from hud.eval.runtime import _local
 
@@ -155,7 +155,7 @@ async def _wait_for_pid_inactive(pid: int, max_wait: float = 2.0) -> bool:
 
 
 async def test_rollout_returns_graded_run_with_trace_id(env_file: Path) -> None:
-    run = await rollout(_add_task(2, 3), _FnAgent(_solve_add), runtime=LocalRuntime(env_file))
+    run = await rollout(_add_task(2, 3), _FnAgent(_solve_add), runtime=SubprocessRuntime(env_file))
 
     assert run.reward == 1.0
     assert run.trace.content == "5"
@@ -249,7 +249,7 @@ async def start_child():
 
     try:
         with pytest.raises(RuntimeError, match="startup boom"):
-            async with LocalRuntime(env_file, ready_timeout=2.0)(Task(env="leaky", id="noop")):
+            async with SubprocessRuntime(env_file, ready_timeout=2.0)(Task(env="leaky", id="noop")):
                 pass
         pid = int(pid_file.read_text())
         assert await _wait_for_pid_inactive(pid)
@@ -263,7 +263,7 @@ async def test_mid_run_failure_keeps_the_real_run_and_its_evidence(env_file: Pat
     def boom(prompt: str) -> str:
         raise RuntimeError("agent exploded")
 
-    run = await rollout(_add_task(2, 3), _FnAgent(boom), runtime=LocalRuntime(env_file))
+    run = await rollout(_add_task(2, 3), _FnAgent(boom), runtime=SubprocessRuntime(env_file))
 
     assert run.trace.is_error
     assert "agent exploded" in (run.trace.error or "")
@@ -291,7 +291,7 @@ async def test_mid_run_failure_still_grades_best_effort(env_file: Path) -> None:
     # The agent answers correctly, then fails. The env is still alive, so the
     # run is graded best-effort: the reward is captured even though it errored.
     run = await rollout(
-        _add_task(2, 3), _AnswerThenBoomAgent(_solve_add), runtime=LocalRuntime(env_file)
+        _add_task(2, 3), _AnswerThenBoomAgent(_solve_add), runtime=SubprocessRuntime(env_file)
     )
 
     assert run.trace.is_error
@@ -358,7 +358,7 @@ def placer(task: TaskRow) -> Any:
         # The scheduler half of placement: the row is the request, so a
         # provider can size/route each substrate per task.
         placed.append(f"{task.env}/{task.id}:{task.args['a']}")
-        return LocalRuntime(env_file)(task)
+        return SubprocessRuntime(env_file)(task)
 
     run = await rollout(_add_task(2, 3), _FnAgent(_solve_add), runtime=placer)
 
@@ -367,7 +367,7 @@ def placer(task: TaskRow) -> Any:
 
 
 async def test_task_run_schedules_a_single_task_job(env_file: Path) -> None:
-    job = await _add_task(2, 3).run(_FnAgent(_solve_add), runtime=LocalRuntime(env_file))
+    job = await _add_task(2, 3).run(_FnAgent(_solve_add), runtime=SubprocessRuntime(env_file))
 
     (run,) = job.runs
     assert job.reward == 1.0
@@ -377,7 +377,7 @@ async def test_task_run_schedules_a_single_task_job(env_file: Path) -> None:
 
 async def test_task_run_has_taskset_scheduling_semantics(env_file: Path) -> None:
     job = await _add_task(1, 2).run(
-        _FnAgent(_solve_add), runtime=LocalRuntime(env_file), group=2, max_concurrent=1
+        _FnAgent(_solve_add), runtime=SubprocessRuntime(env_file), group=2, max_concurrent=1
     )
 
     assert job.group == 2
@@ -388,7 +388,7 @@ async def test_task_run_has_taskset_scheduling_semantics(env_file: Path) -> None
 
 async def test_open_job_spans_multiple_scheduler_calls(env_file: Path) -> None:
     session = await Job.start("session", group=2)
-    provider = LocalRuntime(env_file)
+    provider = SubprocessRuntime(env_file)
 
     job1 = await _add_task(1, 1).run(_FnAgent(_solve_add), runtime=provider, job=session)
     job2 = await _add_task(2, 2).run(_FnAgent(_solve_add), runtime=provider, job=session)
@@ -434,7 +434,7 @@ async def test_one_spawn_serves_each_rows_env_in_a_mixed_taskset(
     # One provider, two envs: each acquisition serves the row it was called
     # with (the task ids only exist on their own env, so a misplacement
     # would fail the rollout).
-    job = await Taskset("zoo", rows).run(_FnAgent(_solve_add), runtime=LocalRuntime(path))
+    job = await Taskset("zoo", rows).run(_FnAgent(_solve_add), runtime=SubprocessRuntime(path))
 
     assert [run.reward for run in job.runs] == [1.0, 1.0]
     assert [run.prompt for run in job.runs] == ["alpha:1:2", "beta:3:4"]
@@ -444,7 +444,7 @@ async def test_rollout_threads_job_and_group_ids(env_file: Path) -> None:
     run = await rollout(
         _add_task(1, 1),
         _FnAgent(_solve_add),
-        runtime=LocalRuntime(env_file),
+        runtime=SubprocessRuntime(env_file),
         job_id="j1",
         group_id="g1",
     )
diff --git a/hud/eval/tests/test_task.py b/hud/eval/tests/test_task.py
index cf258a0d7..5d0de0586 100644
--- a/hud/eval/tests/test_task.py
+++ b/hud/eval/tests/test_task.py
@@ -137,7 +137,7 @@ def test_row_validation_rejects_malformed_entries() -> None:
 # ─── placement ─────────────────────────────────────────────────────────
 
 
-async def test_no_placement_defaults_to_hud_runtime(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_platform_taskset_defaults_to_hud_runtime(monkeypatch: pytest.MonkeyPatch) -> None:
     import hud.eval.taskset as taskset_mod
 
     seen: dict[str, object] = {}
@@ -150,8 +150,9 @@ async def fake_rollout(task: Task, agent: Agent, **kwargs: object) -> Run:
 
     monkeypatch.setattr(taskset_mod, "rollout", fake_rollout)
 
-    v = Task(env="hosted-env", id="solve", args={"n": 1})
-    job = await v.run(cast("Agent", object()))
+    task = Task(env="hosted-env", id="solve", args={"n": 1})
+    taskset = taskset_mod.Taskset("hosted", [task], origin="api:ts_123")
+    job = await taskset.run(cast("Agent", object()))
 
     (run,) = job.runs
     assert run.trace.status == "completed"
diff --git a/hud/tests/test_init.py b/hud/tests/test_init.py
index b11dc88d2..144bed28f 100644
--- a/hud/tests/test_init.py
+++ b/hud/tests/test_init.py
@@ -55,6 +55,7 @@ def test_all_exports_available(self):
             "RuntimeLimits",
             "RuntimeResources",
             "LocalRuntime",
+            "SubprocessRuntime",
             "SyncPlan",
             "Task",
             "Taskset",
diff --git a/hud/tests/test_init_module.py b/hud/tests/test_init_module.py
index 45b458642..79a470974 100644
--- a/hud/tests/test_init_module.py
+++ b/hud/tests/test_init_module.py
@@ -35,6 +35,7 @@ def test_all_exports(self):
             "RuntimeLimits",
             "RuntimeResources",
             "LocalRuntime",
+            "SubprocessRuntime",
             "SyncPlan",
             "Task",
             "Taskset",