From 3493cec89ea7f2f7356a63565409e74acce44944 Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 10:41:00 +0800 Subject: [PATCH 1/8] feat(runtime): add owned Docker isolation providers --- opencollab/adapters/_docker_snapshot.py | 115 +++++++ opencollab/adapters/_env_docker.py | 53 +++- .../adapters/_team_candidate_environment.py | 7 +- opencollab/adapters/docker_isolation.py | 185 ++++++++++++ opencollab/domain/docker.py | 57 ++++ tests/runtime/test_docker_isolation.py | 281 ++++++++++++++++++ 6 files changed, 691 insertions(+), 7 deletions(-) create mode 100644 opencollab/adapters/_docker_snapshot.py create mode 100644 opencollab/adapters/docker_isolation.py create mode 100644 opencollab/domain/docker.py create mode 100644 tests/runtime/test_docker_isolation.py diff --git a/opencollab/adapters/_docker_snapshot.py b/opencollab/adapters/_docker_snapshot.py new file mode 100644 index 00000000..370223ab --- /dev/null +++ b/opencollab/adapters/_docker_snapshot.py @@ -0,0 +1,115 @@ +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: + await asyncio.to_thread(shutil.copytree, source, target, symlinks=True, dirs_exist_ok=True) + else: + await asyncio.to_thread(shutil.copyfile, source, target) + 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 () + temporary = await asyncio.to_thread(tempfile.mkdtemp, prefix="opencollab-docker-import-") + directory = Path(temporary) + 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 asyncio.to_thread(_remove_marker, local_root) + 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)} HEAD", + timeout=900, + ), "Docker source Git bundle") + await _copy_to_host(source, remote, bundle) + local = LocalEnvironment(str(local_root)) + try: + command = ( + f"git -C {shlex.quote(str(local_root))} init --quiet && " + f"git -C {shlex.quote(str(local_root))} fetch --quiet {shlex.quote(str(bundle))} HEAD && " + f"git -C {shlex.quote(str(local_root))} reset --mixed --quiet FETCH_HEAD" + ) + _complete(await local.exec_cmd(command, timeout=900), "Docker private repository metadata") + finally: + await local.cleanup() + bundle.unlink() + else: + local = LocalEnvironment(str(contents)) + try: + git = f"git -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, + ) diff --git a/opencollab/adapters/_env_docker.py b/opencollab/adapters/_env_docker.py index 3bcd30ed..2bb29da6 100644 --- a/opencollab/adapters/_env_docker.py +++ b/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,7 +171,7 @@ class DockerEnvironment(Environment): def __init__( self, - image: str = "python:3.11-slim", + image: str | None = None, workspace: str = "/workspace", *, container_id: str | None = None, @@ -177,8 +179,12 @@ def __init__( command_prefix: Callable[[str], str] | str | None = None, timeout_returncode: int = -1, backing_environment: Environment | None = None, + options: DockerOptions | None = None, ) -> 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 +193,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 = ( @@ -205,10 +213,19 @@ def __init__( self._command_prefix = command_prefix 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 +308,40 @@ 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}", ] + 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 +470,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 +930,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" ) diff --git a/opencollab/adapters/_team_candidate_environment.py b/opencollab/adapters/_team_candidate_environment.py index 1ceefb6b..b85bd3cd 100644 --- a/opencollab/adapters/_team_candidate_environment.py +++ b/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 diff --git a/opencollab/adapters/docker_isolation.py b/opencollab/adapters/docker_isolation.py new file mode 100644 index 00000000..936483f6 --- /dev/null +++ b/opencollab/adapters/docker_isolation.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +import asyncio +import posixpath +import uuid +from collections.abc import Sequence +from dataclasses import dataclass + +from opencollab.adapters._docker_snapshot import _import_snapshot +from opencollab.adapters._team_candidate_environment import _TeamCandidateEnvironment +from opencollab.adapters.env import DockerEnvironment, DockerWorkspaceEnvironment, Environment +from opencollab.adapters.worktree_pool import WorktreePool +from opencollab.application.async_timeout import await_owned_operation +from opencollab.application.exception_notes import add_exception_note +from opencollab.domain.docker import DockerOptions +from opencollab.domain.identity import validate_role_identity + + +@dataclass +class _DockerOwner: + container: DockerEnvironment + source_pools: list[WorktreePool] + pool: WorktreePool | None = None + view: DockerWorkspaceEnvironment | None = None + + async def cleanup(self) -> None: + if self.pool is not None: + await self.pool.release() + for source_pool in tuple(self.source_pools): + await source_pool.release() + self.source_pools.remove(source_pool) + if self.view is not None: + await self.view.cleanup() + await self.container.cleanup() + + +class DockerIsolationPool: + """Own execution containers until their worktree delivery and cleanup finish.""" + + def __init__( + self, + workspace: str, + *, + base_environment: Environment | None = None, + scope: str = "group", + options: DockerOptions | None = None, + owned_save_dir: str | None = None, + enable_adoption: bool = False, + owned_output_paths: Sequence[str] = (), + ) -> None: + if scope not in {"group", "session"}: + raise ValueError("Docker isolation scope must be group or session") + self._workspace = workspace + self._base_environment = base_environment + self._scope = scope + self._options = options or DockerOptions() + for mount in self._options.mounts: + for reserved in ("/workspace", "/tmp", "/var/tmp"): + if posixpath.commonpath((mount.target, reserved)) in {mount.target, reserved}: + raise ValueError("Docker isolation mounts must be outside workspace and temporary directories") + self._owned_save_dir = owned_save_dir + self._owned_output_paths = tuple(owned_output_paths) + self._owners: list[_DockerOwner] = [] + self._leases: dict[Environment, _DockerOwner] = {} + self._group: _DockerOwner | None = None + self._lock = asyncio.Lock() + self._delivery = WorktreePool( + workspace, use_worktrees=False, base_environment=base_environment, + enable_adoption=enable_adoption, + ) + + async def _create_owner(self, role: str, source: Environment | None) -> _DockerOwner: + source_pool = WorktreePool( + self._workspace, use_worktrees=True, base_environment=self._base_environment, + enable_adoption=True, owned_save_dir=self._owned_save_dir, + owned_output_paths=self._owned_output_paths, + ) + owner = _DockerOwner(DockerEnvironment(options=self._options), [source_pool]) + self._owners.append(owner) + candidate = await source_pool.acquire(role, source=source) + workspace = await _import_snapshot(candidate, owner.container) + container_id = await owner.container.setup() + owner.view = DockerWorkspaceEnvironment(container_id=container_id, repo_root=workspace) + await owner.view.setup() + owner.pool = WorktreePool( + workspace, use_worktrees=True, base_environment=owner.view, + enable_adoption=True, + ) + await source_pool.release() + owner.source_pools.remove(source_pool) + return owner + + async def _import_source(self, owner: _DockerOwner, role: str, source: Environment) -> Environment: + source_pool = WorktreePool( + source.workspace, use_worktrees=True, base_environment=source, enable_adoption=True, + ) + owner.source_pools.append(source_pool) + candidate = await source_pool.acquire(role) + container_id = await owner.container.setup() + destination = DockerEnvironment( + container_id=container_id, workspace=f"/tmp/opencollab-import-{uuid.uuid4().hex}", exec_workdir="/", + ) + workspace = await _import_snapshot(candidate, destination) + await source_pool.release() + owner.source_pools.remove(source_pool) + return DockerWorkspaceEnvironment(container_id=container_id, repo_root=workspace) + + async def acquire(self, role: str, *, source: Environment | None = None) -> Environment: + role = validate_role_identity(role) + async with self._lock: + owner = self._group if self._scope == "group" else None + created = owner is None + previous_owners = len(self._owners) + try: + if owner is None: + owner = await self._create_owner(role, source) + assert owner.pool is not None + review_source = None + if source is not None and not created: + review_source = source if self._leases.get(source) is owner else await self._import_source( + owner, role, source, + ) + environment = await owner.pool.acquire(role, source=review_source) + delegate = environment._delegate if isinstance(environment, _TeamCandidateEnvironment) else environment + delegate._execution_owner = owner.container + except BaseException as failure: + if created: + pending = owner if owner is not None else ( + self._owners[-1] if len(self._owners) > previous_owners else None + ) + if pending is not None: + try: + await await_owned_operation(pending.cleanup(), propagate_cancellation=False) + except BaseException as cleanup_error: + add_exception_note(failure, f"Docker isolation cleanup retained for retry: {cleanup_error}") + else: + self._owners.remove(pending) + raise + self._leases[environment] = owner + if self._scope == "group": + self._group = owner + return environment + + async def adopt_captured(self, patch: str, physical_paths: Sequence[str]) -> None: + await self._delivery.adopt_captured(patch, physical_paths) + + async def adopt(self, patch: str) -> None: + await self._delivery.adopt(patch) + + async def release_env(self, env: Environment) -> None: + async with self._lock: + owner = self._leases.get(env) + if owner is None: + return + assert owner.pool is not None + await await_owned_operation(owner.pool.release_env(env), propagate_cancellation=True) + if owner is not self._group and not any( + value is owner and environment is not env for environment, value in self._leases.items() + ): + await await_owned_operation(owner.cleanup(), propagate_cancellation=True) + self._owners.remove(owner) + del self._leases[env] + + async def release(self) -> None: + await await_owned_operation(self._release_owned(), propagate_cancellation=True) + + async def _release_owned(self) -> None: + async with self._lock: + errors: list[BaseException] = [] + for owner in reversed(tuple(self._owners)): + try: + await owner.cleanup() + except BaseException as exc: + errors.append(exc) + else: + self._owners.remove(owner) + self._leases = {env: value for env, value in self._leases.items() if value is not owner} + if self._group is owner: + self._group = None + if errors: + raise OSError("Docker isolation cleanup failed; resources retained for retry") from errors[0] + await self._delivery.release() + + async def cleanup(self) -> None: + await self.release() diff --git a/opencollab/domain/docker.py b/opencollab/domain/docker.py new file mode 100644 index 00000000..efc98ae0 --- /dev/null +++ b/opencollab/domain/docker.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import math +import posixpath +import re +from dataclasses import dataclass + + +@dataclass(frozen=True) +class DockerMount: + source: str + target: str + + def __post_init__(self) -> None: + if not isinstance(self.source, str) or not self.source or any(c in self.source for c in "\0\n\r,"): + raise ValueError("Docker mount source must be a non-empty path without control characters or commas") + if ( + not isinstance(self.target, str) or not self.target.startswith("/") + or self.target == "/" or posixpath.normpath(self.target) != self.target + or any(c in self.target for c in "\0\n\r,") + ): + raise ValueError("Docker mount target must be a normalized absolute non-root path") + + +@dataclass(frozen=True) +class DockerOptions: + image: str = "python:3.11" + network: str = "none" + cpus: float | None = None + memory: str | None = None + pids_limit: int | None = None + mounts: tuple[DockerMount, ...] = () + + def __post_init__(self) -> None: + if ( + not isinstance(self.network, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]*", self.network) + or self.network == "host" + ): + raise ValueError("Docker network must be none, bridge, or a named network") + if self.cpus is not None and ( + isinstance(self.cpus, bool) or not isinstance(self.cpus, (int, float)) + or not math.isfinite(self.cpus) or self.cpus <= 0 + ): + raise ValueError("Docker cpus must be a positive finite number") + if self.memory is not None and ( + not isinstance(self.memory, str) or not re.fullmatch(r"[1-9][0-9]*[bkmgBKMG]?", self.memory) + ): + raise ValueError("Docker memory must be a positive byte count with an optional b, k, m, or g suffix") + if self.pids_limit is not None and ( + isinstance(self.pids_limit, bool) or not isinstance(self.pids_limit, int) or self.pids_limit <= 0 + ): + raise ValueError("Docker pids_limit must be a positive integer") + if not isinstance(self.mounts, tuple) or any(not isinstance(mount, DockerMount) for mount in self.mounts): + raise TypeError("Docker mounts must be a tuple of DockerMount values") + targets = [mount.target for mount in self.mounts] + if len(set(targets)) != len(targets): + raise ValueError("Docker mount targets must be distinct") diff --git a/tests/runtime/test_docker_isolation.py b/tests/runtime/test_docker_isolation.py new file mode 100644 index 00000000..0beec342 --- /dev/null +++ b/tests/runtime/test_docker_isolation.py @@ -0,0 +1,281 @@ +from __future__ import annotations + +import asyncio +import json +import os +import shutil +import socket +import subprocess +import tempfile +from pathlib import Path + +import pytest + +from opencollab.adapters._docker_snapshot import _import_snapshot +from opencollab.adapters.docker_isolation import DockerIsolationPool +from opencollab.adapters.env import DockerEnvironment, LocalEnvironment +from opencollab.domain.docker import DockerMount, DockerOptions +from tests.support.docker_env_test_support import CONTAINER_ID, FakeDocker, _patch, _result + + +def _git(workspace: Path, *args: str) -> str: + return subprocess.check_output( + ["git", "-C", str(workspace), "-c", "user.name=Tester", "-c", "user.email=test@example.com", *args], + text=True, stderr=subprocess.PIPE, + ).strip() + + +def _repository(workspace: Path) -> None: + workspace.mkdir() + _git(workspace, "init", "--quiet") + (workspace / "answer.py").write_text("answer = 1\n") + _git(workspace, "add", ".") + _git(workspace, "commit", "--quiet", "-m", "initial") + + +class _LocalImportTarget(LocalEnvironment): + def __init__(self, workspace): + Path(workspace).mkdir() + super().__init__(workspace) + + async def setup(self, mount_dir=None): + return CONTAINER_ID + + async def _docker(self, *args, **kwargs): + if args[0] == "exec": + assert args[1:5] == ("--user", "0", "--", CONTAINER_ID) + assert args[5:7] == ("chown", "-Rh") + return _result() + assert args[:2] == ("cp", "--") + source, destination = args[2:] + assert destination == f"{CONTAINER_ID}:{self.workspace}" + shutil.copytree(source, self.workspace, symlinks=True, dirs_exist_ok=True) + return _result() + + +async def test_import_rebuilds_linked_git_metadata_and_keeps_physical_contents(tmp_path: Path) -> None: + source = tmp_path / "source" + _repository(source) + linked = tmp_path / "linked" + _git(source, "worktree", "add", "--quiet", "--detach", str(linked)) + (linked / "answer.py").write_text("answer = 2\r\n") + (linked / "new.bin").write_bytes(b"\0\xff") + nested = linked / "subdirectory" + nested.mkdir() + (nested / "file.txt").write_text("inside\n") + target = _LocalImportTarget(str(tmp_path / "target")) + target_workspace = await _import_snapshot(LocalEnvironment(str(nested)), target) + assert target_workspace == str(Path(target.workspace) / "subdirectory") + assert (Path(target.workspace) / ".git").is_dir() + assert (Path(target.workspace) / "answer.py").read_bytes() == b"answer = 2\r\n" + assert (Path(target.workspace) / "new.bin").read_bytes() == b"\0\xff" + _git(source, "worktree", "remove", "--force", str(linked)) + assert _git(Path(target.workspace), "show", "HEAD:answer.py") == "answer = 1" + assert "answer.py" in _git(Path(target.workspace), "diff", "--name-only") + assert not list(linked.glob(".docker-import-*")) + + +async def test_import_preserves_initialized_submodule_without_source_git_directory(tmp_path: Path) -> None: + module = tmp_path / "module" + source = tmp_path / "source" + _repository(module) + _repository(source) + _git(source, "-c", "protocol.file.allow=always", "submodule", "add", "--quiet", str(module), "child") + _git(source, "commit", "--quiet", "-am", "submodule") + (source / "child" / "answer.py").write_text("answer = 3\n") + target = _LocalImportTarget(str(tmp_path / "target")) + await _import_snapshot(LocalEnvironment(str(source)), target) + shutil.rmtree(source) + child = Path(target.workspace) / "child" + assert _git(child, "show", "HEAD:answer.py") == "answer = 1" + assert (child / "answer.py").read_text() == "answer = 3\n" + assert "answer.py" in _git(child, "diff", "--name-only") + + +async def test_import_tracks_changes_in_directory_without_git(tmp_path: Path) -> None: + source = tmp_path / "source" + source.mkdir() + (source / ".gitignore").write_text("cache.txt\n") + (source / "cache.txt").write_text("before\n") + target = _LocalImportTarget(str(tmp_path / "target")) + await _import_snapshot(LocalEnvironment(str(source)), target) + (Path(target.workspace) / "cache.txt").write_text("after\n") + assert "-before" in _git(Path(target.workspace), "diff") + assert "+after" in _git(Path(target.workspace), "diff") + assert not (source / ".git").exists() + + +async def test_options_set_owned_resources_and_readonly_mount_without_socket(monkeypatch, tmp_path) -> None: + fake = FakeDocker(lambda command, kwargs: _result(stdout=f"{CONTAINER_ID}\n".encode())) + _patch(monkeypatch, fake) + assets = tmp_path / "assets" + assets.mkdir() + options = DockerOptions( + image="python:3.11", cpus=1.5, memory="256m", pids_limit=64, + mounts=(DockerMount(str(assets), "/assets"),), + ) + env = DockerEnvironment(options=options) + await env.setup() + argv = fake.calls[0][0] + assert argv[argv.index("--cpus") + 1] == "1.5" + assert argv[argv.index("--memory") + 1] == "256m" + assert argv[argv.index("--pids-limit") + 1] == "64" + assert argv[argv.index("--network") + 1] == "none" + assert argv[argv.index("--mount") + 1] == f"type=bind,source={assets},target=/assets,readonly" + assert "docker.sock" not in " ".join(argv) + + +async def test_options_reject_socket_directory_before_container_launch(monkeypatch, tmp_path) -> None: + fake = FakeDocker(lambda command, kwargs: AssertionError(command)) + _patch(monkeypatch, fake) + with tempfile.TemporaryDirectory(prefix="oc-socket-", dir="/tmp") as directory: + with socket.socket(socket.AF_UNIX) as daemon: + daemon.bind(str(Path(directory) / "docker.sock")) + env = DockerEnvironment(options=DockerOptions(mounts=(DockerMount(directory, "/assets"),))) + with pytest.raises(ValueError, match="sockets"): + await env.setup() + assert fake.calls == [] + + +@pytest.mark.parametrize("changes", [ + {"network": "host"}, {"network": "container:other"}, {"cpus": 0}, {"cpus": float("inf")}, + {"memory": "0"}, {"memory": "256m --privileged"}, {"pids_limit": True}, {"pids_limit": -1}, +]) +def test_options_reject_invalid_or_shared_host_settings(changes) -> None: + with pytest.raises(ValueError): + DockerOptions(**changes) + + +async def test_lost_process_group_stops_owned_container_and_revokes_sibling_tools(monkeypatch) -> None: + owner = DockerEnvironment() + first = DockerEnvironment(container_id=CONTAINER_ID) + sibling = DockerEnvironment(container_id=CONTAINER_ID) + first._execution_owner = sibling._execution_owner = owner + stopped = [] + + async def unproven(token): + return False + + async def stop(): + stopped.append(True) + + monkeypatch.setattr(first, "_cancel_inner", unproven) + monkeypatch.setattr(owner, "_stop_owned_container", stop) + assert await first._recover_inner("lost") is False + assert stopped == [True] + assert owner.revoked and sibling.revoked + with pytest.raises(RuntimeError, match="aborted"): + await sibling.exec_cmd("printf forbidden") + + +@pytest.mark.parametrize("scope", ["group", "session"]) +async def test_docker_sessions_execute_and_deliver_before_owned_cleanup(tmp_path: Path, scope: str) -> None: + image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") + if not image: + pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") + source = tmp_path / "source" + _repository(source) + (source / "untracked.txt").write_text("source input\n") + base = LocalEnvironment(str(source)) + pool = DockerIsolationPool( + str(source), base_environment=base, scope=scope, + options=DockerOptions(image=image, cpus=1, memory="256m", pids_limit=64), + ) + first = second = review = None + try: + first = await pool.acquire("first") + second = await pool.acquire("second") + first_id = first._delegate.container_reference + second_id = second._delegate.container_reference + inspected = json.loads(subprocess.check_output(["docker", "inspect", first_id]))[0] + assert inspected["HostConfig"]["NetworkMode"] == "none" + assert inspected["HostConfig"]["NanoCpus"] == 1_000_000_000 + assert inspected["HostConfig"]["Memory"] == 256 * 1024 * 1024 + assert inspected["HostConfig"]["PidsLimit"] == 64 + assert inspected["Mounts"] == [] + assert (first_id == second_id) is (scope == "group") + assert first.workspace != second.workspace or scope == "session" + assert await first.read_file("untracked.txt") == "source input\n" + result = await first.exec_cmd("printf 'answer = 2\\n' > answer.py; printf private > /tmp/private-marker") + assert result.returncode == 0 + assert await second.read_file("answer.py") == "answer = 1\n" + assert (await second.exec_cmd("test -e /tmp/private-marker")).returncode == (0 if scope == "group" else 1) + review = await pool.acquire("reviewer", source=first) + assert await review.read_file("answer.py") == "answer = 2\n" + assert (review._delegate.container_reference == first_id) is (scope == "group") + await review.write_file("answer.py", "answer = 3\n") + assert await first.read_file("answer.py") == "answer = 2\n" + assert "+answer = 3" in await review.get_diff() + patch = await first.get_diff() + assert "+answer = 2" in patch + assert (source / "answer.py").read_text() == "answer = 1\n" + first.retain_changes() + with pytest.raises(OSError, match="retained"): + await pool.release() + assert first.recovery_location is not None + assert "+answer = 2" in await first.get_diff() + finally: + for env in (first, second, review): + if env is not None and env.recovery_location is not None: + await env.get_diff() + await pool.release() + assert not base.revoked + assert not pool._owners + + +async def test_docker_group_imports_external_review_source_and_cancels_owned_command(tmp_path: Path) -> None: + image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") + if not image: + pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") + source = tmp_path / "source" + source.mkdir() + (source / "input.txt").write_text("original\n") + external = tmp_path / "external" + _repository(external) + (external / "answer.py").write_text("answer = 4\n") + assets = tmp_path / "assets" + assets.mkdir() + (assets / "data.txt").write_text("read-only\n") + pool = DockerIsolationPool(str(source), options=DockerOptions( + image=image, mounts=(DockerMount(str(assets), "/assets"),), + )) + running = None + try: + first = await pool.acquire("first") + review = await pool.acquire("review", source=LocalEnvironment(str(external))) + assert first.container_reference == review.container_reference + assert await review.read_file("answer.py") == "answer = 4\n" + assert await first.read_file("input.txt") == "original\n" + assert await first.read_file("/assets/data.txt") == "read-only\n" + assert (await first.exec_cmd("printf changed > /assets/data.txt")).returncode != 0 + assert (assets / "data.txt").read_text() == "read-only\n" + running = asyncio.create_task(first.exec_cmd("printf ready > started; sleep 30; printf late > late")) + for _ in range(100): + if (await first.exec_cmd("test -e started")).returncode == 0: + break + await asyncio.sleep(0.01) + else: + pytest.fail("owned Docker command did not reach its executable probe") + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + control = pool._owners[0].container + container_id = first.container_reference + if control.revoked: + inspected = json.loads(subprocess.check_output(["docker", "inspect", container_id]))[0] + assert inspected["State"]["Running"] is False + artifact = tmp_path / "started.txt" + subprocess.run(["docker", "cp", f"{container_id}:{first.workspace}/started", str(artifact)], check=True) + assert artifact.read_text() == "ready" + subprocess.run(["docker", "start", container_id], check=True, capture_output=True) + else: + assert (await control.exec_cmd(f"test -e {first.workspace}/late")).returncode == 1 + first.retain_changes() + assert "started" in await first.get_diff() + finally: + if running is not None and not running.done(): + running.cancel() + await asyncio.gather(running, return_exceptions=True) + await pool.release() + assert not (source / "started").exists() + assert not (source / ".git").exists() From d46b491cd186ca9d93d5b5c2794ac7d534537c43 Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 10:48:12 +0800 Subject: [PATCH 2/8] feat(sdk): expose Docker session isolation presets --- docs/session-isolation.md | 42 ++++ docs/session-isolation.zh-CN.md | 25 +++ opencollab/adapters/session_environments.py | 8 +- opencollab/bootstrap/docker.py | 19 ++ opencollab/bootstrap/programmatic.py | 4 + opencollab/environments.py | 4 + tests/interfaces/test_sdk_api.py | 3 +- tests/interfaces/test_sdk_docker_isolation.py | 188 ++++++++++++++++++ 8 files changed, 291 insertions(+), 2 deletions(-) create mode 100644 opencollab/bootstrap/docker.py create mode 100644 tests/interfaces/test_sdk_docker_isolation.py diff --git a/docs/session-isolation.md b/docs/session-isolation.md index b39ba667..389165e0 100644 --- a/docs/session-isolation.md +++ b/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`. diff --git a/docs/session-isolation.zh-CN.md b/docs/session-isolation.zh-CN.md index 6ffecfa8..c1644ef3 100644 --- a/docs/session-isolation.zh-CN.md +++ b/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 测试中的真实容器路径。 diff --git a/opencollab/adapters/session_environments.py b/opencollab/adapters/session_environments.py index a65fd0e2..d4389d42 100644 --- a/opencollab/adapters/session_environments.py +++ b/opencollab/adapters/session_environments.py @@ -2,8 +2,10 @@ import asyncio from collections.abc import Callable, Mapping, Sequence +from functools import partial from typing import Any +from opencollab.adapters.docker_isolation import DockerIsolationPool from opencollab.adapters.env import LocalEnvironment from opencollab.adapters.worktree_pool import WorktreePool from opencollab.application.async_timeout import await_owned_operation @@ -26,7 +28,11 @@ def __init__( ) -> None: self._workspace = workspace self._default = isolation_name(isolation) - self._providers = dict(providers or {}) + self._providers = { + "docker-group": partial(DockerIsolationPool, scope="group"), + "docker-session": partial(DockerIsolationPool, scope="session"), + **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) diff --git a/opencollab/bootstrap/docker.py b/opencollab/bootstrap/docker.py new file mode 100644 index 00000000..3bb2c016 --- /dev/null +++ b/opencollab/bootstrap/docker.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from collections.abc import Callable +from functools import partial + +from opencollab.adapters.docker_isolation import DockerIsolationPool +from opencollab.application.ports import WorktreePoolPort +from opencollab.domain.docker import DockerMount, DockerOptions + + +def docker_isolation_provider( + options: DockerOptions | None = None, *, scope: str = "group", +) -> Callable[..., WorktreePoolPort]: + if scope not in {"group", "session"}: + raise ValueError("Docker isolation scope must be group or session") + return partial(DockerIsolationPool, options=options, scope=scope) + + +__all__ = ["DockerMount", "DockerOptions", "docker_isolation_provider"] diff --git a/opencollab/bootstrap/programmatic.py b/opencollab/bootstrap/programmatic.py index 1a236e50..209f82d7 100644 --- a/opencollab/bootstrap/programmatic.py +++ b/opencollab/bootstrap/programmatic.py @@ -62,6 +62,7 @@ run_workflow as _run_workflow, ) from opencollab.domain.agent import Agent +from opencollab.domain.docker import DockerOptions DEFAULT_TEAM_CLEANUP_TIMEOUT_SECONDS = 10.0 @@ -176,11 +177,14 @@ def worktree_environment( def docker_environment( image: str, backing_environment: EnvironmentPort | None = None, + *, + options: DockerOptions | None = None, ) -> EnvironmentPort: """Create an uninitialized image-backed Docker environment.""" return DockerEnvironment( image=image, backing_environment=backing_environment, + options=options, ) def _claim_artifacts(path: Path | None) -> None: diff --git a/opencollab/environments.py b/opencollab/environments.py index 61079ba3..bf16efab 100644 --- a/opencollab/environments.py +++ b/opencollab/environments.py @@ -2,6 +2,7 @@ from opencollab.application.ports import EnvironmentPort as Environment from opencollab.application.ports import WorktreePoolPort as EnvironmentPool +from opencollab.bootstrap.docker import DockerMount, DockerOptions, docker_isolation_provider from opencollab.bootstrap.programmatic import ( attach_container, build_repo_map_via_env, @@ -13,9 +14,12 @@ __all__ = [ "Environment", "EnvironmentPool", + "DockerMount", + "DockerOptions", "attach_container", "build_repo_map_via_env", "docker_environment", + "docker_isolation_provider", "local_environment", "worktree_environment", ] diff --git a/tests/interfaces/test_sdk_api.py b/tests/interfaces/test_sdk_api.py index e8c86f7f..452ad4a0 100644 --- a/tests/interfaces/test_sdk_api.py +++ b/tests/interfaces/test_sdk_api.py @@ -165,12 +165,13 @@ def test_advanced_capabilities_live_in_small_opt_in_modules() -> None: "profile_tool_names", ] assert environments.__all__ == [ - "Environment", "EnvironmentPool", + "Environment", "EnvironmentPool", "DockerMount", "DockerOptions", "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. "build_repo_map_via_env", "docker_environment", + "docker_isolation_provider", "local_environment", "worktree_environment", ] diff --git a/tests/interfaces/test_sdk_docker_isolation.py b/tests/interfaces/test_sdk_docker_isolation.py new file mode 100644 index 00000000..a1627267 --- /dev/null +++ b/tests/interfaces/test_sdk_docker_isolation.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +import asyncio +import json +import os +import re +import subprocess +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest + +from opencollab import OpenCollab +from opencollab.bootstrap import _workflow_runtime_session as wiring +from opencollab.environments import DockerOptions, docker_isolation_provider, local_environment +from opencollab.tools import builtin_tools + + +def _repo(path: Path) -> None: + path.mkdir() + (path / "seed.txt").write_text("input\n") + for args in (("init", "-q"), ("add", "."), ( + "-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "-qm", "initial", + )): + subprocess.run(["git", "-C", str(path), *args], check=True, capture_output=True) + + +@pytest.fixture +def loopback_provider(): + requests = [] + reply = None + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + requests.append(body) + message = reply(body) + payload = json.dumps({ + "id": "chatcmpl_local", "object": "chat.completion", "created": 1, "model": body["model"], + "choices": [{"index": 0, "message": {"role": "assistant", **message}, + "finish_reason": "tool_calls" if "tool_calls" in message else "stop"}], + "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}, + }).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + def install(handler): + nonlocal reply + reply = handler + return f"http://127.0.0.1:{server.server_port}/v1", requests + + try: + yield install + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def _call(name, arguments, call_id): + return {"content": None, "tool_calls": [{ + "id": call_id, "type": "function", "function": {"name": name, "arguments": json.dumps(arguments)}, + }]} + + +def _image(): + image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") + if not image: + pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") + return image + + +@pytest.mark.parametrize("scope", ["group", "session"]) +async def test_workflow_sdk_uses_docker_tools_through_loopback_api(tmp_path, monkeypatch, loopback_provider, scope): + image = _image() + workspace = tmp_path / "workspace" + _repo(workspace) + sessions = [] + build = wiring.build_session + + def remember(**kwargs): + session = build(**kwargs) + sessions.append(session) + return session + + monkeypatch.setattr(wiring, "build_session", remember) + + def respond(body): + if any(message["role"] == "tool" for message in body["messages"]): + return {"content": "finished"} + return _call("bash", {"command": ( + "test ! -e marker.txt && printf private > marker.txt && cat /etc/hostname && pwd && cat seed.txt" + )}, "write") + + endpoint, requests = loopback_provider(respond) + name = f"docker-{scope}" + borrowed = local_environment(workspace) + client = OpenCollab( + workspace, model="offline", provider="openai", base_url=endpoint, + api_key="offline-unused", # pragma: allowlist secret + environment=borrowed, + environment_providers={name: docker_isolation_provider(DockerOptions(image=image), scope=scope)}, + config={"llm_max_retries": 0}, + ) + + async def flow(ctx, inputs): + first = await ctx.agent("first", tools=list(builtin_tools("bash"))) + second = await ctx.agent("second", tools=list(builtin_tools("bash"))) + assert all(await asyncio.gather(*(session.env.read_file("marker.txt") for session in sessions))) + ids = [session.env.container_reference for session in sessions] + assert (ids[0] == ids[1]) is (scope == "group") + assert sessions[0].env.workspace != sessions[1].env.workspace + return first, second + + result = await client.workflow(flow, isolation=name, trace=False) + assert result.ok and result.output == ("finished", "finished") + assert len(requests) == 4 + assert all("input" in str(body["messages"]) for body in requests if any( + message["role"] == "tool" for message in body["messages"] + )) + assert not (workspace / "marker.txt").exists() + assert not borrowed.revoked + + +@pytest.mark.parametrize("scope", ["group", "session"]) +async def test_team_entry_and_child_use_docker_and_save_entry_patch(tmp_path, loopback_provider, scope): + image = _image() + workspace = tmp_path / "workspace" + _repo(workspace) + team = tmp_path / "team.yaml" + team.write_text("""entry: manager +roles: + manager: + prompt: Manager + tools: [bash, spawn_agent] + coder: + prompt: Coder + tools: [bash] +topology: + manager: [coder] +""") + + def respond(body): + messages = body["messages"] + tools = [message for message in messages if message["role"] == "tool"] + manager = messages[0]["content"].startswith("Manager") + if not tools: + command = "printf manager > entry.txt; cat /etc/hostname; pwd" if manager else ( + "cat entry.txt; printf '\\n'; printf child > child.txt; cat /etc/hostname; pwd" + ) + return _call("bash", {"command": command}, "probe") + if manager and len(tools) == 1: + return _call("spawn_agent", {"role": "coder", "task": "edit"}, "spawn") + return {"content": "finished"} + + endpoint, requests = loopback_provider(respond) + name = f"docker-{scope}" + result = await OpenCollab( + workspace, model="offline", provider="openai", base_url=endpoint, + api_key="offline-unused", # pragma: allowlist secret + environment_providers={name: docker_isolation_provider(DockerOptions(image=image), scope=scope)}, + config={"llm_max_retries": 0}, + ).team("delegate", config=team, isolation=name, artifacts=tmp_path / "artifacts", trace=False) + assert result.ok + entry_patch = Path(result.metrics["entry_patch"]).read_text() + assert "+manager" in entry_patch + assert "child.txt" not in entry_patch + assert not (workspace / "entry.txt").exists() + assert not (workspace / "child.txt").exists() + completed = [body for body in requests if any(message["role"] == "tool" for message in body["messages"])] + outputs = {} + for body in completed: + role = "manager" if body["messages"][0]["content"].startswith("Manager") else "coder" + outputs.setdefault(role, next(message["content"] for message in body["messages"] if message["role"] == "tool")) + hostnames = {role: re.search(r"\b[0-9a-f]{12}\b", text).group() for role, text in outputs.items()} + assert (hostnames["manager"] == hostnames["coder"]) is (scope == "group") + assert "manager\n" in outputs["coder"] From a6f7299298c464ff58f83cd1de5b0846f7a13f5f Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 10:52:18 +0800 Subject: [PATCH 3/8] fix(runtime): isolate Docker import Git configuration --- opencollab/adapters/_docker_snapshot.py | 15 +++++++++++---- tests/runtime/test_docker_isolation.py | 15 ++++++++++++++- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/opencollab/adapters/_docker_snapshot.py b/opencollab/adapters/_docker_snapshot.py index 370223ab..d1d30213 100644 --- a/opencollab/adapters/_docker_snapshot.py +++ b/opencollab/adapters/_docker_snapshot.py @@ -71,19 +71,26 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: 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))}" + ) command = ( - f"git -C {shlex.quote(str(local_root))} init --quiet && " - f"git -C {shlex.quote(str(local_root))} fetch --quiet {shlex.quote(str(bundle))} HEAD && " - f"git -C {shlex.quote(str(local_root))} reset --mixed --quiet FETCH_HEAD" + f"{git} init --quiet && {git} fetch --quiet {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 asyncio.to_thread(_remove_marker, contents) local = LocalEnvironment(str(contents)) try: - git = f"git -C {shlex.quote(str(contents))}" + 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 " diff --git a/tests/runtime/test_docker_isolation.py b/tests/runtime/test_docker_isolation.py index 0beec342..7a795ccf 100644 --- a/tests/runtime/test_docker_isolation.py +++ b/tests/runtime/test_docker_isolation.py @@ -92,13 +92,26 @@ async def test_import_preserves_initialized_submodule_without_source_git_directo assert "answer.py" in _git(child, "diff", "--name-only") -async def test_import_tracks_changes_in_directory_without_git(tmp_path: Path) -> None: +async def test_import_tracks_changes_in_directory_without_git(tmp_path: Path, monkeypatch) -> None: source = tmp_path / "source" source.mkdir() (source / ".gitignore").write_text("cache.txt\n") (source / "cache.txt").write_text("before\n") + hooks = tmp_path / "hooks" + hooks.mkdir() + marker = tmp_path / "hook-ran" + post_commit = hooks / "post-commit" + post_commit.write_text(f"#!/bin/sh\ntouch '{marker}'\n") + post_commit.chmod(0o700) + configuration = tmp_path / "global-config" + configuration.write_text(f"[core]\n\thooksPath = {hooks}\n") + monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(configuration)) + _repository(tmp_path / "control") + assert marker.exists() + marker.unlink() target = _LocalImportTarget(str(tmp_path / "target")) await _import_snapshot(LocalEnvironment(str(source)), target) + assert not marker.exists() (Path(target.workspace) / "cache.txt").write_text("after\n") assert "-before" in _git(Path(target.workspace), "diff") assert "+after" in _git(Path(target.workspace), "diff") From 5e4320b075ce986d62b430733c6e8747a9bc35af Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 11:07:20 +0800 Subject: [PATCH 4/8] fix(runtime): preserve Docker import ownership and Git format --- opencollab/adapters/_docker_snapshot.py | 18 ++-- tests/runtime/test_docker_isolation.py | 104 ++++++++++++++++++++++-- 2 files changed, 106 insertions(+), 16 deletions(-) diff --git a/opencollab/adapters/_docker_snapshot.py b/opencollab/adapters/_docker_snapshot.py index d1d30213..368ac040 100644 --- a/opencollab/adapters/_docker_snapshot.py +++ b/opencollab/adapters/_docker_snapshot.py @@ -21,9 +21,10 @@ async def _copy_to_host(environment: Any, source: str, target: Path, *, director delegate = environment._delegate if isinstance(environment, _TeamCandidateEnvironment) else environment if getattr(delegate, "local_filesystem", False): if directory: - await asyncio.to_thread(shutil.copytree, source, target, symlinks=True, dirs_exist_ok=True) + operation = asyncio.to_thread(shutil.copytree, source, target, symlinks=True, dirs_exist_ok=True) else: - await asyncio.to_thread(shutil.copyfile, source, target) + 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") @@ -49,8 +50,7 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: 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 () - temporary = await asyncio.to_thread(tempfile.mkdtemp, prefix="opencollab-docker-import-") - directory = Path(temporary) + directory = Path(tempfile.mkdtemp(prefix="opencollab-docker-import-")) try: contents = directory / "workspace" contents.mkdir() @@ -61,7 +61,10 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: 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 asyncio.to_thread(_remove_marker, local_root) + 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( @@ -76,7 +79,8 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: f"git -c core.hooksPath=/dev/null -C {shlex.quote(str(local_root))}" ) command = ( - f"{git} init --quiet && {git} fetch --quiet {shlex.quote(str(bundle))} HEAD && " + f"{git} init --quiet --object-format={shlex.quote(object_format)} && " + f"{git} fetch --quiet {shlex.quote(str(bundle))} HEAD && " f"{git} reset --mixed --quiet FETCH_HEAD" ) _complete(await local.exec_cmd(command, timeout=900), "Docker private repository metadata") @@ -84,7 +88,7 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: await local.cleanup() bundle.unlink() else: - await asyncio.to_thread(_remove_marker, contents) + await await_owned_operation(asyncio.to_thread(_remove_marker, contents), propagate_cancellation=True) local = LocalEnvironment(str(contents)) try: git = ( diff --git a/tests/runtime/test_docker_isolation.py b/tests/runtime/test_docker_isolation.py index 7a795ccf..e77139d8 100644 --- a/tests/runtime/test_docker_isolation.py +++ b/tests/runtime/test_docker_isolation.py @@ -7,10 +7,12 @@ import socket import subprocess import tempfile +import threading from pathlib import Path import pytest +from opencollab.adapters import _docker_snapshot as snapshot_module from opencollab.adapters._docker_snapshot import _import_snapshot from opencollab.adapters.docker_isolation import DockerIsolationPool from opencollab.adapters.env import DockerEnvironment, LocalEnvironment @@ -25,9 +27,9 @@ def _git(workspace: Path, *args: str) -> str: ).strip() -def _repository(workspace: Path) -> None: +def _repository(workspace: Path, *, object_format: str = "sha1") -> None: workspace.mkdir() - _git(workspace, "init", "--quiet") + _git(workspace, "init", "--quiet", f"--object-format={object_format}") (workspace / "answer.py").write_text("answer = 1\n") _git(workspace, "add", ".") _git(workspace, "commit", "--quiet", "-m", "initial") @@ -53,9 +55,12 @@ async def _docker(self, *args, **kwargs): return _result() -async def test_import_rebuilds_linked_git_metadata_and_keeps_physical_contents(tmp_path: Path) -> None: +@pytest.mark.parametrize("object_format", ["sha1", "sha256"]) +async def test_import_rebuilds_linked_git_metadata_and_keeps_physical_contents( + tmp_path: Path, monkeypatch, object_format: str, +) -> None: source = tmp_path / "source" - _repository(source) + _repository(source, object_format=object_format) linked = tmp_path / "linked" _git(source, "worktree", "add", "--quiet", "--detach", str(linked)) (linked / "answer.py").write_text("answer = 2\r\n") @@ -64,22 +69,27 @@ async def test_import_rebuilds_linked_git_metadata_and_keeps_physical_contents(t nested.mkdir() (nested / "file.txt").write_text("inside\n") target = _LocalImportTarget(str(tmp_path / "target")) + monkeypatch.setenv("GIT_DEFAULT_HASH", "sha256" if object_format == "sha1" else "sha1") target_workspace = await _import_snapshot(LocalEnvironment(str(nested)), target) assert target_workspace == str(Path(target.workspace) / "subdirectory") assert (Path(target.workspace) / ".git").is_dir() assert (Path(target.workspace) / "answer.py").read_bytes() == b"answer = 2\r\n" assert (Path(target.workspace) / "new.bin").read_bytes() == b"\0\xff" + assert _git(Path(target.workspace), "rev-parse", "--show-object-format") == object_format _git(source, "worktree", "remove", "--force", str(linked)) assert _git(Path(target.workspace), "show", "HEAD:answer.py") == "answer = 1" assert "answer.py" in _git(Path(target.workspace), "diff", "--name-only") assert not list(linked.glob(".docker-import-*")) -async def test_import_preserves_initialized_submodule_without_source_git_directory(tmp_path: Path) -> None: +@pytest.mark.parametrize("object_format", ["sha1", "sha256"]) +async def test_import_preserves_initialized_submodule_without_source_git_directory( + tmp_path: Path, object_format: str, +) -> None: module = tmp_path / "module" source = tmp_path / "source" - _repository(module) - _repository(source) + _repository(module, object_format=object_format) + _repository(source, object_format=object_format) _git(source, "-c", "protocol.file.allow=always", "submodule", "add", "--quiet", str(module), "child") _git(source, "commit", "--quiet", "-am", "submodule") (source / "child" / "answer.py").write_text("answer = 3\n") @@ -92,6 +102,79 @@ async def test_import_preserves_initialized_submodule_without_source_git_directo assert "answer.py" in _git(child, "diff", "--name-only") +@pytest.mark.parametrize("stage", ["workspace-copy", "bundle-copy", "metadata-removal"]) +async def test_cancelled_import_waits_for_host_io_before_removing_its_source(tmp_path, monkeypatch, stage): + source = tmp_path / "source" + _repository(source) + pool = DockerIsolationPool(str(source)) + entered, resume, finished = threading.Event(), threading.Event(), threading.Event() + temporary = None + + def hold(operation, directory, *args, **kwargs): + nonlocal temporary + temporary = directory + entered.set() + try: + assert resume.wait(5) + return operation(*args, **kwargs) + finally: + finished.set() + + if stage == "workspace-copy": + original = shutil._copytree + + def copytree(entries, src, dst, *args, **kwargs): + if "opencollab-docker-import-" in str(dst) and not entered.is_set(): + return hold(original, Path(dst).parent, entries, src, dst, *args, **kwargs) + return original(entries, src, dst, *args, **kwargs) + + monkeypatch.setattr(shutil, "_copytree", copytree) + elif stage == "bundle-copy": + original = shutil.copyfile + + def copyfile(src, dst, *args, **kwargs): + if Path(dst).name == "repository.bundle": + return hold(original, Path(dst).parent, src, dst, *args, **kwargs) + return original(src, dst, *args, **kwargs) + + monkeypatch.setattr(shutil, "copyfile", copyfile) + else: + original = snapshot_module._remove_marker + + def remove_marker(path): + return hold(original, path.parent, path) + + monkeypatch.setattr(snapshot_module, "_remove_marker", remove_marker) + + task = asyncio.create_task(pool.acquire("cancelled")) + try: + assert await asyncio.to_thread(entered.wait, 2) + candidate = Path(pool._owners[0].source_pools[0]._envs[0].workspace) + task.cancel("first cancellation") + await asyncio.sleep(0) + task.cancel("later cancellation") + await asyncio.sleep(0.05) + assert not task.done() + assert candidate.exists() + assert temporary.exists() + resume.set() + with pytest.raises(asyncio.CancelledError) as failure: + await task + assert failure.value.args == ("first cancellation",) + assert finished.is_set() + assert not candidate.exists() + assert not temporary.exists() + assert not pool._owners + finally: + resume.set() + await asyncio.gather(task, return_exceptions=True) + if entered.is_set(): + assert await asyncio.to_thread(finished.wait, 2) + await pool.release() + if temporary is not None: + shutil.rmtree(temporary, ignore_errors=True) + + async def test_import_tracks_changes_in_directory_without_git(tmp_path: Path, monkeypatch) -> None: source = tmp_path / "source" source.mkdir() @@ -182,12 +265,15 @@ async def stop(): @pytest.mark.parametrize("scope", ["group", "session"]) -async def test_docker_sessions_execute_and_deliver_before_owned_cleanup(tmp_path: Path, scope: str) -> None: +@pytest.mark.parametrize("object_format", ["sha1", "sha256"]) +async def test_docker_sessions_execute_and_deliver_before_owned_cleanup( + tmp_path: Path, scope: str, object_format: str, +) -> None: image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") if not image: pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") source = tmp_path / "source" - _repository(source) + _repository(source, object_format=object_format) (source / "untracked.txt").write_text("source input\n") base = LocalEnvironment(str(source)) pool = DockerIsolationPool( From 1e71d6540c9a0c1eda3604cd134b688fdc692176 Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 11:21:56 +0800 Subject: [PATCH 5/8] fix(runtime): reap Docker isolation command descendants --- opencollab/adapters/_env_docker.py | 4 ++ opencollab/adapters/docker_isolation.py | 2 +- tests/runtime/test_docker_isolation.py | 84 +++++++++++++++++++++---- 3 files changed, 77 insertions(+), 13 deletions(-) diff --git a/opencollab/adapters/_env_docker.py b/opencollab/adapters/_env_docker.py index 2bb29da6..ffa57a54 100644 --- a/opencollab/adapters/_env_docker.py +++ b/opencollab/adapters/_env_docker.py @@ -180,6 +180,7 @@ def __init__( 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() @@ -211,6 +212,7 @@ 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 @@ -319,6 +321,8 @@ async def _setup_locked(self, mount_dir: str | None) -> str: "--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), diff --git a/opencollab/adapters/docker_isolation.py b/opencollab/adapters/docker_isolation.py index 936483f6..abb2d7a8 100644 --- a/opencollab/adapters/docker_isolation.py +++ b/opencollab/adapters/docker_isolation.py @@ -75,7 +75,7 @@ async def _create_owner(self, role: str, source: Environment | None) -> _DockerO enable_adoption=True, owned_save_dir=self._owned_save_dir, owned_output_paths=self._owned_output_paths, ) - owner = _DockerOwner(DockerEnvironment(options=self._options), [source_pool]) + owner = _DockerOwner(DockerEnvironment(options=self._options, init_process=True), [source_pool]) self._owners.append(owner) candidate = await source_pool.acquire(role, source=source) workspace = await _import_snapshot(candidate, owner.container) diff --git a/tests/runtime/test_docker_isolation.py b/tests/runtime/test_docker_isolation.py index e77139d8..35d77231 100644 --- a/tests/runtime/test_docker_isolation.py +++ b/tests/runtime/test_docker_isolation.py @@ -219,6 +219,15 @@ async def test_options_set_owned_resources_and_readonly_mount_without_socket(mon assert argv[argv.index("--network") + 1] == "none" assert argv[argv.index("--mount") + 1] == f"type=bind,source={assets},target=/assets,readonly" assert "docker.sock" not in " ".join(argv) + assert "--init" not in argv + + +async def test_owned_docker_init_process_is_opt_in(monkeypatch) -> None: + fake = FakeDocker(lambda command, kwargs: _result(stdout=f"{CONTAINER_ID}\n".encode())) + _patch(monkeypatch, fake) + env = DockerEnvironment(init_process=True) + await env.setup() + assert "--init" in fake.calls[0][0] async def test_options_reject_socket_directory_before_container_launch(monkeypatch, tmp_path) -> None: @@ -287,6 +296,7 @@ async def test_docker_sessions_execute_and_deliver_before_owned_cleanup( first_id = first._delegate.container_reference second_id = second._delegate.container_reference inspected = json.loads(subprocess.check_output(["docker", "inspect", first_id]))[0] + assert inspected["HostConfig"]["Init"] is True assert inspected["HostConfig"]["NetworkMode"] == "none" assert inspected["HostConfig"]["NanoCpus"] == 1_000_000_000 assert inspected["HostConfig"]["Memory"] == 256 * 1024 * 1024 @@ -322,7 +332,8 @@ async def test_docker_sessions_execute_and_deliver_before_owned_cleanup( assert not pool._owners -async def test_docker_group_imports_external_review_source_and_cancels_owned_command(tmp_path: Path) -> None: +@pytest.mark.parametrize("scope", ["group", "session"]) +async def test_docker_imports_external_review_source_and_cancels_owned_command(tmp_path: Path, scope: str) -> None: image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") if not image: pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") @@ -335,14 +346,16 @@ async def test_docker_group_imports_external_review_source_and_cancels_owned_com assets = tmp_path / "assets" assets.mkdir() (assets / "data.txt").write_text("read-only\n") - pool = DockerIsolationPool(str(source), options=DockerOptions( + pool = DockerIsolationPool(str(source), scope=scope, options=DockerOptions( image=image, mounts=(DockerMount(str(assets), "/assets"),), )) running = None + container_ids = set() try: first = await pool.acquire("first") review = await pool.acquire("review", source=LocalEnvironment(str(external))) - assert first.container_reference == review.container_reference + container_ids.update((first.container_reference, review.container_reference)) + assert (first.container_reference == review.container_reference) is (scope == "group") assert await review.read_file("answer.py") == "answer = 4\n" assert await first.read_file("input.txt") == "original\n" assert await first.read_file("/assets/data.txt") == "read-only\n" @@ -360,15 +373,10 @@ async def test_docker_group_imports_external_review_source_and_cancels_owned_com await running control = pool._owners[0].container container_id = first.container_reference - if control.revoked: - inspected = json.loads(subprocess.check_output(["docker", "inspect", container_id]))[0] - assert inspected["State"]["Running"] is False - artifact = tmp_path / "started.txt" - subprocess.run(["docker", "cp", f"{container_id}:{first.workspace}/started", str(artifact)], check=True) - assert artifact.read_text() == "ready" - subprocess.run(["docker", "start", container_id], check=True, capture_output=True) - else: - assert (await control.exec_cmd(f"test -e {first.workspace}/late")).returncode == 1 + assert not control.revoked + inspected = json.loads(subprocess.check_output(["docker", "inspect", container_id]))[0] + assert inspected["State"]["Running"] is True + assert (await control.exec_cmd(f"test -e {first.workspace}/late")).returncode == 1 first.retain_changes() assert "started" in await first.get_diff() finally: @@ -378,3 +386,55 @@ async def test_docker_group_imports_external_review_source_and_cancels_owned_com await pool.release() assert not (source / "started").exists() assert not (source / ".git").exists() + for container_id in container_ids: + assert subprocess.run(["docker", "inspect", container_id], capture_output=True).returncode != 0 + + +async def test_unproven_docker_cancellation_retains_unexported_candidate(tmp_path: Path) -> None: + image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") + if not image: + pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") + source = tmp_path / "source" + _repository(source) + pool = DockerIsolationPool(str(source), scope="session", options=DockerOptions(image=image)) + running = None + first = None + try: + first = await pool.acquire("first") + container_id = first.container_reference + running = asyncio.create_task(first.exec_cmd( + "rm -f /tmp/.opencollab-exec-*.pid; printf 'answer = 5\\n' > answer.py; " + "printf ready > started; sleep 300; printf late > late", + timeout=600, + )) + for _ in range(100): + if (await first.exec_cmd("test -e started")).returncode == 0: + break + await asyncio.sleep(0.01) + else: + pytest.fail("owned Docker command did not reach its executable probe") + running.cancel() + with pytest.raises(asyncio.CancelledError): + await running + assert first.revoked + inspected = json.loads(subprocess.check_output(["docker", "inspect", container_id]))[0] + assert inspected["HostConfig"]["Init"] is True + assert inspected["State"]["Running"] is False + first.retain_changes() + with pytest.raises(OSError, match="retained"): + await pool.release() + assert pool._owners + assert first.recovery_location == f"{container_id}:{first.workspace}" + artifact = tmp_path / "answer.py" + subprocess.run(["docker", "cp", f"{container_id}:{first.workspace}/answer.py", str(artifact)], check=True) + assert artifact.read_text() == "answer = 5\n" + finally: + if running is not None and not running.done(): + running.cancel() + await asyncio.gather(running, return_exceptions=True) + if first is not None and first.recovery_location is not None: + subprocess.run(["docker", "start", first.container_reference], check=True, capture_output=True) + assert "+answer = 5" in await first.get_diff() + await pool.release() + assert (source / "answer.py").read_text() == "answer = 1\n" + assert subprocess.run(["docker", "inspect", container_id], capture_output=True).returncode != 0 From 30a49768d349272679bd59f035b46540cd64b21d Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 11:53:19 +0800 Subject: [PATCH 6/8] test(runtime): honor Python 3.10 cancellation metadata --- tests/runtime/test_docker_isolation.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/runtime/test_docker_isolation.py b/tests/runtime/test_docker_isolation.py index 35d77231..b8b9578c 100644 --- a/tests/runtime/test_docker_isolation.py +++ b/tests/runtime/test_docker_isolation.py @@ -17,6 +17,7 @@ from opencollab.adapters.docker_isolation import DockerIsolationPool from opencollab.adapters.env import DockerEnvironment, LocalEnvironment from opencollab.domain.docker import DockerMount, DockerOptions +from tests.support.asyncio_test_support import assert_cancel_reason from tests.support.docker_env_test_support import CONTAINER_ID, FakeDocker, _patch, _result @@ -160,7 +161,7 @@ def remove_marker(path): resume.set() with pytest.raises(asyncio.CancelledError) as failure: await task - assert failure.value.args == ("first cancellation",) + assert_cancel_reason(failure.value, "first cancellation") assert finished.is_set() assert not candidate.exists() assert not temporary.exists() From 82e83e23902f3b87da47988837de229f1ad5b00f Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 13:16:19 +0800 Subject: [PATCH 7/8] fix(runtime): preserve release tags in Docker imports Include source tag references in workspace bundles and restore them alongside HEAD so container builds retain Git-based version discovery. Preserve both lightweight and annotated tags with their release history. Verify native group and session builds, historical release reads, tag object types, and linked source imports in SHA-1 and SHA-256 repositories. --- opencollab/adapters/_docker_snapshot.py | 4 +-- tests/runtime/test_docker_isolation.py | 33 +++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/opencollab/adapters/_docker_snapshot.py b/opencollab/adapters/_docker_snapshot.py index 368ac040..1e231698 100644 --- a/opencollab/adapters/_docker_snapshot.py +++ b/opencollab/adapters/_docker_snapshot.py @@ -68,7 +68,7 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: 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)} HEAD", + 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) @@ -80,7 +80,7 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: ) command = ( f"{git} init --quiet --object-format={shlex.quote(object_format)} && " - f"{git} fetch --quiet {shlex.quote(str(bundle))} HEAD && " + 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") diff --git a/tests/runtime/test_docker_isolation.py b/tests/runtime/test_docker_isolation.py index b8b9578c..20cd05c5 100644 --- a/tests/runtime/test_docker_isolation.py +++ b/tests/runtime/test_docker_isolation.py @@ -62,6 +62,8 @@ async def test_import_rebuilds_linked_git_metadata_and_keeps_physical_contents( ) -> None: source = tmp_path / "source" _repository(source, object_format=object_format) + _git(source, "tag", "v1.0.0") + _git(source, "tag", "-a", "v1.1.0", "-m", "release") linked = tmp_path / "linked" _git(source, "worktree", "add", "--quiet", "--detach", str(linked)) (linked / "answer.py").write_text("answer = 2\r\n") @@ -77,6 +79,9 @@ async def test_import_rebuilds_linked_git_metadata_and_keeps_physical_contents( assert (Path(target.workspace) / "answer.py").read_bytes() == b"answer = 2\r\n" assert (Path(target.workspace) / "new.bin").read_bytes() == b"\0\xff" assert _git(Path(target.workspace), "rev-parse", "--show-object-format") == object_format + assert _git(Path(target.workspace), "describe", "--tags", "--dirty") == "v1.1.0-dirty" + assert _git(Path(target.workspace), "cat-file", "-t", "refs/tags/v1.0.0") == "commit" + assert _git(Path(target.workspace), "cat-file", "-t", "refs/tags/v1.1.0") == "tag" _git(source, "worktree", "remove", "--force", str(linked)) assert _git(Path(target.workspace), "show", "HEAD:answer.py") == "answer = 1" assert "answer.py" in _git(Path(target.workspace), "diff", "--name-only") @@ -274,6 +279,34 @@ async def stop(): await sibling.exec_cmd("printf forbidden") +@pytest.mark.parametrize("scope", ["group", "session"]) +async def test_docker_sessions_preserve_tagged_build_versions(tmp_path: Path, scope: str) -> None: + image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") + if not image: + pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") + source = tmp_path / "source" + _repository(source) + _git(source, "tag", "v1.0.0") + (source / "answer.py").write_text("answer = 2\n") + _git(source, "commit", "--quiet", "-am", "second") + _git(source, "tag", "-a", "v1.1.0", "-m", "release") + (source / "answer.py").write_text("answer = 3\n") + _git(source, "commit", "--quiet", "-am", "third") + version = _git(source, "describe", "--tags", "--dirty") + pool = DockerIsolationPool(str(source), scope=scope, options=DockerOptions(image=image)) + try: + environment = await pool.acquire("builder") + result = await environment.exec_cmd( + "git describe --tags --dirty && git describe --dirty && " + "git show v1.0.0:answer.py && git show v1.1.0:answer.py && " + "git cat-file -t refs/tags/v1.0.0 && git cat-file -t refs/tags/v1.1.0" + ) + assert result.returncode == 0, result.stderr + assert result.stdout.splitlines() == [version, version, "answer = 1", "answer = 2", "commit", "tag"] + finally: + await pool.release() + + @pytest.mark.parametrize("scope", ["group", "session"]) @pytest.mark.parametrize("object_format", ["sha1", "sha256"]) async def test_docker_sessions_execute_and_deliver_before_owned_cleanup( From ae44bce945bffb38856e4b240460ce737540849b Mon Sep 17 00:00:00 2001 From: Kai Date: Sat, 24 Oct 2026 13:38:55 +0800 Subject: [PATCH 8/8] fix(runtime): preserve shallow history in Docker imports Copy each source repository's existing shallow boundary into its private Git metadata before fetching the workspace bundle. Resolve the boundary through Git so linked worktrees and initialized submodules retain their own history. Cover shallow SHA-1 and SHA-256 repositories, linked worktrees, submodules, and native group/session review handoffs while preserving source boundaries. --- opencollab/adapters/_docker_snapshot.py | 12 ++++- tests/runtime/test_docker_isolation.py | 67 +++++++++++++++++++++++++ 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/opencollab/adapters/_docker_snapshot.py b/opencollab/adapters/_docker_snapshot.py index 1e231698..e22f07f8 100644 --- a/opencollab/adapters/_docker_snapshot.py +++ b/opencollab/adapters/_docker_snapshot.py @@ -78,8 +78,18 @@ async def _import_snapshot(source: Any, target: DockerEnvironment) -> str: 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} init --quiet --object-format={shlex.quote(object_format)} && " f"{git} fetch --quiet --tags {shlex.quote(str(bundle))} HEAD && " f"{git} reset --mixed --quiet FETCH_HEAD" ) diff --git a/tests/runtime/test_docker_isolation.py b/tests/runtime/test_docker_isolation.py index 20cd05c5..b83e261b 100644 --- a/tests/runtime/test_docker_isolation.py +++ b/tests/runtime/test_docker_isolation.py @@ -36,6 +36,15 @@ def _repository(workspace: Path, *, object_format: str = "sha1") -> None: _git(workspace, "commit", "--quiet", "-m", "initial") +def _shallow_repository(workspace: Path, *, object_format: str = "sha1") -> None: + origin = workspace.with_name(workspace.name + "-origin") + _repository(origin, object_format=object_format) + (origin / "answer.py").write_text("answer = 2\n") + _git(origin, "commit", "--quiet", "-am", "second") + _git(origin, "tag", "v1.2") + _git(origin, "clone", "--quiet", "--depth=1", origin.as_uri(), str(workspace)) + + class _LocalImportTarget(LocalEnvironment): def __init__(self, workspace): Path(workspace).mkdir() @@ -88,6 +97,39 @@ async def test_import_rebuilds_linked_git_metadata_and_keeps_physical_contents( assert not list(linked.glob(".docker-import-*")) +@pytest.mark.parametrize("object_format", ["sha1", "sha256"]) +@pytest.mark.parametrize("layout", ["repository", "linked", "submodule"]) +async def test_import_preserves_shallow_history(tmp_path: Path, object_format: str, layout: str) -> None: + source = tmp_path / "source" + _shallow_repository(source, object_format=object_format) + workspace = source + relative = "." + if layout == "linked": + workspace = tmp_path / "linked" + _git(source, "worktree", "add", "--quiet", "--detach", str(workspace)) + elif layout == "submodule": + workspace = tmp_path / "parent" + _repository(workspace, object_format=object_format) + _git(workspace, "-c", "protocol.file.allow=always", "submodule", "add", "--quiet", "--depth=1", + source.as_uri(), "child") + _git(workspace, "commit", "--quiet", "-am", "submodule") + relative = "child" + repository = workspace / relative + shallow = Path(_git(repository, "rev-parse", "--path-format=absolute", "--git-path", "shallow")) + boundary = shallow.read_bytes() + target = _LocalImportTarget(str(tmp_path / "target")) + await _import_snapshot(LocalEnvironment(str(workspace)), target) + imported = Path(target.workspace) / relative + assert _git(imported, "rev-parse", "--show-object-format") == object_format + assert _git(imported, "rev-parse", "--is-shallow-repository") == "true" + assert _git(imported, "rev-list", "--count", "HEAD") == "1" + assert _git(imported, "describe", "--tags") == "v1.2" + assert _git(imported, "show", "HEAD:answer.py") == "answer = 2" + assert shallow.read_bytes() == boundary + shutil.rmtree(workspace) + assert _git(imported, "rev-list", "--count", "HEAD") == "1" + + @pytest.mark.parametrize("object_format", ["sha1", "sha256"]) async def test_import_preserves_initialized_submodule_without_source_git_directory( tmp_path: Path, object_format: str, @@ -307,6 +349,31 @@ async def test_docker_sessions_preserve_tagged_build_versions(tmp_path: Path, sc await pool.release() +@pytest.mark.parametrize("scope", ["group", "session"]) +@pytest.mark.parametrize("object_format", ["sha1", "sha256"]) +async def test_docker_sessions_preserve_shallow_source_history(tmp_path: Path, scope: str, object_format: str) -> None: + image = os.environ.get("OPENCOLLAB_DOCKER_TEST_IMAGE") + if not image: + pytest.skip("OPENCOLLAB_DOCKER_TEST_IMAGE selects a local image with bash and git") + source = tmp_path / "source" + _shallow_repository(source, object_format=object_format) + pool = DockerIsolationPool(str(source), scope=scope, options=DockerOptions(image=image)) + try: + first = await pool.acquire("builder") + reviewer = await pool.acquire("reviewer", source=first) + for environment in (first, reviewer): + result = await environment.exec_cmd( + "git rev-parse --is-shallow-repository && git rev-list --count HEAD && " + "git describe --tags && git show HEAD:answer.py" + ) + assert result.returncode == 0, result.stderr + assert result.stdout.splitlines() == ["true", "1", "v1.2", "answer = 2"] + finally: + await pool.release() + assert _git(source, "rev-parse", "--is-shallow-repository") == "true" + assert _git(source, "rev-list", "--count", "HEAD") == "1" + + @pytest.mark.parametrize("scope", ["group", "session"]) @pytest.mark.parametrize("object_format", ["sha1", "sha256"]) async def test_docker_sessions_execute_and_deliver_before_owned_cleanup(