feat(runtime): add configurable session isolation strategies

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.

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` 收到当前父任务或审阅任务的真实工作环境，返回已准备好执行工具的环境。提供器负责自己分配的资源，传入的源环境保持借用关系。清理等待在途的环境申请完成，并在失败时保留可恢复材料，由同一个池继续处理清理。工具沿用已有权限和工作区检查。
+
+未知策略会返回错误。模型端点和请求并发采用独立配置。

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,

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,

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()

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()}]"

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

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

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