feat(environments): add group and session Docker isolation

docs/session-isolation.md

@@ -82,3 +82,45 @@ 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.
+
+## Docker strategies
+
+`docker-group` allocates one container for each run using that strategy. Every
+session receives a private worktree in that container, including the Team entry
+agent. `docker-session` allocates a container and private workspace for each
+session. Python orchestration stays in the caller process while tools execute
+inside the selected container. The default image is `python:3.11` and requires
+a working Docker daemon. Custom images need Bash, Git, and an entrypoint that
+accepts the container's `sleep infinity` command.
+
+```python
+from opencollab import OpenCollab
+from opencollab.environments import DockerMount, DockerOptions, docker_isolation_provider
+
+options = DockerOptions(
+    image="python:3.11",
+    cpus=2,
+    memory="2g",
+    pids_limit=256,
+    mounts=(DockerMount("/path/to/read-only-assets", "/assets"),),
+)
+client = OpenCollab(".", environment_providers={
+    "docker-session": docker_isolation_provider(options, scope="session"),
+})
+result = await client.workflow(solve, {"task": "Inspect the code"}, isolation="docker-session")
+```
+
+Container networking defaults to `none`. The `network` option accepts `bridge`
+or a named Docker network when the task needs network access. Additional mounts
+are read-only and reject host sockets. Workspace data is copied into private
+container storage. Existing worktree and submodule capture preserves physical
+file contents while portable Git metadata lets the container operate
+independently from the source checkout. The source may be local or an existing
+Docker environment and remains owned by its caller.
+
+Full patches are captured before owned containers are removed. A failed or
+truncated export retains its workspace and container for recovery and cleanup
+retry. Recovery locations include the full container ID. The runtime's existing
+process cancellation and environment revocation behavior continues to apply.
+The Docker adapter tests can execute these paths against a local image through
+`OPENCOLLAB_DOCKER_TEST_IMAGE`.

docs/session-isolation.zh-CN.md

@@ -55,3 +55,28 @@ Workflow CLI 支持 `--isolation shared` 和 `--isolation worktree`。
 `acquire` 收到当前父任务或审阅任务的真实工作环境，返回已准备好执行工具的环境。提供器负责自己分配的资源，传入的源环境保持借用关系。清理等待在途的环境申请完成，并在失败时保留可恢复材料，由同一个池继续处理清理。工具沿用已有权限和工作区检查。
 
 未知策略会返回错误。模型端点和请求并发采用独立配置。
+
+## Docker 策略
+
+`docker-group` 为每次运行中的该策略分配一个容器，每个 Session 在容器中获得独立工作树，Team 入口 Agent 也使用该容器。`docker-session` 为每个 Session 分配独立容器和工作目录。Python 调度继续在调用方进程中运行，工具在选定容器内执行。默认镜像为 `python:3.11`，需要 Docker daemon 可用。自定义镜像需要 Bash、Git，并且入口程序能够接收 `sleep infinity`。
+
+```python
+from opencollab import OpenCollab
+from opencollab.environments import DockerMount, DockerOptions, docker_isolation_provider
+
+options = DockerOptions(
+    image="python:3.11",
+    cpus=2,
+    memory="2g",
+    pids_limit=256,
+    mounts=(DockerMount("/path/to/read-only-assets", "/assets"),),
+)
+client = OpenCollab(".", environment_providers={
+    "docker-session": docker_isolation_provider(options, scope="session"),
+})
+result = await client.workflow(solve, {"task": "Inspect the code"}, isolation="docker-session")
+```
+
+容器网络默认为 `none`。任务需要网络访问时，`network` 可以指定 `bridge` 或已有 Docker 网络名。附加挂载均为只读，并拒绝暴露宿主 socket。工作区内容复制到容器私有存储，已有工作树和子模块捕获逻辑保留文件的物理内容，可移植 Git 元数据让容器独立使用自己的仓库。源工作区可以来自本机或已有 Docker 环境，其所有权继续由调用方持有。
+
+完整补丁在拥有的容器移除前捕获。导出失败或被截断时，工作区及其容器会保留，用于恢复和清理重试，恢复位置包含完整容器 ID。进程取消与环境撤销沿用已有行为。可以通过 `OPENCOLLAB_DOCKER_TEST_IMAGE` 选择本地镜像，执行 Docker adapter 测试中的真实容器路径。

opencollab/adapters/_docker_snapshot.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import posixpath
+import shlex
+import shutil
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from opencollab.adapters._candidate_physical import _initialized_submodule_paths
+from opencollab.adapters._env_docker import DockerEnvironment
+from opencollab.adapters._env_local import LocalEnvironment
+from opencollab.adapters._git_patch_bytes import _complete, _patch_temporary
+from opencollab.adapters._team_candidate_environment import _TeamCandidateEnvironment
+from opencollab.application.async_timeout import await_owned_operation
+
+
+async def _copy_to_host(environment: Any, source: str, target: Path, *, directory: bool = False) -> None:
+    delegate = environment._delegate if isinstance(environment, _TeamCandidateEnvironment) else environment
+    if getattr(delegate, "local_filesystem", False):
+        if directory:
+            operation = asyncio.to_thread(shutil.copytree, source, target, symlinks=True, dirs_exist_ok=True)
+        else:
+            operation = asyncio.to_thread(shutil.copyfile, source, target)
+        await await_owned_operation(operation, propagate_cancellation=True)
+        return
+    if not isinstance(delegate, DockerEnvironment):
+        raise TypeError("Docker isolation requires a local or Docker source environment")
+    await delegate.setup()
+    source_path = source.rstrip("/") + "/." if directory else source
+    result = await delegate._docker("cp", "--", f"{delegate._container_id}:{source_path}", str(target))
+    if result.returncode or result.stdout_dropped_bytes or result.stderr_dropped_bytes:
+        raise OSError("Docker source workspace copy failed")
+
+
+def _remove_marker(path: Path) -> None:
+    marker = path / ".git"
+    if marker.is_dir() and not marker.is_symlink():
+        shutil.rmtree(marker)
+    elif os.path.lexists(marker):
+        marker.unlink()
+
+
+async def _import_snapshot(source: Any, target: DockerEnvironment) -> str:
+    probe = await source.exec_cmd(f"git -C {shlex.quote(source.workspace)} rev-parse --show-toplevel")
+    _complete(probe, "Docker source repository probe", allowed=(0, 128))
+    repository = probe.returncode == 0
+    root = probe.stdout.removesuffix("\n") if repository else source.workspace
+    prefix = posixpath.relpath(source.workspace, root) if repository else "."
+    modules = await _initialized_submodule_paths(source, root, strict=True) if repository else ()
+    directory = Path(tempfile.mkdtemp(prefix="opencollab-docker-import-"))
+    try:
+        contents = directory / "workspace"
+        contents.mkdir()
+        await _copy_to_host(source, root, contents, directory=True)
+        if repository:
+            for relative in (".", *modules):
+                source_root = posixpath.normpath(posixpath.join(root, relative))
+                local_root = contents.joinpath(*relative.split("/"))
+                if not local_root.is_dir() or local_root.is_symlink():
+                    raise RuntimeError("Docker source repository must be a physical directory")
+                await await_owned_operation(asyncio.to_thread(_remove_marker, local_root), propagate_cancellation=True)
+                object_format = _complete(await source.exec_cmd(
+                    f"git -C {shlex.quote(source_root)} rev-parse --show-object-format",
+                ), "Docker source object format").strip()
+                bundle = directory / "repository.bundle"
+                async with _patch_temporary(source, root, "", prefix=".docker-import-", suffix=".bundle") as remote:
+                    _complete(await source.exec_cmd(
+                        f"git -C {shlex.quote(source_root)} bundle create {shlex.quote(remote)} --tags HEAD",
+                        timeout=900,
+                    ), "Docker source Git bundle")
+                    await _copy_to_host(source, remote, bundle)
+                local = LocalEnvironment(str(local_root))
+                try:
+                    git = (
+                        f"GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null "
+                        f"git -c core.hooksPath=/dev/null -C {shlex.quote(str(local_root))}"
+                    )
+                    _complete(await local.exec_cmd(
+                        f"{git} init --quiet --object-format={shlex.quote(object_format)}", timeout=900,
+                    ), "Docker private repository initialization")
+                    shallow = _complete(await source.exec_cmd(
+                        f"git -C {shlex.quote(source_root)} rev-parse --is-shallow-repository",
+                    ), "Docker source shallow history").strip()
+                    if shallow == "true":
+                        shallow_path = _complete(await source.exec_cmd(
+                            f"git -C {shlex.quote(source_root)} rev-parse --path-format=absolute --git-path shallow",
+                        ), "Docker source shallow boundary").removesuffix("\n")
+                        await _copy_to_host(source, shallow_path, local_root / ".git" / "shallow")
+                    command = (
+                        f"{git} fetch --quiet --tags {shlex.quote(str(bundle))} HEAD && "
+                        f"{git} reset --mixed --quiet FETCH_HEAD"
+                    )
+                    _complete(await local.exec_cmd(command, timeout=900), "Docker private repository metadata")
+                finally:
+                    await local.cleanup()
+                bundle.unlink()
+        else:
+            await await_owned_operation(asyncio.to_thread(_remove_marker, contents), propagate_cancellation=True)
+            local = LocalEnvironment(str(contents))
+            try:
+                git = (
+                    "GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null "
+                    f"git -c core.hooksPath=/dev/null -C {shlex.quote(str(contents))}"
+                )
+                _complete(await local.exec_cmd(
+                    f"{git} init --quiet && {git} add --force --all && "
+                    f"{git} -c user.name='OpenCollab Agent' -c user.email=agent@opencollab.invalid "
+                    "commit --quiet --allow-empty --no-verify -m 'Initial workspace'",
+                    timeout=900,
+                ), "Docker directory snapshot")
+            finally:
+                await local.cleanup()
+        container = await target.setup()
+        _complete(await target.exec_cmd(f"mkdir -p -- {shlex.quote(target.workspace)}"), "Docker workspace directory")
+        copied = await target._docker("cp", "--", str(contents) + "/.", f"{container}:{target.workspace}", timeout=900)
+        if copied.returncode or copied.stdout_dropped_bytes or copied.stderr_dropped_bytes:
+            raise OSError("Docker isolated workspace import failed")
+        identity = _complete(await target.exec_cmd("id -u; id -g"), "Docker workspace user").splitlines()
+        if len(identity) != 2 or any(not item.isdecimal() for item in identity):
+            raise RuntimeError("Docker workspace user identity is incomplete")
+        # Docker's archive transport can retain host IDs. Normalize the private
+        # copy before Git checks ownership, including for images with USER set.
+        ownership = await target._docker(
+            "exec", "--user", "0", "--", container, "chown", "-Rh", ":".join(identity), "--", target.workspace,
+            timeout=900,
+        )
+        if ownership.returncode or ownership.stdout_dropped_bytes or ownership.stderr_dropped_bytes:
+            raise OSError("Docker isolated workspace ownership could not be assigned")
+        return posixpath.normpath(posixpath.join(target.workspace, prefix))
+    finally:
+        await await_owned_operation(
+            asyncio.to_thread(shutil.rmtree, directory), propagate_cancellation=True,
+        )

opencollab/adapters/_env_docker.py

@@ -10,6 +10,7 @@
 import posixpath
 import re
 import shlex
+import stat
 import threading
 import uuid
 from collections.abc import AsyncIterator, Callable
@@ -31,6 +32,7 @@
 from opencollab.adapters._safe_file_common import _normalize_newlines, _split_lines
 from opencollab.application.async_timeout import await_owned_operation
 from opencollab.application.exception_notes import add_exception_note
+from opencollab.domain.docker import DockerOptions
 
 DOCKER_OWNER_LABEL = "opencollab.owner"
 DOCKER_SETUP_TIMEOUT_SECONDS = 120.0
@@ -169,16 +171,21 @@ class DockerEnvironment(Environment):
 
     def __init__(
         self,
-        image: str = "python:3.11-slim",
+        image: str | None = None,
         workspace: str = "/workspace",
         *,
         container_id: str | None = None,
         exec_workdir: str | None = None,
         command_prefix: Callable[[str], str] | str | None = None,
         timeout_returncode: int = -1,
         backing_environment: Environment | None = None,
+        options: DockerOptions | None = None,
+        init_process: bool = False,
     ) -> None:
         super().__init__()
+        self._options = options or DockerOptions()
+        if container_id is not None and options is not None:
+            raise ValueError("Docker options apply only to newly owned containers")
         if container_id is not None and backing_environment is not None:
             raise ValueError("an attached Docker environment cannot own a backing environment")
         if (
@@ -187,7 +194,9 @@ def __init__(
             or timeout_returncode == 0
         ):
             raise ValueError("timeout_returncode must be a non-zero integer")
-        self._image = _validate_image(image)
+        self._image = _validate_image(
+            image if image is not None else options.image if options is not None else "python:3.11-slim"
+        )
         self.workspace = workspace
         self._attached = container_id is not None
         self._attached_reference = (
@@ -203,12 +212,22 @@ def __init__(
         self._owner_token = None if self._attached else uuid.uuid4().hex
         self._exec_workdir = exec_workdir
         self._command_prefix = command_prefix
+        self._init_process = init_process
         self._timeout_returncode = timeout_returncode
         self._backing_environment = backing_environment
+        self._execution_owner: DockerEnvironment | None = None
         self.source_workspace = getattr(backing_environment, "source_workspace", None)
         self.host_workspace = None
         self._temporary_files: set[str] = set()
 
+    @property
+    def revoked(self) -> bool:
+        return self._aborted or (self._execution_owner is not None and self._execution_owner.revoked)
+
+    def _ensure_active(self) -> None:
+        if self.revoked:
+            raise RuntimeError("Execution environment has been aborted.")
+
     @property
     def container_reference(self) -> str | None:
         """Return the attached container, or ``None`` for an owned container.
@@ -291,20 +310,42 @@ async def _setup_locked(self, mount_dir: str | None) -> str:
             host_mount = os.path.realpath(os.path.abspath(mount_dir))
             if not os.path.isdir(host_mount):
                 raise NotADirectoryError(host_mount)
-        self._container_name = f"opencollab-{uuid.uuid4().hex[:16]}"
+        container_name = f"opencollab-{uuid.uuid4().hex[:16]}"
         args = [
             "run",
             "-d",
             "--network",
-            "none",
+            self._options.network,
             "--name",
-            self._container_name,
+            container_name,
             "--label",
             f"{DOCKER_OWNER_LABEL}={self._owner_token}",
         ]
+        if self._init_process:
+            args.append("--init")
+        for flag, value in (
+            ("--cpus", self._options.cpus), ("--memory", self._options.memory),
+            ("--pids-limit", self._options.pids_limit),
+        ):
+            if value is not None:
+                args.extend((flag, str(value)))
+        for mount in self._options.mounts:
+            source = os.path.realpath(os.path.abspath(mount.source))
+            if not os.path.exists(source):
+                raise FileNotFoundError(source)
+            paths = (
+                (os.path.join(directory, name) for directory, _dirs, files in os.walk(source) for name in files)
+                if os.path.isdir(source) else (source,)
+            )
+            if any(stat.S_ISSOCK(os.stat(path).st_mode) for path in paths):
+                raise ValueError("Docker mounts cannot expose host sockets")
+            if "," in source:
+                raise ValueError("Docker mount source cannot contain commas")
+            args.extend(("--mount", f"type=bind,source={source},target={mount.target},readonly"))
         if host_mount is not None:
             args.extend(("-v", f"{host_mount}:{self.workspace}"))
         args.extend(("-w", self.workspace, self._image, "sleep", "infinity"))
+        self._container_name = container_name
         try:
             result = await self._docker(*args, timeout=DOCKER_SETUP_TIMEOUT_SECONDS)
         except BaseException as exc:
@@ -433,6 +474,9 @@ async def _recover_inner(self, token: str) -> bool:
             return True
         if self._attached:
             self.revoke()
+            if self._execution_owner is not None:
+                self._execution_owner.revoke()
+                await self._execution_owner._stop_owned_container()
             return False
         # Losing the command group must not discard the caller's workspace or
         # masquerade as a recoverable tool timeout. Retain it until cleanup().
@@ -890,6 +934,9 @@ async def _abort_resources_locked(self) -> None:
             for result, task in zip(cancelled, active.values())
         )
         if failed_cancellations or pending:
+            if self._execution_owner is not None:
+                self._execution_owner.revoke()
+                await self._execution_owner._stop_owned_container()
             raise ProcessCleanupError(
                 "attached container commands did not quiesce during abort"
             )

opencollab/adapters/_team_candidate_environment.py

@@ -47,11 +47,14 @@ def diff_base(self) -> str:
 
     def retain_changes(self) -> str:
         self._delivery_pending = True
-        return self.workspace
+        return self.recovery_location or self.workspace
 
     @property
     def recovery_location(self) -> str | None:
-        return self.workspace if self._delivery_pending else None
+        if not self._delivery_pending:
+            return None
+        container = getattr(self._delegate, "container_reference", None)
+        return f"{container}:{self.workspace}" if container is not None else self.workspace
 
     def _capture_environment(self, *, allow_delivered: bool = False) -> Any:
         delegate = self._delegate