diff --git a/docs/session-isolation.md b/docs/session-isolation.md
new file mode 100644
index 00000000..b39ba667
--- /dev/null
+++ b/docs/session-isolation.md
@@ -0,0 +1,84 @@
+# Session isolation
+
+[Chinese](session-isolation.zh-CN.md)
+
+Team and Workflow calls can select an execution environment strategy. Session
+messages remain private to each agent. The strategy controls the workspace and
+tool execution environment.
+
+`shared` reuses the parent environment and directory. `worktree` creates an
+independent checkout inside the parent execution environment. Worktrees share
+the operating system and Git object storage. A container-backed parent keeps
+its children inside that container. Existing non-Git copy behavior is preserved.
+
+## Choosing a strategy
+
+```python
+from opencollab import OpenCollab
+
+client = OpenCollab(".")
+
+async def solve(ctx, inputs):
+    answer = await ctx.agent(inputs["task"], tools="coding")
+    summary = await ctx.agent("Summarize the task", tools=[], isolation="shared")
+    return {"answer": answer, "summary": summary}
+
+result = await client.workflow(solve, {"task": "Inspect the code"}, isolation="worktree")
+```
+
+Workflow agents inherit the run strategy when `isolation` is omitted.
+`isolation=True` selects `worktree`; `False` selects `shared`. Omitting the run
+strategy preserves the historical shared Workflow default, including existing
+worktree requirements for a clean source checkout. `ctx.agent_run` and candidate
+child workflows use the same selection and cleanup path.
+
+A Team file can select a default and role overrides.
+
+```yaml
+entry: manager
+isolation: worktree
+roles:
+  manager:
+    prompt: Coordinate the task.
+    isolation: shared
+    tools: [spawn_agent]
+  coder:
+    prompt: Implement the requested change.
+    tools: [file_read, file_write]
+topology:
+  manager: [coder]
+```
+
+`client.team(..., isolation="shared")` overrides the Team-file default. Role
+settings take precedence over the run default. With every new setting omitted,
+`use_worktrees` keeps its existing behavior and the entry agent retains its
+original workspace. The CLI reads these same Team-file settings.
+
+An explicitly isolated entry agent keeps its full patch in the run artifacts
+before the workspace is reclaimed. The SDK returns its path in
+`result.metrics["entry_patch"]`. Trace-free runs also save this patch; without
+an artifact directory, OC creates a recovery directory. Applying the patch to
+the caller's source remains an explicit operation.
+
+The Workflow CLI accepts `--isolation shared` or `--isolation worktree`.
+
+## Custom environment providers
+
+Pass `environment_providers={"custom": factory}` to `OpenCollab`, then select
+`isolation="custom"`. Each factory receives `workspace`, `base_environment`,
+`owned_save_dir`, `enable_adoption`, and `owned_output_paths` as keyword arguments.
+It creates a fresh pool for that run and strategy, implementing the public
+`opencollab.environments.EnvironmentPool` protocol. The existing operations are
+`acquire(role, source=...)`, `release_env(environment)`, `release()`, and
+`adopt(patch)`.
+
+The source passed to `acquire` is the actual parent or review workspace. An
+environment returned by the provider is ready for tool execution. The provider
+owns allocations it creates; the supplied source remains borrowed. Cleanup
+waits for pending acquisitions and preserves recovery material when it fails.
+OC retries retained cleanup through the same pool. Tools continue to use their
+existing permission and workspace checks.
+
+Unknown strategies report an error instead of selecting a different execution
+environment. Model endpoints and request concurrency are configured separately
+from workspace isolation.
diff --git a/docs/session-isolation.zh-CN.md b/docs/session-isolation.zh-CN.md
new file mode 100644
index 00000000..6ffecfa8
--- /dev/null
+++ b/docs/session-isolation.zh-CN.md
@@ -0,0 +1,57 @@
+# Session 隔离
+
+[English](session-isolation.md)
+
+Team 和 Workflow 可以选择执行环境策略。各 Agent 的对话状态独立保存，隔离策略控制工作目录和工具执行环境。
+
+`shared` 复用父任务的环境与目录。`worktree` 在父执行环境中创建独立工作树，工作树之间共享操作系统和 Git 对象存储。父任务使用容器时，子工作树继续位于该容器中。非 Git 目录沿用已有的副本处理方式。
+
+## 选择策略
+
+```python
+from opencollab import OpenCollab
+
+client = OpenCollab(".")
+
+async def solve(ctx, inputs):
+    answer = await ctx.agent(inputs["task"], tools="coding")
+    summary = await ctx.agent("Summarize the task", tools=[], isolation="shared")
+    return {"answer": answer, "summary": summary}
+
+result = await client.workflow(solve, {"task": "Inspect the code"}, isolation="worktree")
+```
+
+省略单次调用的 `isolation` 时，Workflow Agent 继承运行默认策略。`True` 选择 `worktree`，`False` 选择 `shared`。运行默认策略也省略时，保留原有的共享环境行为，已有工作树对干净源目录的要求继续生效。`ctx.agent_run` 和候选子工作流采用相同的选择与清理过程。
+
+Team 配置文件可以声明默认策略，并为角色覆盖。
+
+```yaml
+entry: manager
+isolation: worktree
+roles:
+  manager:
+    prompt: Coordinate the task.
+    isolation: shared
+    tools: [spawn_agent]
+  coder:
+    prompt: Implement the requested change.
+    tools: [file_read, file_write]
+topology:
+  manager: [coder]
+```
+
+`client.team(..., isolation="shared")` 覆盖 Team 文件的默认策略，角色设置优先于运行默认值。新选项全部省略时，`use_worktrees` 保留原语义，入口 Agent 继续使用原工作目录。CLI 读取相同的 Team 文件设置。
+
+显式隔离的入口 Agent 在工作区回收前保存完整补丁。SDK 通过 `result.metrics["entry_patch"]` 返回补丁位置。关闭轨迹记录时也会保存补丁，调用方省略产物目录时则创建恢复目录。调用方可以显式将补丁应用到源仓库。
+
+Workflow CLI 支持 `--isolation shared` 和 `--isolation worktree`。
+
+## 自定义环境提供器
+
+向 `OpenCollab` 传入 `environment_providers={"custom": factory}`，随后使用 `isolation="custom"`。每个工厂通过关键字参数接收 `workspace`、`base_environment`、`owned_save_dir`、`enable_adoption` 和 `owned_output_paths`。
+
+工厂为每次运行中的该策略创建一个环境池，实现公开的 `opencollab.environments.EnvironmentPool` 接口。操作包括 `acquire(role, source=...)`、`release_env(environment)`、`release()` 和 `adopt(patch)`。
+
+`acquire` 收到当前父任务或审阅任务的真实工作环境，返回已准备好执行工具的环境。提供器负责自己分配的资源，传入的源环境保持借用关系。清理等待在途的环境申请完成，并在失败时保留可恢复材料，由同一个池继续处理清理。工具沿用已有权限和工作区检查。
+
+未知策略会返回错误。模型端点和请求并发采用独立配置。
diff --git a/opencollab/adapters/cli/main.py b/opencollab/adapters/cli/main.py
index 10f5738d..7c9fd261 100644
--- a/opencollab/adapters/cli/main.py
+++ b/opencollab/adapters/cli/main.py
@@ -331,7 +331,7 @@ async def _run(
         TuiPermissionPolicy,
     )
     from opencollab.application.event_bus import EventBus
-    from opencollab.bootstrap import build_runtime_context, build_scheduler
+    from opencollab.bootstrap import build_runtime_context, build_scheduler, prepare_scheduler
 
     tui = TUI(console, filter_messages=cfg["filter_messages"])
     # A redirected or piped run has no bottom region to own: nothing is painted
@@ -376,8 +376,8 @@ async def _run(
     tracer_failure: BaseException | None = None
     repeated_cancellation: asyncio.CancelledError | None = None
     try:
-        scheduler = build_scheduler(
-            ctx, use_worktrees=use_worktrees, interactive=True,
+        scheduler = await prepare_scheduler(
+            ctx, builder=build_scheduler, use_worktrees=use_worktrees, interactive=True,
             session_file=session_file, auto_save=True,
             team_config_path=team_config_path,
             allow_unisolated_child_shell=allow_local_child_shell,
diff --git a/opencollab/adapters/cli/workflow.py b/opencollab/adapters/cli/workflow.py
index bf8ef3d0..360ab12d 100644
--- a/opencollab/adapters/cli/workflow.py
+++ b/opencollab/adapters/cli/workflow.py
@@ -96,6 +96,7 @@ def run_cmd(
         "--task-concurrency",
         help="Max active parallel/pipeline units (defaults to --concurrency)",
     ),
+    isolation: Optional[str] = typer.Option(None, "--isolation", help="Default agent environment strategy"),
     agent_profile: Optional[str] = typer.Option(
         None,
         "--agent-profile",
@@ -167,6 +168,7 @@ def run_cmd(
             max_concurrency=concurrency,
             task_concurrency=task_concurrency,
             agent_profile=agent_profile,
+            **({"isolation": isolation} if isolation is not None else {}),
             save_dir=save_dir,
             trace=trace,
             return_details=True,
diff --git a/opencollab/adapters/session_environments.py b/opencollab/adapters/session_environments.py
new file mode 100644
index 00000000..a65fd0e2
--- /dev/null
+++ b/opencollab/adapters/session_environments.py
@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable, Mapping, Sequence
+from typing import Any
+
+from opencollab.adapters.env import LocalEnvironment
+from opencollab.adapters.worktree_pool import WorktreePool
+from opencollab.application.async_timeout import await_owned_operation
+from opencollab.application.isolation import isolation_name
+from opencollab.application.ports import EnvironmentPort, WorktreePoolPort
+
+
+class SessionEnvironmentPool:
+    """Own per-run strategy pools; injected factories create a fresh pool per strategy."""
+
+    inherit_parent_environment = True
+
+    def __init__(
+        self, workspace: str, *, isolation: str = "worktree",
+        base_environment: EnvironmentPort | None = None,
+        providers: Mapping[str, Callable[..., WorktreePoolPort]] | None = None,
+        roles: Mapping[str, str | None] | None = None,
+        owned_save_dir: str | None = None, enable_adoption: bool = False,
+        owned_output_paths: Sequence[str] = (),
+    ) -> None:
+        self._workspace = workspace
+        self._default = isolation_name(isolation)
+        self._providers = dict(providers or {})
+        self._roles = dict(roles or {})
+        self._options = dict(owned_save_dir=owned_save_dir, enable_adoption=enable_adoption,
+                             owned_output_paths=owned_output_paths)
+        self._pools: dict[str, WorktreePoolPort] = {}
+        self._owners: dict[int, tuple[EnvironmentPort, WorktreePoolPort]] = {}
+        self._references: dict[int, int] = {}
+        self._lifecycle = asyncio.Condition()
+        self._inflight = 0
+        self._closed = False
+        for name in (self._default, *self._roles.values()):
+            if name is not None:
+                self._validate(name)
+        self._source = base_environment or LocalEnvironment(workspace)
+        self._owned_source = self._source if base_environment is None else None
+
+    def _validate(self, name: str) -> None:
+        isolation_name(name)
+        if name not in {"shared", "worktree"} and name not in self._providers:
+            raise ValueError(f"isolation strategy {name!r} needs an environment provider")
+
+    def _pool(self, name: str, source: EnvironmentPort) -> WorktreePoolPort:
+        self._validate(name)
+        key = name
+        if key not in self._pools:
+            factory = self._providers.get(name)
+            kwargs = dict(workspace=self._workspace, base_environment=source, **self._options)
+            pool = (factory(**kwargs) if factory is not None else
+                    WorktreePool(use_worktrees=name == "worktree", **kwargs))
+            self._pools[key] = pool
+        return self._pools[key]
+
+    async def acquire(
+        self, role: str, *, source: EnvironmentPort | None = None,
+        isolation: str | bool | None = None,
+    ) -> Any:
+        async with self._lifecycle:
+            if self._closed:
+                raise RuntimeError("session environment pool is closed")
+            self._inflight += 1
+        try:
+            name = isolation_name(isolation, default=self._roles.get(role) or self._default)
+            base = self._source if source is None else source
+            pool = self._pool(name, base)
+            env = await pool.acquire(role, source=base)
+            if env is not base or id(env) in self._owners:
+                self._owners.setdefault(id(env), (env, pool))
+                self._references[id(env)] = self._references.get(id(env), 0) + 1
+            if self._closed:
+                await self.release_env(env)
+                raise RuntimeError("session environment pool closed during acquisition")
+            return env
+        finally:
+            async with self._lifecycle:
+                self._inflight -= 1
+                self._lifecycle.notify_all()
+
+    def set_source(self, environment: EnvironmentPort) -> None:
+        self._source = environment
+
+    async def adopt(self, patch: str) -> None:
+        pool = self._delivery_pool()
+        await pool.adopt(patch)
+
+    def _delivery_pool(self):
+        key = "_delivery"
+        if key not in self._pools:
+            self._pools[key] = WorktreePool(
+                self._workspace, use_worktrees=True, base_environment=self._source, **self._options,
+            )
+        return self._pools[key]
+
+    async def adopt_captured(self, patch, physical_paths):
+        await self._delivery_pool().adopt_captured(patch, physical_paths)
+
+    async def release_env(self, env: EnvironmentPort) -> None:
+        owner = self._owners.get(id(env))
+        if owner is not None:
+            if self._references[id(env)] > 1:
+                self._references[id(env)] -= 1
+                return
+            await owner[1].release_env(env)
+            self._owners.pop(id(env), None)
+            self._references.pop(id(env), None)
+
+    async def release(self) -> None:
+        self._closed = True
+        await await_owned_operation(self._release(), propagate_cancellation=True)
+
+    async def _release(self) -> None:
+        async with self._lifecycle:
+            while self._inflight:
+                await self._lifecycle.wait()
+        # A later environment may use an earlier strategy's workspace as its source.
+        for env, _pool in reversed(tuple(self._owners.values())):
+            self._references[id(env)] = 1
+            await self.release_env(env)
+        failures = []
+        for key, pool in reversed(tuple(self._pools.items())):
+            try:
+                await pool.release()
+            except Exception as exc:
+                failures.append(exc)
+                break
+            else:
+                del self._pools[key]
+        if not failures and self._owned_source is not None:
+            await self._owned_source.cleanup()
+            self._owned_source = None
+        if failures:
+            raise OSError("session environment cleanup failed: " + "; ".join(map(str, failures)))
+        self._owners.clear()
diff --git a/opencollab/application/_scheduler_team.py b/opencollab/application/_scheduler_team.py
index 66b89449..48de8bf2 100644
--- a/opencollab/application/_scheduler_team.py
+++ b/opencollab/application/_scheduler_team.py
@@ -576,6 +576,9 @@ async def _append_worktree_diff(
         if not diff:
             return result
         recovery = ""
+        if aid == 0 and self.preserve_entry_changes:
+            self.entry_patch_path = await self._preserve_adoption_candidate(aid, diff)
+            recovery = f"\n\n[Entry changes saved at {self.entry_patch_path}]"
         if len(diff) > WORKTREE_DIFF_MAX_CHARS:
             if partial and isinstance(env, RecoverableChangesPort):
                 recovery = f"\n\n[Full partial changes retained at {env.retain_changes()}]"
diff --git a/opencollab/application/isolation.py b/opencollab/application/isolation.py
new file mode 100644
index 00000000..0f697d77
--- /dev/null
+++ b/opencollab/application/isolation.py
@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def isolation_name(value: Any, *, default: str | None = None) -> str | None:
+    if value is None:
+        return default
+    if isinstance(value, bool):
+        return "worktree" if value else "shared"
+    if not isinstance(value, str) or not value or value != value.strip():
+        raise ValueError("isolation must be a strategy name, boolean, or None")
+    if any(not (character.isalnum() or character in "-_") for character in value):
+        raise ValueError("isolation strategy names may contain letters, numbers, '-' and '_'")
+    return value
diff --git a/opencollab/application/scheduler.py b/opencollab/application/scheduler.py
index d92a4264..5adb0297 100644
--- a/opencollab/application/scheduler.py
+++ b/opencollab/application/scheduler.py
@@ -82,6 +82,8 @@ def __init__(
         entry_role: str | None = None,
         candidate_diff_writer: Callable[[int, str, str], Awaitable[str]] | None = None,
     ):
+        self.preserve_entry_changes = False
+        self.entry_patch_path: str | None = None
         self._session_factory = session_factory
         self._worktree_pool = worktree_pool
         self._event_sink = event_sink
diff --git a/opencollab/application/scheduler_lifecycle.py b/opencollab/application/scheduler_lifecycle.py
index 289fbc97..8aeea9fd 100644
--- a/opencollab/application/scheduler_lifecycle.py
+++ b/opencollab/application/scheduler_lifecycle.py
@@ -170,9 +170,12 @@ async def spawn(
                     "Cannot spawn agent: team token budget is fully allocated."
                 )
 
+            source = review_source
+            if source is None and getattr(self._worktree_pool, "inherit_parent_environment", False) is True:
+                source = self._sessions[parent_aid].env
             env = (
-                await self._worktree_pool.acquire(role, source=review_source)
-                if review_source is not None
+                await self._worktree_pool.acquire(role, source=source)
+                if source is not None
                 else await self._worktree_pool.acquire(role)
             )
             self._startup_envs[aid] = env
diff --git a/opencollab/application/workflow.py b/opencollab/application/workflow.py
index db1c81b5..d7ff7943 100644
--- a/opencollab/application/workflow.py
+++ b/opencollab/application/workflow.py
@@ -40,6 +40,7 @@
 from opencollab.application.async_timeout import (
     abandon_on_timeout as abandon_on_timeout,
 )
+from opencollab.application.isolation import isolation_name
 from opencollab.application.ports import (
     CandidateWorkspacePort,
     EventPublisherPort,
@@ -335,7 +336,7 @@ async def _build_workflow_session(
         label: str | None,
         budget: int | None,
         tools: Sequence[Any] | None = None,
-        isolation: bool = False,
+        isolation: bool | str | None = None,
         **kwargs: Any,
     ) -> Any:
         """Build one one-shot session and record that this agent existed.
@@ -355,7 +356,17 @@ async def _build_workflow_session(
         Emitted after the build, because an agent that failed to build has no
         id to attribute anything to; the caller records that failure instead.
         """
-        env = await self._factory.acquire_isolated_env(label=label) if isolation else None
+        default_isolation = getattr(self._factory, "default_isolation", "shared")
+        explicit_strategy = isinstance(isolation, str) or default_isolation != "shared"
+        strategy = isolation_name(isolation, default=default_isolation)
+        acquire = getattr(self._factory, "acquire_environment", None)
+        if callable(acquire):
+            env = await acquire(isolation=strategy, label=label)
+        elif strategy in {"shared", "worktree"}:
+            env = await self._factory.acquire_isolated_env(label=label) if strategy == "worktree" else None
+        else:
+            raise ValueError(f"session factory does not support isolation strategy {strategy!r}")
+        isolation = strategy != "shared"
         session = self._factory.build_workflow_session(
             label=label,
             budget=budget,
@@ -380,6 +391,7 @@ async def _build_workflow_session(
                 # An isolated writer needs a recorded handoff because siblings cannot see
                 # its edits through the shared filesystem.
                 "isolated": bool(isolation),
+                **({"isolation": strategy} if explicit_strategy else {}),
             },
         )
         return session
@@ -495,7 +507,7 @@ async def agent(
         schema: dict[str, Any] | None = None,
         label: str | None = None,
         tools: Sequence[Any] | None = None,
-        isolation: bool = False,
+        isolation: bool | str | None = None,
         tool_choice: Any = None,
         thinking: bool | None = None,
         timeout: float | None = None,
@@ -627,7 +639,7 @@ async def _run_agent(
         *,
         label: str | None,
         tools: Sequence[Any] | None,
-        isolation: bool,
+        isolation: bool | str | None,
         tool_choice: Any = None,
         thinking: bool | None = None,
         timeout: float | None = None,
diff --git a/opencollab/application/workflow_agent_run.py b/opencollab/application/workflow_agent_run.py
index 843e5f80..9cda034a 100644
--- a/opencollab/application/workflow_agent_run.py
+++ b/opencollab/application/workflow_agent_run.py
@@ -58,7 +58,7 @@ async def agent_run(
         *,
         label: str | None = None,
         tools: str | Sequence[Any] | None = None,
-        isolation: bool = False,
+        isolation: bool | str | None = None,
         budget: int | None = None,
         timeout: float | None = None,
         max_steps: int | None = None,
diff --git a/opencollab/application/workflow_agents.py b/opencollab/application/workflow_agents.py
index 9abb4856..6a741a57 100644
--- a/opencollab/application/workflow_agents.py
+++ b/opencollab/application/workflow_agents.py
@@ -39,7 +39,7 @@ async def _run_enforced_agent(
         *,
         label: str | None,
         tools: Sequence[Any] | None,
-        isolation: bool,
+        isolation: bool | str | None,
         tool_choice: Any,
         thinking: bool | None,
         timeout: float | None,
diff --git a/opencollab/application/workflow_candidates.py b/opencollab/application/workflow_candidates.py
index a8090c0c..5c49ca7d 100644
--- a/opencollab/application/workflow_candidates.py
+++ b/opencollab/application/workflow_candidates.py
@@ -133,6 +133,26 @@ def environment_revoked(self) -> bool:
     def has_pending_isolated_cleanup(self) -> bool:
         return bool(self._isolated_environments)
 
+    @property
+    def default_isolation(self):
+        return getattr(self._factory, "default_isolation", "shared")
+
+    async def acquire_environment(self, *, isolation, label=None, environment=None):
+        owner = self._environment if environment is None else environment
+        if isolation == "shared":
+            return owner
+        acquire = getattr(self._factory, "acquire_environment", None)
+        if callable(acquire):
+            release = getattr(self._factory, "release_isolated_envs", None)
+            if not callable(release) or not _accepts_environment(release):
+                raise TypeError("candidate isolation requires environment-aware acquisition and scoped release")
+            isolated = await acquire(isolation=isolation, label=label, environment=owner)
+            self._isolated_environments.append((owner, isolated))
+            return isolated
+        if isolation == "worktree":
+            return await self.acquire_isolated_env(label=label, environment=owner)
+        raise ValueError(f"session factory does not support isolation strategy {isolation!r}")
+
     async def acquire_isolated_env(
         self, *, label: str | None = None, environment: Any | None = None,
     ) -> Any:
diff --git a/opencollab/application/workflow_structured.py b/opencollab/application/workflow_structured.py
index a00df06e..c58cb0b3 100644
--- a/opencollab/application/workflow_structured.py
+++ b/opencollab/application/workflow_structured.py
@@ -74,7 +74,7 @@ async def _run_structured_agent(
         schema: dict[str, Any],
         label: str | None,
         tools: Sequence[Any] | None,
-        isolation: bool,
+        isolation: bool | str | None,
         timeout: float | None = None,
         budget: int | None = None,
     ) -> dict | None:
@@ -177,7 +177,7 @@ async def _forced_structured_commit(
         *,
         schema: dict[str, Any],
         label: str | None,
-        isolation: bool,
+        isolation: bool | str | None,
         timeout: float | None,
         budget: int | None,
     ) -> dict | None:
diff --git a/opencollab/bootstrap/__init__.py b/opencollab/bootstrap/__init__.py
index a1d46512..f28d2fd3 100644
--- a/opencollab/bootstrap/__init__.py
+++ b/opencollab/bootstrap/__init__.py
@@ -21,6 +21,7 @@
     load_session,
     snapshot_session,
 )
+from opencollab.bootstrap.isolation import prepare_scheduler
 
 __all__ = [
     "AgentRuntimeLifecycleError",
@@ -40,6 +41,7 @@
     "build_workspace_safety_policy",
     "get_config",
     "load_session",
+    "prepare_scheduler",
     "run_agent",
     "snapshot_session",
 ]
diff --git a/opencollab/bootstrap/_workflow_runtime_execution.py b/opencollab/bootstrap/_workflow_runtime_execution.py
index 2a1dcfe0..0d82f927 100644
--- a/opencollab/bootstrap/_workflow_runtime_execution.py
+++ b/opencollab/bootstrap/_workflow_runtime_execution.py
@@ -95,6 +95,8 @@ async def run_workflow(
     llm: Any | None = None,
     session_builder: Callable[..., Any] | None = None,
     limit_mode: str = "environment",
+    isolation: str | None = None,
+    environment_providers: Any | None = None,
 ) -> Any:
     """Run one workflow and return only after cleanup and evidence persistence."""
     cleanup_timeout = _positive_cleanup_timeout(cleanup_timeout)
@@ -134,6 +136,7 @@ async def run_workflow(
             llm=llm,
             session_builder=session_builder,
             limit_mode=limit_mode,
+            isolation=isolation, environment_providers=environment_providers,
             run_id=run_id,
             candidate_workspace=candidate_workspace,
             save_dir=save_dir,
diff --git a/opencollab/bootstrap/_workflow_runtime_session.py b/opencollab/bootstrap/_workflow_runtime_session.py
index 59384512..98fd70b8 100644
--- a/opencollab/bootstrap/_workflow_runtime_session.py
+++ b/opencollab/bootstrap/_workflow_runtime_session.py
@@ -15,8 +15,10 @@
     DEFAULT_MAX_OUTPUT_TOKENS,
     model_capabilities,
 )
+from opencollab.adapters.session_environments import SessionEnvironmentPool
 from opencollab.adapters.working_tree import MAX_WORKING_TREE_DIFF_CHARS, EnvWorkingTreeProbe
 from opencollab.adapters.worktree_pool import WorktreePool
+from opencollab.application.isolation import isolation_name
 from opencollab.application.ports import EventPublisherPort, TracePort
 from opencollab.application.tool_execution_runtime import ToolRuntime
 from opencollab.application.workflow import WorkflowContext
@@ -110,10 +112,16 @@ def __init__(
         llm: Any | None = None,
         session_builder: Callable[..., Any] | None = None,
         run_id: str | None = None,
+        isolation: str | None = None,
+        environment_providers: Any | None = None,
     ) -> None:
         self._llm = llm
         self._session_builder = session_builder
         self._run_id = run_id
+        self.default_isolation = isolation_name(isolation, default="shared")
+        self._environment_providers = environment_providers
+        self._strategy_pool = None
+        self._strategy_envs = []
         self._model = model
         self._provider = provider
         self._wire_protocol = wire_protocol
@@ -195,6 +203,23 @@ def _save_path(self, aid: int, label: str | None) -> str | None:
             return None
         return workflow_transcript_path(self._save_dir, aid, label)
 
+    async def acquire_environment(self, *, isolation: str, label=None, environment=None):
+        if isolation == "shared":
+            return environment
+        if isolation == "worktree" and not self._environment_providers:
+            if environment is None:
+                return await self.acquire_isolated_env(label=label)
+            return await self.acquire_isolated_env(label=label, environment=environment)
+        if self._strategy_pool is None:
+            self._strategy_pool = SessionEnvironmentPool(
+                self._workspace or ".", isolation=self.default_isolation,
+                base_environment=self._env, providers=self._environment_providers,
+                owned_save_dir=self._save_dir,
+            )
+        env = await self._strategy_pool.acquire(label or "workflow-agent", source=environment, isolation=isolation)
+        self._strategy_envs.append((environment, env))
+        return env
+
     async def acquire_isolated_env(
         self, *, label: str | None = None, environment: Any | None = None
     ) -> Any:
@@ -228,6 +253,19 @@ async def acquire_isolated_env(
     async def release_isolated_envs(self, *, environment: Any | None = None) -> None:
         """Release each owned environment and retain failed leases for retry."""
         errors: list[Exception] = []
+        if self._strategy_pool is not None:
+            for owner, env in tuple(self._strategy_envs):
+                if environment is not None and owner is not environment:
+                    continue
+                try:
+                    await self._strategy_pool.release_env(env)
+                except Exception as exc:
+                    errors.append(exc)
+                else:
+                    self._strategy_envs.remove((owner, env))
+            if environment is None and not errors:
+                await self._strategy_pool.release()
+                self._strategy_pool = None
         for owner, lease in tuple(self._candidate_isolation_leases):
             if environment is not None and owner is not environment:
                 continue
@@ -455,6 +493,8 @@ def build_workflow_context(
     llm: Any | None = None,
     session_builder: Callable[..., Any] | None = None,
     limit_mode: str = "environment",
+    isolation: str | None = None,
+    environment_providers: Any | None = None,
     run_id: str | None = None,
 ) -> WorkflowContext:
     """Build a :class:`WorkflowContext` wired to the concrete session factory.
@@ -473,6 +513,7 @@ def build_workflow_context(
         llm=llm,
         session_builder=session_builder,
         run_id=run_id,
+        isolation=isolation, environment_providers=environment_providers,
         model=cfg["model"],
         provider=cfg["provider"],
         wire_protocol=cfg.get("wire_protocol", "chat_completions"),
diff --git a/opencollab/bootstrap/isolation.py b/opencollab/bootstrap/isolation.py
new file mode 100644
index 00000000..8e566d35
--- /dev/null
+++ b/opencollab/bootstrap/isolation.py
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import Any
+
+from opencollab.adapters.session_environments import SessionEnvironmentPool
+from opencollab.application.isolation import isolation_name
+from opencollab.bootstrap.team_config import load_team_config
+
+
+async def prepare_scheduler(ctx: Any, *, builder: Any, isolation=None,
+                            environment_providers=None, **kwargs: Any) -> Any:
+    team = kwargs.get("resolved_team_config") or load_team_config(
+        ctx.workspace, path=kwargs.get("team_config_path"),
+    )
+    selected = isolation_name(isolation, default=team.isolation)
+    roles = {name: role.isolation for name, role in team.roles.items() if role.isolation is not None}
+    if selected is None and not roles:
+        return builder(ctx, **kwargs)
+    pool = SessionEnvironmentPool(
+        ctx.workspace, isolation=selected or ("worktree" if kwargs["use_worktrees"] else "shared"),
+        base_environment=kwargs.get("environment"), providers=environment_providers, roles=roles,
+        owned_save_dir=kwargs.get("save_dir"),
+        enable_adoption=any("adopt_agent" in role.tools for role in team.roles.values()),
+        owned_output_paths=getattr(ctx.tracer, "owned_paths", ()),
+    )
+    try:
+        # An unset strategy keeps the historical entry workspace even when a child overrides it.
+        entry = await pool.acquire(team.entry, isolation=roles.get(team.entry) or selected or "shared")
+        pool.set_source(entry)
+        scheduler = builder(ctx, **{**kwargs, "environment": entry, "worktree_pool": pool,
+                                    "resolved_team_config": team})
+        scheduler.preserve_entry_changes = (roles.get(team.entry) or selected or "shared") != "shared"
+        return scheduler
+    except BaseException:
+        await pool.release()
+        raise
diff --git a/opencollab/bootstrap/programmatic.py b/opencollab/bootstrap/programmatic.py
index 0bcc8d43..1a236e50 100644
--- a/opencollab/bootstrap/programmatic.py
+++ b/opencollab/bootstrap/programmatic.py
@@ -501,6 +501,8 @@ async def run_workflow(
     run_id: str | None = None,
     llm: Any | None = None,
     limit_mode: str = "environment",
+    isolation: str | None = None,
+    environment_providers: Mapping[str, Any] | None = None,
 ) -> ProgrammaticResult:
     """Run one workflow and return its live metrics directly."""
     run_id = run_id if run_id is not None else f"workflow-{uuid.uuid4().hex}"
@@ -545,6 +547,7 @@ async def run_workflow(
                 agent_profile=agent_profile,
                 llm=llm,
                 limit_mode=limit_mode,
+                isolation=isolation, environment_providers=environment_providers,
                 candidate_workspace=candidate_workspace,
                 run_id=run_id,
                 return_details=True,
@@ -731,6 +734,8 @@ async def run_team(
     artifacts: Path | None,
     trace: bool,
     use_worktrees: bool,
+    isolation: str | None = None,
+    environment_providers: Mapping[str, Any] | None = None,
     prebuild_team: bool = False,
     allow_unisolated_shell: bool | None = None,
     max_steps: int | None = SESSION_MAX_STEPS,
@@ -761,6 +766,7 @@ async def run_team(
         artifacts=artifacts,
         trace=trace,
         use_worktrees=use_worktrees,
+        isolation=isolation, environment_providers=environment_providers,
         prebuild_team=prebuild_team,
         allow_unisolated_shell=allow_unisolated_shell,
         max_steps=max_steps,
diff --git a/opencollab/bootstrap/programmatic_team.py b/opencollab/bootstrap/programmatic_team.py
index 8127038c..2371e4e8 100644
--- a/opencollab/bootstrap/programmatic_team.py
+++ b/opencollab/bootstrap/programmatic_team.py
@@ -14,6 +14,7 @@
 from opencollab.application.exception_notes import add_exception_note
 from opencollab.application.scheduler_types import SchedulerTurnError
 from opencollab.bootstrap import programmatic as _programmatic
+from opencollab.bootstrap.isolation import prepare_scheduler
 from opencollab.bootstrap.programmatic import (
     DEFAULT_TEAM_CLEANUP_TIMEOUT_SECONDS,
     ProgrammaticResult,
@@ -36,6 +37,8 @@ async def run_team(
     artifacts: Path | None,
     trace: bool,
     use_worktrees: bool,
+    isolation: str | None = None,
+    environment_providers: Mapping[str, Any] | None = None,
     prebuild_team: bool = False,
     allow_unisolated_shell: bool | None = None,
     max_steps: int | None = SESSION_MAX_STEPS,
@@ -75,8 +78,9 @@ async def run_team(
             filename="trajectory.jsonl",
         )
     try:
-        scheduler = _programmatic.build_scheduler(
-            context,
+        scheduler = await prepare_scheduler(
+            context, builder=_programmatic.build_scheduler, isolation=isolation,
+            environment_providers=environment_providers,
             use_worktrees=use_worktrees,
             # Programmatic runs cannot answer ``ask_user``. The caller separately
             # chooses whether to allow an unsandboxed shell.
@@ -222,6 +226,7 @@ async def run_team(
             "run_id": run_id,
             "steps": int(getattr(lead, "step_count", 0)),
             "sessions": len(scheduler.table.entries),
+            **({"entry_patch": scheduler.entry_patch_path} if getattr(scheduler, "entry_patch_path", None) else {}),
             **({"tree_snapshots": [dict(row) for row in scheduler.delivery_tree_snapshots]}
                if record_delivery_tree else {}),
             # Cleanup also persists the terminal snapshot and releases owned
diff --git a/opencollab/bootstrap/scheduler_factory.py b/opencollab/bootstrap/scheduler_factory.py
index 25376a7a..dbd51a40 100644
--- a/opencollab/bootstrap/scheduler_factory.py
+++ b/opencollab/bootstrap/scheduler_factory.py
@@ -99,6 +99,7 @@ def build_scheduler(
     environment: Environment | None = None,
     record_delivery_tree: bool = False,
     run_id: str | None = None,
+    worktree_pool=None,
 ) -> Scheduler:
     """Compose a team and initialize agent 0 with the selected configuration.
 
@@ -203,7 +204,7 @@ def build_scheduler(
     )
     try:
         lead_environment = session_factory.lead_environment
-        worktree_pool = WorktreePool(
+        worktree_pool = worktree_pool if worktree_pool is not None else WorktreePool(
             ctx.workspace,
             use_worktrees=use_worktrees,
             base_environment=environment,
diff --git a/opencollab/bootstrap/team_config.py b/opencollab/bootstrap/team_config.py
index acdcf740..ef0a3395 100644
--- a/opencollab/bootstrap/team_config.py
+++ b/opencollab/bootstrap/team_config.py
@@ -30,6 +30,7 @@
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from opencollab.adapters.safe_files import read_regular_text
+from opencollab.application.isolation import isolation_name
 from opencollab.bootstrap.context_policy import ContextPolicy, resolve_context_policy
 from opencollab.bootstrap.tool_registry import (
     COORDINATION_TOOL_NAMES,
@@ -133,6 +134,12 @@ class RoleConfig(BaseModel):
     # tool caps. ``ContextBuilder`` appends the role card and applies tool caps.
     # ``DefaultSessionFactory`` wires the shaper and safety wrapper.
     profile: str | None = None
+    isolation: str | None = None
+
+    @field_validator("isolation", mode="before")
+    @classmethod
+    def _isolation_name(cls, value):
+        return isolation_name(value)
     # This role's own token allowance, from the team file (``budget.tokens`` at
     # the team level, overridable per role). ``None`` on every role leaves the
     # team on the shared ``per_agent_cap`` rule; see ``TeamConfig.role_budgets``.
@@ -214,6 +221,12 @@ class _RoleFileModel(BaseModel):
     thinking_params: dict | None = None
     tools: list[str] = Field(default_factory=list)
     profile: str | None = None
+    isolation: str | None = None
+
+    @field_validator("isolation", mode="before")
+    @classmethod
+    def _isolation_name(cls, value):
+        return isolation_name(value)
     budget: _BudgetFileModel | None = None
 
     @field_validator("profile")
@@ -275,6 +288,13 @@ class _TeamFileModel(BaseModel):
 
     model_config = ConfigDict(extra="forbid")
 
+    isolation: str | None = None
+
+    @field_validator("isolation", mode="before")
+    @classmethod
+    def _isolation_name(cls, value):
+        return isolation_name(value)
+
     roles: dict[str, _RoleFileModel] = Field(default_factory=dict)
     topology: dict[str, list[str]] = Field(default_factory=dict)
     hooks: dict[str, list[_HookActionFileModel]] = Field(default_factory=dict)
@@ -315,8 +335,10 @@ class TeamConfig:
     # Tool name -> constructor kwargs (output caps); applied by the registry.
     tool_limits: dict[str, dict[str, int]] = field(default_factory=dict)
     context: ContextPolicy = field(default_factory=ContextPolicy)
+    isolation: str | None = None
 
     def __post_init__(self) -> None:
+        object.__setattr__(self, "isolation", isolation_name(self.isolation))
         normalized_tool_limits = validate_tool_limits(self.tool_limits)
         normalized_roles: dict[str, RoleConfig] = {}
         role_names: dict[str, str] = {}
@@ -567,6 +589,7 @@ def _build_team_config(data: Any, base_dir: Path) -> TeamConfig:
             thinking_params=entry.thinking_params,
             tools=list(entry.tools),
             profile=entry.profile,
+            isolation=entry.isolation,
             budget_tokens=(
                 entry.budget.tokens
                 if entry.budget is not None
@@ -608,6 +631,7 @@ def _build_team_config(data: Any, base_dir: Path) -> TeamConfig:
         entry=_resolve_entry_role(model.entry, roles),
         tool_limits={name: dict(kwargs) for name, kwargs in model.tool_limits.items()},
         context=resolve_context_policy(model.context),
+        isolation=model.isolation,
     )
 
 def resolve_team_file(workspace: str | None = None) -> Path | None:
diff --git a/opencollab/bootstrap/workflow_runtime.py b/opencollab/bootstrap/workflow_runtime.py
index 8da527e5..ba977377 100644
--- a/opencollab/bootstrap/workflow_runtime.py
+++ b/opencollab/bootstrap/workflow_runtime.py
@@ -97,6 +97,8 @@ async def run_workflow(
     llm: Any | None = None,
     session_builder: Callable[..., Any] | None = None,
     limit_mode: str = "environment",
+    isolation: str | None = None,
+    environment_providers: Any | None = None,
 ) -> Any:
     """Run through one owned lifecycle with an optional wall-clock deadline."""
     if isinstance(agent_profile, str):
@@ -129,6 +131,7 @@ async def run_owner() -> Any:
                 llm=llm,
                 session_builder=session_builder,
                 limit_mode=limit_mode,
+                isolation=isolation, environment_providers=environment_providers,
                 candidate_workspace=candidate_workspace,
                 run_id=run_id,
                 save_dir=save_dir,
diff --git a/opencollab/environments.py b/opencollab/environments.py
index 63493366..61079ba3 100644
--- a/opencollab/environments.py
+++ b/opencollab/environments.py
@@ -1,6 +1,7 @@
 """Public execution-environment contract and thin composition helpers."""
 
 from opencollab.application.ports import EnvironmentPort as Environment
+from opencollab.application.ports import WorktreePoolPort as EnvironmentPool
 from opencollab.bootstrap.programmatic import (
     attach_container,
     build_repo_map_via_env,
@@ -11,6 +12,7 @@
 
 __all__ = [
     "Environment",
+    "EnvironmentPool",
     "attach_container",
     "build_repo_map_via_env",
     "docker_environment",
diff --git a/opencollab/sdk/client.py b/opencollab/sdk/client.py
index f328100d..74c9cf05 100644
--- a/opencollab/sdk/client.py
+++ b/opencollab/sdk/client.py
@@ -11,6 +11,7 @@
 from types import MappingProxyType
 from typing import Any
 
+from opencollab.application.isolation import isolation_name
 from opencollab.application.run_control import RunControl
 from opencollab.bootstrap.config import _optional_string, build_config, resolve_provider_base_url
 from opencollab.bootstrap.programmatic import (
@@ -116,6 +117,7 @@ def __init__(
         base_url: str | None = None,
         config: Mapping[str, Any] | None = None,
         environment: Any | None = None,
+        environment_providers: Mapping[str, Any] | None = None,
     ) -> None:
         raw_workspace = os.fspath(workspace)
         if not raw_workspace or "\x00" in raw_workspace:
@@ -144,6 +146,7 @@ def __init__(
             self._config["provider"], self._config["base_url"],
         )
         self._environment = environment
+        self._environment_providers = dict(environment_providers or {})
 
     @property
     def configuration(self) -> Mapping[str, Any]:
@@ -310,6 +313,7 @@ async def team(
         artifacts: str | os.PathLike[str] | None = None,
         trace: bool = True,
         use_worktrees: bool = True,
+        isolation: str | None = None,
         prebuild_team: bool = False,
         allow_unisolated_shell: bool | None = None,
         max_steps: int | None = SESSION_MAX_STEPS,
@@ -372,6 +376,9 @@ async def team(
                 artifacts=_path(artifacts, "artifacts"),
                 trace=trace,
                 use_worktrees=use_worktrees,
+                **({"isolation": isolation_name(isolation),
+                    "environment_providers": self._environment_providers}
+                   if isolation is not None or self._environment_providers else {}),
                 prebuild_team=prebuild_team,
                 allow_unisolated_shell=allow_unisolated_shell,
                 max_steps=resolved_team_max_steps,
@@ -404,6 +411,7 @@ async def workflow(
         run_id: str | None = None,
         llm: Any | None = None,
         limit_mode: str = "environment",
+        isolation: str | None = None,
     ) -> RunResult[Any]:
         """Run a workflow name, decorated function, or plain async function.
 
@@ -489,6 +497,9 @@ async def workflow(
                 agent_profile=resolved_agent_profile,
                 llm=llm,
                 limit_mode=limit_mode,
+                **({"isolation": isolation_name(isolation),
+                    "environment_providers": self._environment_providers}
+                   if isolation is not None or self._environment_providers else {}),
                 cleanup_timeout=_required_positive_timeout(
                     cleanup_timeout,
                     "cleanup_timeout",
diff --git a/tests/interfaces/test_cli_main_lifecycle.py b/tests/interfaces/test_cli_main_lifecycle.py
index d3cbc8b7..fa16485d 100644
--- a/tests/interfaces/test_cli_main_lifecycle.py
+++ b/tests/interfaces/test_cli_main_lifecycle.py
@@ -203,6 +203,11 @@ def install_cli_fakes(monkeypatch, scheduler, tracer):
     monkeypatch.setattr(tui_mod, "TuiEventSink", lambda tui: object())
     monkeypatch.setattr(bootstrap, "build_runtime_context", lambda *a, **k: ctx)
     monkeypatch.setattr(bootstrap, "build_scheduler", lambda *a, **k: scheduler)
+
+    async def prepare(ctx, *, builder, **kwargs):
+        return builder(ctx, **kwargs)
+
+    monkeypatch.setattr(bootstrap, "prepare_scheduler", prepare)
     return ctx
 
 def config():
diff --git a/tests/interfaces/test_sdk_api.py b/tests/interfaces/test_sdk_api.py
index 9528e939..e8c86f7f 100644
--- a/tests/interfaces/test_sdk_api.py
+++ b/tests/interfaces/test_sdk_api.py
@@ -50,7 +50,7 @@ def test_public_class_and_method_shapes_stay_lean() -> None:
             positional: {"workspace": "."},
             keyword: {
                 "model": None, "provider": None, "api_key": None,
-                "base_url": None, "config": None, "environment": None,
+                "base_url": None, "config": None, "environment": None, "environment_providers": None,
             },
         },
         sdk.OpenCollab.agent: {
@@ -72,7 +72,7 @@ def test_public_class_and_method_shapes_stay_lean() -> None:
             keyword: {
                 "config": None, "budget": None, "timeout": None,
                 "cleanup_timeout": 10.0, "artifacts": None, "trace": True,
-                "use_worktrees": True, "prebuild_team": False,
+                "use_worktrees": True, "prebuild_team": False, "isolation": None,
                 "allow_unisolated_shell": None, "max_steps": 100,
                 "serialize_turns": False, "record_delivery_tree": False,
                 "run_id": None,
@@ -85,7 +85,7 @@ def test_public_class_and_method_shapes_stay_lean() -> None:
                 "timeout": None, "max_steps": 100, "system_prompt": None,
                 "agent_profile": None, "cleanup_timeout": 2.0,
                 "artifacts": None, "trace": True, "candidate_workspace": None,
-                "run_id": None, "llm": None, "limit_mode": "environment",
+                "run_id": None, "llm": None, "limit_mode": "environment", "isolation": None,
             },
         },
     }
@@ -165,7 +165,7 @@ def test_advanced_capabilities_live_in_small_opt_in_modules() -> None:
         "profile_tool_names",
     ]
     assert environments.__all__ == [
-        "Environment",
+        "Environment", "EnvironmentPool",
         "attach_container",
         # A workspace an agent reads is not always on this host, so the listing
         # of it is asked of the environment rather than walked here.
diff --git a/tests/interfaces/test_session_isolation.py b/tests/interfaces/test_session_isolation.py
new file mode 100644
index 00000000..826854a4
--- /dev/null
+++ b/tests/interfaces/test_session_isolation.py
@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from opencollab import OpenCollab
+from opencollab.adapters.llm.types import LLMResponse, Usage
+from opencollab.adapters.worktree_pool import WorktreePool
+from opencollab.bootstrap import _workflow_runtime_session as wiring
+from opencollab.bootstrap import container
+from opencollab.tools import builtin_tools
+
+
+def _repo(path, *, dirty=True):
+    for args in (("init", "-q"), ("config", "user.name", "Test"),
+                 ("config", "user.email", "test@example.invalid")):
+        subprocess.run(["git", "-C", str(path), *args], check=True, capture_output=True)
+    (path / "seed.txt").write_text("initial")
+    subprocess.run(["git", "-C", str(path), "add", "."], check=True, capture_output=True)
+    subprocess.run(["git", "-C", str(path), "commit", "-qm", "initial"], check=True, capture_output=True)
+    if dirty:
+        (path / "seed.txt").write_text("dirty")
+
+
+def _call(name, args):
+    return {"id": "call", "type": "function", "function": {"name": name, "arguments": json.dumps(args)}}
+
+
+class Writer:
+    def __init__(self, **kwargs):
+        pass
+
+    def context_window(self):
+        return 100_000
+
+    async def close(self):
+        pass
+
+    async def complete(self, messages, **kwargs):
+        if any(m["role"] == "tool" for m in messages):
+            return LLMResponse(content="finished", usage=Usage(3, 2), finish_reason="stop")
+        name = "marker"
+        return LLMResponse(tool_calls=[_call("file_write", {
+            "mode": "create", "path": name + ".txt", "content": name,
+        })], usage=Usage(3, 2), finish_reason="tool_calls")
+
+
+@pytest.mark.parametrize("default, override, separate", [
+    (None, None, False), ("worktree", None, True), ("worktree", False, False),
+    ("shared", True, True), ("worktree", "shared", False), ("shared", "worktree", True),
+])
+async def test_workflow_defaults_and_per_call_overrides(tmp_path, monkeypatch, default, override, separate):
+    _repo(tmp_path, dirty=False)
+    sessions = []
+    original = wiring.build_session
+
+    def build(**kwargs):
+        session = original(**kwargs)
+        sessions.append(session)
+        return session
+
+    monkeypatch.setattr(wiring, "build_session", build)
+
+    async def flow(ctx, args):
+        result = await ctx.agent("marker", tools=list(builtin_tools("file_write")), isolation=override)
+        assert await sessions[-1].env.read_file("marker.txt") == "marker"
+        assert await sessions[-1].env.read_file("seed.txt") == "initial"
+        return result
+
+    result = await OpenCollab(tmp_path).workflow(flow, isolation=default, llm=Writer(), trace=False)
+    assert result.ok and result.output == "finished"
+    assert (tmp_path / "marker.txt").exists() is (not separate)
+    assert (tmp_path / "seed.txt").read_text() == "initial"
+
+
+async def test_custom_provider_is_fresh_per_workflow_and_owns_cleanup(tmp_path):
+    _repo(tmp_path)
+    created = []
+
+    class PrivatePool(WorktreePool):
+        def __init__(self, **kwargs):
+            super().__init__(use_worktrees=True, **kwargs)
+            self.released = False
+
+        async def release(self):
+            await super().release()
+            self.released = True
+
+    def provider(**kwargs):
+        pool = PrivatePool(**kwargs)
+        created.append(pool)
+        return pool
+
+    client = OpenCollab(tmp_path, environment_providers={"custom": provider})
+
+    async def flow(ctx, args):
+        return await ctx.agent("private", tools=list(builtin_tools("file_write")))
+
+    for _ in range(2):
+        result = await client.workflow(flow, isolation="custom", llm=Writer(), trace=False)
+        assert result.ok
+    assert len(created) == 2 and all(pool.released for pool in created)
+    assert not (tmp_path / "marker.txt").exists()
+
+
+@pytest.mark.parametrize("entry_isolation", ["shared", "worktree"])
+async def test_team_role_override_and_isolated_entry_patch(tmp_path, monkeypatch, entry_isolation):
+    _repo(tmp_path)
+    team = tmp_path / "team.yaml"
+    team.write_text("""entry: manager
+isolation: worktree
+roles:
+  manager:
+    prompt: Manager
+    tools: [spawn_agent]
+  coder:
+    prompt: Coder
+    isolation: shared
+    tools: [file_write]
+topology:
+  manager: [coder]
+""")
+
+    class TeamLLM(Writer):
+        async def complete(self, messages, **kwargs):
+            if messages[0]["content"].startswith("Manager"):
+                if any(m["role"] == "tool" for m in messages):
+                    return LLMResponse(content="collected", usage=Usage(3, 2), finish_reason="stop")
+                return LLMResponse(tool_calls=[_call("spawn_agent", {"role": "coder", "task": "edit"})],
+                                   usage=Usage(3, 2), finish_reason="tool_calls")
+            if any(m["role"] == "tool" for m in messages):
+                return LLMResponse(content="edited", usage=Usage(3, 2), finish_reason="stop")
+            return LLMResponse(tool_calls=[_call("file_write", {
+                "mode": "create", "path": "child.txt", "content": "delivered",
+            })], usage=Usage(3, 2), finish_reason="tool_calls")
+
+    monkeypatch.setattr(container, "LLMClient", TeamLLM)
+    result = await OpenCollab(tmp_path).team("delegate", config=team, isolation=entry_isolation,
+                                           artifacts=tmp_path / "run", trace=False)
+    assert result.ok and result.output == "collected"
+    assert (tmp_path / "child.txt").exists() is (entry_isolation == "shared")
+    if entry_isolation == "worktree":
+        patch = Path(result.metrics["entry_patch"])
+        assert patch.exists() and "+delivered" in patch.read_text()
+
+
+async def test_unknown_strategy_fails_before_model_use(tmp_path):
+    async def flow(ctx, args):
+        return await ctx.agent("marker", tools=[])
+
+    class NoModel(Writer):
+        async def complete(self, *args, **kwargs):
+            raise AssertionError("model must not be called")
+
+    result = await OpenCollab(tmp_path).workflow(flow, isolation="missing", llm=NoModel(), trace=False)
+    assert result.agent_failures
+    assert result.agent_failures[0]["exception_type"] == "ValueError"
diff --git a/tests/interfaces/test_session_isolation_parent_cleanup.py b/tests/interfaces/test_session_isolation_parent_cleanup.py
new file mode 100644
index 00000000..a1407835
--- /dev/null
+++ b/tests/interfaces/test_session_isolation_parent_cleanup.py
@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import asyncio
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from opencollab.adapters import session_environments
+from opencollab.adapters.llm.types import LLMResponse, Usage
+from opencollab.adapters.session_environments import SessionEnvironmentPool
+from opencollab.adapters.worktree_pool import WorktreePool
+from opencollab.application.workflow_candidates import _CandidateWorkflowSessionFactory
+from opencollab.bootstrap import build_runtime_context, build_scheduler, container
+from opencollab.bootstrap.isolation import prepare_scheduler
+
+
+async def test_candidate_custom_strategy_checks_scoped_cleanup_before_allocating():
+    class LegacyCleanupFactory:
+        allocated = False
+
+        async def acquire_environment(self, **kwargs):
+            self.allocated = True
+            return object()
+
+        async def release_isolated_envs(self):
+            raise AssertionError("another owner's allocations must stay untouched")
+
+    factory = LegacyCleanupFactory()
+    candidate = _CandidateWorkflowSessionFactory(factory, object())
+    with pytest.raises(TypeError, match="scoped release"):
+        await candidate.acquire_environment(isolation="custom")
+    assert not factory.allocated
+
+
+def _repository(path: Path) -> None:
+    for args in (("init", "-q"), ("config", "user.name", "Test"),
+                 ("config", "user.email", "test@example.invalid")):
+        subprocess.run(["git", "-C", str(path), *args], check=True, capture_output=True)
+    (path / "seed.txt").write_text("initial")
+    subprocess.run(["git", "-C", str(path), "add", "."], check=True, capture_output=True)
+    subprocess.run(["git", "-C", str(path), "commit", "-qm", "initial"], check=True, capture_output=True)
+
+
+@pytest.mark.parametrize("tester_isolation", ["shared", "worktree"])
+async def test_nested_team_role_uses_its_spawning_parent_environment(tmp_path, monkeypatch, tester_isolation):
+    _repository(tmp_path)
+    team = tmp_path / "team.yaml"
+    team.write_text(f"""entry: manager
+isolation: shared
+roles:
+  manager:
+    prompt: Manager
+    tools: [spawn_agent]
+  coder:
+    prompt: Coder
+    isolation: worktree
+    tools: [spawn_agent]
+  tester:
+    prompt: Tester
+    isolation: {tester_isolation}
+    tools: []
+topology:
+  manager: [coder]
+  coder: [tester]
+""")
+    release = asyncio.Event()
+
+    class HoldingLLM:
+        def __init__(self, **kwargs):
+            pass
+
+        def context_window(self):
+            return 100_000
+
+        async def complete(self, *args, **kwargs):
+            await release.wait()
+            return LLMResponse(content="done", usage=Usage(1, 1), finish_reason="stop")
+
+        async def close(self):
+            pass
+
+    monkeypatch.setattr(container, "LLMClient", HoldingLLM)
+    context = build_runtime_context(str(tmp_path), {
+        "model": "fixture-model", "provider": "openai", "api_key": "local-fixture",  # pragma: allowlist secret
+        "base_url": "https://controlled.invalid/v1", "budget": 100_000,
+    }, trace=False)
+    scheduler = await prepare_scheduler(
+        context, builder=build_scheduler, use_worktrees=True, interactive=False,
+        auto_save=False, team_config_path=team,
+    )
+    try:
+        coder = await scheduler.spawn(0, "coder", "edit")
+        coder_environment = scheduler.agent_environment(coder)
+        await coder_environment.write_file("coder-only.txt", "candidate")
+        tester = await scheduler.spawn(coder, "tester", "test")
+        tester_environment = scheduler.agent_environment(tester)
+        assert await tester_environment.read_file("coder-only.txt") == "candidate"
+        assert (tester_environment is coder_environment) is (tester_isolation == "shared")
+        assert not (tmp_path / "coder-only.txt").exists()
+    finally:
+        release.set()
+        await asyncio.gather(*tuple(scheduler._tasks.values()), return_exceptions=True)
+        await scheduler.cleanup()
+
+
+async def test_strategy_cleanup_releases_reviewer_before_its_custom_source(tmp_path):
+    _repository(tmp_path)
+
+    def custom(**kwargs):
+        return WorktreePool(use_worktrees=True, **kwargs)
+
+    pool = SessionEnvironmentPool(str(tmp_path), isolation="worktree", providers={"custom": custom})
+    environments = []
+    try:
+        entry = await pool.acquire("manager")
+        environments.append(entry)
+        pool.set_source(entry)
+        coder = await pool.acquire("coder", isolation="custom")
+        environments.append(coder)
+        reviewer = await pool.acquire("reviewer", source=coder, isolation="worktree")
+        environments.append(reviewer)
+        paths = [Path(environment.workspace) for environment in environments]
+        await pool.release()
+        assert all(not path.exists() for path in paths)
+    finally:
+        for environment in reversed(environments):
+            shutil.rmtree(environment.workspace, ignore_errors=True)
+
+
+async def test_failed_shared_spawn_keeps_its_isolated_parent_usable(tmp_path, monkeypatch):
+    _repository(tmp_path)
+    team = tmp_path / "team.yaml"
+    team.write_text("""entry: manager
+isolation: worktree
+roles:
+  manager:
+    prompt: Manager
+    tools: [spawn_agent]
+  coder:
+    prompt: Coder
+    isolation: shared
+    tools: []
+topology:
+  manager: [coder]
+""")
+    context = build_runtime_context(str(tmp_path), {
+        "model": "fixture-model", "provider": "openai", "api_key": "local-fixture",  # pragma: allowlist secret
+        "base_url": "https://controlled.invalid/v1", "budget": 100_000,
+    }, trace=False)
+    scheduler = await prepare_scheduler(
+        context, builder=build_scheduler, use_worktrees=True, interactive=False,
+        auto_save=False, team_config_path=team,
+    )
+
+    def fail_build(**kwargs):
+        raise ValueError("child construction failed")
+
+    monkeypatch.setattr(scheduler._session_factory, "build_spawn_session", fail_build)
+    try:
+        parent = scheduler.agent_environment(0)
+        with pytest.raises(ValueError, match="child construction failed"):
+            await scheduler.spawn(0, "coder", "edit")
+        assert await parent.read_file("seed.txt") == "initial"
+        assert Path(parent.workspace).is_dir()
+    finally:
+        await scheduler.cleanup()
+
+
+def test_unknown_strategy_does_not_retain_an_owned_directory_descriptor(tmp_path, monkeypatch):
+    created = []
+    original = session_environments.LocalEnvironment
+
+    def record(*args, **kwargs):
+        environment = original(*args, **kwargs)
+        created.append(environment)
+        return environment
+
+    monkeypatch.setattr(session_environments, "LocalEnvironment", record)
+    try:
+        with pytest.raises(ValueError, match="needs an environment provider"):
+            SessionEnvironmentPool(str(tmp_path), isolation="missing")
+        assert all(environment._workspace_fd is None for environment in created)
+    finally:
+        for environment in created:
+            environment._close_workspace()
diff --git a/tests/packaging/test_public_readiness.py b/tests/packaging/test_public_readiness.py
index f4e10801..51d995d1 100644
--- a/tests/packaging/test_public_readiness.py
+++ b/tests/packaging/test_public_readiness.py
@@ -35,7 +35,7 @@
 }
 _ACTION_REF = re.compile(r"^\s*(?:-\s+)?uses:\s+([^#\s]+)", re.MULTILINE)
 _FULL_GIT_SHA = re.compile(r"[0-9a-f]{40}")
-_LOCALIZED_README = re.compile(r"README\.[a-z]{2}(?:-[A-Z]{2})?\.md")
+_LOCALIZED_DOCUMENT = re.compile(r".+\.[a-z]{2}(?:-[A-Z]{2})?\.md")
 
 def _repository_files() -> list[Path]:
     completed = subprocess.run(
@@ -72,7 +72,7 @@ def test_public_text_is_english_and_uses_canonical_project_names() -> None:
         relative = path.relative_to(_REPO_ROOT)
         for line_number, line in enumerate(text.splitlines(), start=1):
             contains_chinese = any("\u4e00" <= char <= "\u9fff" for char in line)
-            if contains_chinese and not _LOCALIZED_README.fullmatch(path.name):
+            if contains_chinese and not _LOCALIZED_DOCUMENT.fullmatch(path.name):
                 findings.append(f"{relative}:{line_number}: non-English public text")
             for value in forbidden:
                 if value in line: