Environment management

Functions:

Name	Description
local_package_content_identity	Return a deterministic content identity for an immutable local package tree.

__version__ = version('wetlands') module-attribute

EnvironmentManager

Manage isolated Pixi environments without construction-time side effects.

Methods:

Name	Description
provision	Provision or reuse one environment asynchronously.
managed_environments	Discover ready and incomplete environment targets owned by this root.
remove	Remove a managed environment after proving it has no live resources.
running_workers	Return live workers belonging to the environment's current generation.
start_debugger	Lazily start debugpy in a live worker without claiming its task controller.
close	Cancel operations and close pools and processes with bounded cleanup.

Source code in src/wetlands/environment_manager.py

class EnvironmentManager:
    """Manage isolated Pixi environments without construction-time side effects."""

    def __init__(
        self,
        root: str | Path = Path("wetlands"),
        *,
        pixi_executable: str | Path | None = None,
        network: Mapping[str, str] | None = None,
        termination_grace: float = 5.0,
    ) -> None:
        if type(termination_grace) not in {int, float} or not math.isfinite(termination_grace) or termination_grace < 0:
            raise ValueError("termination_grace must be a finite non-negative number")
        self._root = Path(root).expanduser().resolve(strict=False)
        self._pixi_executable = (
            Path(pixi_executable).expanduser().resolve(strict=False) if pixi_executable is not None else None
        )
        normalized_network: dict[str, str] = {}
        for raw_key, raw_value in (network or {}).items():
            key = str(raw_key).lower()
            if key != "no_proxy" and key.endswith("_proxy"):
                key = key.removesuffix("_proxy")
            if key not in _NETWORK_KEYS:
                raise ValueError("network keys must be http, https, or no_proxy (optionally with a _proxy suffix)")
            normalized_network[key] = str(raw_value)
        self._network = MappingProxyType(normalized_network) if normalized_network else None
        self._termination_grace = float(termination_grace)
        self._environments_root = self.root / "environments"
        self._state_root = self.root / "state"

        self._prepare_condition = threading.Condition()
        self._preparing = False
        self._prepared: PixiInfo | None = None
        self._environment_lock = threading.RLock()
        self._environments: dict[str, ManagedEnvironment] = {}
        self._environment_epochs: dict[str, int] = {}
        self._lifecycle_condition = threading.Condition(threading.RLock())
        self._close_lock = threading.Lock()
        self._active_operations: set[Operation[Any]] = set()
        self._active_manager_work = 0
        self._closed = False
        self._close_complete = False
        self._environment_reclaimer = EnvironmentReclaimer(self)

    @property
    def root(self) -> Path:
        return self._root

    @property
    def pixi_executable(self) -> Path | None:
        return self._pixi_executable

    @property
    def network(self) -> Mapping[str, str] | None:
        return self._network

    @property
    def termination_grace(self) -> float:
        return self._termination_grace

    @property
    def environments_root(self) -> Path:
        return self._environments_root

    @property
    def state_root(self) -> Path:
        return self._state_root

    def prepare(self) -> PreparationOperation[PixiInfo]:
        operation: PreparationOperation[PixiInfo] = PreparationOperation()
        self._start_operation(
            operation,
            lambda: self._prepare_sync(operation),
            thread_name=f"wetlands-prepare-{operation.id[:8]}",
        )
        self._environment_reclaimer.wake()
        return operation

    def _prepare_sync(
        self,
        operation: Operation[Any],
        on_mutation_started: Callable[[], None] | None = None,
    ) -> PixiInfo:
        reconcile_shared_memory_leases(self.root)
        with self._prepare_condition:
            while self._preparing:
                if operation.cancellation_requested:
                    raise OperationCanceled(operation.id)
                self._prepare_condition.wait(0.1)
            if self._prepared is not None:
                return self._prepared
            self._preparing = True
        try:
            pixi = prepare_pixi(
                self,
                operation,
                on_mutation_started=on_mutation_started,
            )
        except BaseException:
            with self._prepare_condition:
                self._preparing = False
                self._prepare_condition.notify_all()
            raise
        with self._prepare_condition:
            self._prepared = pixi
            self._preparing = False
            self._prepare_condition.notify_all()
            return pixi

    def provision(
        self,
        name: str,
        spec: EnvironmentSpec,
        *,
        replace_existing: bool = False,
        on_mutation_started: Callable[[], None] | None = None,
    ) -> ProvisioningOperation[ManagedEnvironment]:
        """Provision or reuse one environment asynchronously.

        ``on_mutation_started`` runs at most once on the operation thread,
        immediately before Wetlands first mutates its managed Pixi installation
        or the environment target. It is not called when both are reused without
        changes.
        """

        normalized_name = validate_environment_name(name)
        if not isinstance(spec, EnvironmentSpec):
            raise TypeError("spec must be an EnvironmentSpec")
        if on_mutation_started is not None and not callable(on_mutation_started):
            raise TypeError("on_mutation_started must be callable or None")
        operation: ProvisioningOperation[ManagedEnvironment] = ProvisioningOperation(environment=normalized_name)
        key = environment_name_key(normalized_name)
        with self._environment_lock:
            initial_epoch = self._environment_epochs.get(key, 0)

        def run() -> ManagedEnvironment:
            environment = provision_environment(
                self,
                operation,
                normalized_name,
                spec,
                replace_existing,
                on_mutation_started,
            )
            with self._environment_lock:
                if self._environment_epochs.get(key, 0) != initial_epoch:
                    return environment
                existing = self._environments.get(key)
                if (
                    existing is not None
                    and existing.name == environment.name
                    and existing.generation_id == environment.generation_id
                    and existing.path == environment.path
                ):
                    return existing
                self._environments[key] = environment
            return environment

        self._start_operation(
            operation,
            run,
            thread_name=f"wetlands-provision-{normalized_name}-{operation.id[:8]}",
        )
        self._environment_reclaimer.wake()
        return operation

    def environment(self, name: str) -> ManagedEnvironment:
        with self._manager_work():
            normalized_name = validate_environment_name(name)
            key = environment_name_key(normalized_name)
            with environment_lifecycle_gate(self, normalized_name):
                with self._environment_lock:
                    existing = self._environments.get(key)
                if existing is not None:
                    if existing.name != normalized_name:
                        raise EnvironmentNotReadyError(
                            f"Environment name {normalized_name!r} aliases managed name {existing.name!r}"
                        )
                target = self.environments_root / normalized_name
                metadata = _read_ready(target)
                if metadata is None:
                    with self._environment_lock:
                        self._environments.pop(key, None)
                    raise EnvironmentNotReadyError(f"Environment {normalized_name!r} is not ready")
                if existing is not None and existing.generation_id == metadata.get("generation_id"):
                    return existing
                environment = ManagedEnvironment._from_ready(self, normalized_name, target, metadata)
                with self._environment_lock:
                    self._environments[key] = environment
                return environment

    def managed_environments(self) -> tuple[ManagedEnvironmentInfo, ...]:
        """Discover ready and incomplete environment targets owned by this root."""

        with self._manager_work():
            return discover_managed_environments(self)

    def remove(self, name: str) -> RemovalOperation[ManagedEnvironmentInfo]:
        """Remove a managed environment after proving it has no live resources."""

        normalized_name = validate_environment_name(name)
        operation: RemovalOperation[ManagedEnvironmentInfo] = RemovalOperation(environment=normalized_name)
        self._start_operation(
            operation,
            lambda: remove_managed_environment(self, operation, normalized_name),
            thread_name=f"wetlands-remove-{normalized_name}-{operation.id[:8]}",
        )
        self._environment_reclaimer.wake()
        return operation

    def _quarantine_environment(
        self,
        target: Path,
        *,
        operation_id: str,
        expected_identity: tuple[int, int, int] | None = None,
        before_commit: Callable[[], None] | None = None,
        lock_operation: Operation[Any] | None = None,
    ) -> QuarantinedEnvironment:
        record = quarantine_environment(
            self,
            target,
            operation_id=operation_id,
            expected_identity=expected_identity,
            before_commit=before_commit,
            lock_operation=lock_operation,
        )
        self._environment_reclaimer.enqueue(record)
        return record

    def _running_worker_entries(self, name: str) -> list[dict[str, Any]]:
        environment = self.environment(name)
        return runtime_state.live_workers_for_env(
            self.root,
            environment.name,
            expected_identity={
                "env_path": str(environment.path),
                "generation_id": environment.generation_id,
                "recipe_hash": environment.recipe_hash,
                "worker_runtime_version": WORKER_RUNTIME_VERSION,
                "protocol_version": EXECUTION_PROTOCOL_VERSION,
            },
            include_nonpersistent=True,
        )

    def _cached_environment_generation_in_use(self, name: str) -> str | None:
        """Return the cached generation when controller-owned resources are live."""
        key = environment_name_key(name)
        with self._environment_lock:
            environment = self._environments.get(key)
        if environment is None or environment.name != name or not environment._has_live_resources():
            return None
        return environment.generation_id

    @staticmethod
    def _public_worker(entry: dict[str, Any]) -> RunningWorker:
        raw_debugger = entry.get("debugger")
        debugger = (
            None
            if not isinstance(raw_debugger, dict)
            else DebugEndpoint(
                worker_id=str(entry["worker_id"]),
                adapter="debugpy",
                host=str(raw_debugger["host"]),
                port=int(raw_debugger["port"]),
            )
        )
        return RunningWorker(
            id=str(entry["worker_id"]),
            environment=str(entry["env_name"]),
            pool_id=str(entry["pool_id"]) if entry.get("pool_id") is not None else None,
            index=int(entry["worker_index"]),
            process_id=int(entry["pid"]),
            persistent=bool(entry["persistent"]),
            debugger=debugger,
        )

    def running_workers(self, environment: str) -> tuple[RunningWorker, ...]:
        """Return live workers belonging to the environment's current generation."""
        with self._manager_work():
            normalized_name = validate_environment_name(environment)
            return tuple(self._public_worker(entry) for entry in self._running_worker_entries(normalized_name))

    def start_debugger(
        self,
        environment: str,
        *,
        worker: str | None = None,
    ) -> DebugEndpoint:
        """Lazily start debugpy in a live worker without claiming its task controller."""
        with self._manager_work():
            normalized_name = validate_environment_name(environment)
            entries = self._running_worker_entries(normalized_name)
            if worker is None:
                if not entries:
                    raise RuntimeError(f"Environment {normalized_name!r} has no running workers")
                if len(entries) != 1:
                    raise ValueError(
                        f"Environment {normalized_name!r} has {len(entries)} running workers; select one by ID"
                    )
                entry = entries[0]
            else:
                if not isinstance(worker, str) or not worker:
                    raise ValueError("worker must be a nonempty worker ID")
                matches = [entry for entry in entries if entry.get("worker_id") == worker]
                if not matches:
                    raise ValueError(f"Worker {worker!r} is not running in environment {normalized_name!r}")
                entry = matches[0]

            authkey = runtime_state.load_or_create_root_authkey(self.root)
            response = management.start_debugger(entry, authkey)
            adapter = response.get("adapter")
            host = response.get("host")
            port = response.get("port")
            if adapter != "debugpy" or host != "127.0.0.1" or type(port) is not int or not (0 < port <= 65535):
                raise management.ManagementConnectionError("Worker returned an invalid debugger endpoint")
            endpoint = DebugEndpoint(
                worker_id=str(entry["worker_id"]),
                adapter="debugpy",
                host=host,
                port=port,
            )
            runtime_state.record_debugger(
                self.root,
                worker_id=endpoint.worker_id,
                adapter=endpoint.adapter,
                host=endpoint.host,
                port=endpoint.port,
            )
            return endpoint

    def close(self, *, timeout: float | None = None) -> None:
        """Cancel operations and close pools and processes with bounded cleanup.

        Failed resource cleanup is reported with :class:`ManagerCloseError`; a
        later call retries resources whose cleanup did not complete.
        """
        if timeout is not None and (type(timeout) not in {int, float} or not math.isfinite(timeout) or timeout < 0):
            raise ValueError("timeout must be a finite non-negative number or None")
        normalized_timeout = None if timeout is None else float(timeout)
        deadline = None if normalized_timeout is None else time.monotonic() + normalized_timeout

        lock_timeout = -1 if deadline is None else max(0.0, deadline - time.monotonic())
        if not self._close_lock.acquire(timeout=lock_timeout):
            assert normalized_timeout is not None
            raise ManagerCloseError((ManagerCloseTimeoutError("close serialization", normalized_timeout),))
        try:
            with self._lifecycle_condition:
                if self._close_complete:
                    return
                operations = tuple(self._active_operations)
                if any(operation._runs_on_current_thread() for operation in operations):
                    raise RuntimeError(
                        "EnvironmentManager.close() cannot run from an active operation listener; "
                        "schedule shutdown on another thread"
                    )
                self._closed = True

            errors: list[BaseException] = []
            for operation in operations:
                operation.cancel()
            lifecycle_ready = True
            for operation in operations:
                try:
                    remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
                    operation.wait_for(remaining)
                except OperationCanceled:
                    pass
                except TimeoutError:
                    assert normalized_timeout is not None
                    errors.append(ManagerCloseTimeoutError("active operations", normalized_timeout))
                    lifecycle_ready = False
                    break
                except BaseException as error:
                    errors.append(error)

            if lifecycle_ready:
                with self._lifecycle_condition:
                    while self._active_manager_work:
                        remaining = None if deadline is None else deadline - time.monotonic()
                        if remaining is not None and remaining <= 0:
                            assert normalized_timeout is not None
                            errors.append(ManagerCloseTimeoutError("manager work", normalized_timeout))
                            lifecycle_ready = False
                            break
                        self._lifecycle_condition.wait(remaining)

            # Provisioning publishes its ManagedEnvironment before it reaches a
            # terminal state, so taking this snapshot after joining operations
            # cannot miss an environment that completed concurrently with close.
            environments: tuple[ManagedEnvironment, ...] = ()
            if lifecycle_ready:
                with self._environment_lock:
                    environments = tuple(self._environments.values())
                pool_attempts = tuple(
                    (environment, environment._start_pool_close_attempts()) for environment in environments
                )
                process_attempts = tuple(
                    (environment, environment._start_process_close_attempts()) for environment in environments
                )
                for environment, pool_environment_attempts in pool_attempts:
                    errors.extend(
                        environment._collect_pool_close_attempts(
                            pool_environment_attempts,
                            deadline=deadline,
                            timeout=normalized_timeout,
                        )
                    )
                for environment, process_environment_attempts in process_attempts:
                    errors.extend(
                        environment._collect_process_close_attempts(
                            process_environment_attempts,
                            deadline=deadline,
                            timeout=normalized_timeout,
                        )
                    )

            with self._lifecycle_condition:
                remaining_resources = any(environment._has_live_resources() for environment in environments)
            if deadline is None:
                # Background physical reclamation was always a bounded,
                # best-effort part of close(). Keep the reclaimer's default wait
                # instead of turning timeout=None into an indefinite join.
                self._environment_reclaimer.close()
                reclaimer_closed = True
            else:
                reclaimer_timeout = max(0.0, deadline - time.monotonic())
                reclaimer_closed = self._environment_reclaimer.close(timeout=reclaimer_timeout)
                if not reclaimer_closed:
                    assert normalized_timeout is not None
                    errors.append(ManagerCloseTimeoutError("environment reclaimer", normalized_timeout))
            with self._lifecycle_condition:
                self._close_complete = lifecycle_ready and not remaining_resources and reclaimer_closed
            if errors:
                raise ManagerCloseError(tuple(errors))
        finally:
            self._close_lock.release()

    def _ensure_open(self) -> None:
        with self._lifecycle_condition:
            self._ensure_open_locked()

    def _ensure_open_locked(self) -> None:
        if self._closed:
            raise RuntimeError("EnvironmentManager is closed")

    def _start_operation(
        self,
        operation: Operation[Any],
        runner: Callable[[], Any],
        *,
        thread_name: str,
    ) -> None:
        with self._lifecycle_condition:
            self._ensure_open_locked()
            self._active_operations.add(operation)

            def unregister(event: OperationEvent) -> None:
                if event.state.terminal:
                    operation.remove_listener(unregister)
                    self._unregister_operation(operation)

            operation.listen(unregister, replay=False)
            try:
                operation._start_runner(runner, thread_name=thread_name)
            except BaseException:
                operation.remove_listener(unregister)
                self._active_operations.discard(operation)
                self._lifecycle_condition.notify_all()
                raise

    def _unregister_operation(self, operation: Operation[Any]) -> None:
        with self._lifecycle_condition:
            self._active_operations.discard(operation)
            self._lifecycle_condition.notify_all()

    @contextmanager
    def _manager_work(self) -> Iterator[None]:
        """Keep shutdown from overtaking synchronous manager-owned work."""

        with self._lifecycle_condition:
            self._ensure_open_locked()
            self._active_manager_work += 1
        try:
            yield
        finally:
            with self._lifecycle_condition:
                self._active_manager_work -= 1
                self._lifecycle_condition.notify_all()

    def __enter__(self) -> EnvironmentManager:
        self._ensure_open()
        return self

    def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
        self.close()

provision(name, spec, *, replace_existing=False, on_mutation_started=None)

Provision or reuse one environment asynchronously.

on_mutation_started runs at most once on the operation thread, immediately before Wetlands first mutates its managed Pixi installation or the environment target. It is not called when both are reused without changes.

Source code in src/wetlands/environment_manager.py

def provision(
    self,
    name: str,
    spec: EnvironmentSpec,
    *,
    replace_existing: bool = False,
    on_mutation_started: Callable[[], None] | None = None,
) -> ProvisioningOperation[ManagedEnvironment]:
    """Provision or reuse one environment asynchronously.

    ``on_mutation_started`` runs at most once on the operation thread,
    immediately before Wetlands first mutates its managed Pixi installation
    or the environment target. It is not called when both are reused without
    changes.
    """

    normalized_name = validate_environment_name(name)
    if not isinstance(spec, EnvironmentSpec):
        raise TypeError("spec must be an EnvironmentSpec")
    if on_mutation_started is not None and not callable(on_mutation_started):
        raise TypeError("on_mutation_started must be callable or None")
    operation: ProvisioningOperation[ManagedEnvironment] = ProvisioningOperation(environment=normalized_name)
    key = environment_name_key(normalized_name)
    with self._environment_lock:
        initial_epoch = self._environment_epochs.get(key, 0)

    def run() -> ManagedEnvironment:
        environment = provision_environment(
            self,
            operation,
            normalized_name,
            spec,
            replace_existing,
            on_mutation_started,
        )
        with self._environment_lock:
            if self._environment_epochs.get(key, 0) != initial_epoch:
                return environment
            existing = self._environments.get(key)
            if (
                existing is not None
                and existing.name == environment.name
                and existing.generation_id == environment.generation_id
                and existing.path == environment.path
            ):
                return existing
            self._environments[key] = environment
        return environment

    self._start_operation(
        operation,
        run,
        thread_name=f"wetlands-provision-{normalized_name}-{operation.id[:8]}",
    )
    self._environment_reclaimer.wake()
    return operation

managed_environments()

Discover ready and incomplete environment targets owned by this root.

Source code in src/wetlands/environment_manager.py

def managed_environments(self) -> tuple[ManagedEnvironmentInfo, ...]:
    """Discover ready and incomplete environment targets owned by this root."""

    with self._manager_work():
        return discover_managed_environments(self)

remove(name)

Remove a managed environment after proving it has no live resources.

Source code in src/wetlands/environment_manager.py

def remove(self, name: str) -> RemovalOperation[ManagedEnvironmentInfo]:
    """Remove a managed environment after proving it has no live resources."""

    normalized_name = validate_environment_name(name)
    operation: RemovalOperation[ManagedEnvironmentInfo] = RemovalOperation(environment=normalized_name)
    self._start_operation(
        operation,
        lambda: remove_managed_environment(self, operation, normalized_name),
        thread_name=f"wetlands-remove-{normalized_name}-{operation.id[:8]}",
    )
    self._environment_reclaimer.wake()
    return operation

running_workers(environment)

Return live workers belonging to the environment's current generation.

Source code in src/wetlands/environment_manager.py

def running_workers(self, environment: str) -> tuple[RunningWorker, ...]:
    """Return live workers belonging to the environment's current generation."""
    with self._manager_work():
        normalized_name = validate_environment_name(environment)
        return tuple(self._public_worker(entry) for entry in self._running_worker_entries(normalized_name))

start_debugger(environment, *, worker=None)

Lazily start debugpy in a live worker without claiming its task controller.

Source code in src/wetlands/environment_manager.py

def start_debugger(
    self,
    environment: str,
    *,
    worker: str | None = None,
) -> DebugEndpoint:
    """Lazily start debugpy in a live worker without claiming its task controller."""
    with self._manager_work():
        normalized_name = validate_environment_name(environment)
        entries = self._running_worker_entries(normalized_name)
        if worker is None:
            if not entries:
                raise RuntimeError(f"Environment {normalized_name!r} has no running workers")
            if len(entries) != 1:
                raise ValueError(
                    f"Environment {normalized_name!r} has {len(entries)} running workers; select one by ID"
                )
            entry = entries[0]
        else:
            if not isinstance(worker, str) or not worker:
                raise ValueError("worker must be a nonempty worker ID")
            matches = [entry for entry in entries if entry.get("worker_id") == worker]
            if not matches:
                raise ValueError(f"Worker {worker!r} is not running in environment {normalized_name!r}")
            entry = matches[0]

        authkey = runtime_state.load_or_create_root_authkey(self.root)
        response = management.start_debugger(entry, authkey)
        adapter = response.get("adapter")
        host = response.get("host")
        port = response.get("port")
        if adapter != "debugpy" or host != "127.0.0.1" or type(port) is not int or not (0 < port <= 65535):
            raise management.ManagementConnectionError("Worker returned an invalid debugger endpoint")
        endpoint = DebugEndpoint(
            worker_id=str(entry["worker_id"]),
            adapter="debugpy",
            host=host,
            port=port,
        )
        runtime_state.record_debugger(
            self.root,
            worker_id=endpoint.worker_id,
            adapter=endpoint.adapter,
            host=endpoint.host,
            port=endpoint.port,
        )
        return endpoint

close(*, timeout=None)

Cancel operations and close pools and processes with bounded cleanup.

Failed resource cleanup is reported with :class:ManagerCloseError; a later call retries resources whose cleanup did not complete.

Source code in src/wetlands/environment_manager.py

def close(self, *, timeout: float | None = None) -> None:
    """Cancel operations and close pools and processes with bounded cleanup.

    Failed resource cleanup is reported with :class:`ManagerCloseError`; a
    later call retries resources whose cleanup did not complete.
    """
    if timeout is not None and (type(timeout) not in {int, float} or not math.isfinite(timeout) or timeout < 0):
        raise ValueError("timeout must be a finite non-negative number or None")
    normalized_timeout = None if timeout is None else float(timeout)
    deadline = None if normalized_timeout is None else time.monotonic() + normalized_timeout

    lock_timeout = -1 if deadline is None else max(0.0, deadline - time.monotonic())
    if not self._close_lock.acquire(timeout=lock_timeout):
        assert normalized_timeout is not None
        raise ManagerCloseError((ManagerCloseTimeoutError("close serialization", normalized_timeout),))
    try:
        with self._lifecycle_condition:
            if self._close_complete:
                return
            operations = tuple(self._active_operations)
            if any(operation._runs_on_current_thread() for operation in operations):
                raise RuntimeError(
                    "EnvironmentManager.close() cannot run from an active operation listener; "
                    "schedule shutdown on another thread"
                )
            self._closed = True

        errors: list[BaseException] = []
        for operation in operations:
            operation.cancel()
        lifecycle_ready = True
        for operation in operations:
            try:
                remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
                operation.wait_for(remaining)
            except OperationCanceled:
                pass
            except TimeoutError:
                assert normalized_timeout is not None
                errors.append(ManagerCloseTimeoutError("active operations", normalized_timeout))
                lifecycle_ready = False
                break
            except BaseException as error:
                errors.append(error)

        if lifecycle_ready:
            with self._lifecycle_condition:
                while self._active_manager_work:
                    remaining = None if deadline is None else deadline - time.monotonic()
                    if remaining is not None and remaining <= 0:
                        assert normalized_timeout is not None
                        errors.append(ManagerCloseTimeoutError("manager work", normalized_timeout))
                        lifecycle_ready = False
                        break
                    self._lifecycle_condition.wait(remaining)

        # Provisioning publishes its ManagedEnvironment before it reaches a
        # terminal state, so taking this snapshot after joining operations
        # cannot miss an environment that completed concurrently with close.
        environments: tuple[ManagedEnvironment, ...] = ()
        if lifecycle_ready:
            with self._environment_lock:
                environments = tuple(self._environments.values())
            pool_attempts = tuple(
                (environment, environment._start_pool_close_attempts()) for environment in environments
            )
            process_attempts = tuple(
                (environment, environment._start_process_close_attempts()) for environment in environments
            )
            for environment, pool_environment_attempts in pool_attempts:
                errors.extend(
                    environment._collect_pool_close_attempts(
                        pool_environment_attempts,
                        deadline=deadline,
                        timeout=normalized_timeout,
                    )
                )
            for environment, process_environment_attempts in process_attempts:
                errors.extend(
                    environment._collect_process_close_attempts(
                        process_environment_attempts,
                        deadline=deadline,
                        timeout=normalized_timeout,
                    )
                )

        with self._lifecycle_condition:
            remaining_resources = any(environment._has_live_resources() for environment in environments)
        if deadline is None:
            # Background physical reclamation was always a bounded,
            # best-effort part of close(). Keep the reclaimer's default wait
            # instead of turning timeout=None into an indefinite join.
            self._environment_reclaimer.close()
            reclaimer_closed = True
        else:
            reclaimer_timeout = max(0.0, deadline - time.monotonic())
            reclaimer_closed = self._environment_reclaimer.close(timeout=reclaimer_timeout)
            if not reclaimer_closed:
                assert normalized_timeout is not None
                errors.append(ManagerCloseTimeoutError("environment reclaimer", normalized_timeout))
        with self._lifecycle_condition:
            self._close_complete = lifecycle_ready and not remaining_resources and reclaimer_closed
        if errors:
            raise ManagerCloseError(tuple(errors))
    finally:
        self._close_lock.release()

EnvironmentSpec dataclass

The complete immutable recipe for a managed Pixi environment.

Dependency strings use Pixi's Conda syntax in :attr:conda and PEP 508 requirement syntax in :attr:pypi.

Methods:

Name	Description
normalized	Return the canonical recipe representation used for identity.

Source code in src/wetlands/specs.py

@dataclass(frozen=True)
class EnvironmentSpec:
    """The complete immutable recipe for a managed Pixi environment.

    Dependency strings use Pixi's Conda syntax in :attr:`conda` and PEP 508
    requirement syntax in :attr:`pypi`.
    """

    python: str = ">=3.9"
    conda: tuple[str, ...] = ()
    pypi: tuple[str, ...] = ()
    channels: tuple[str, ...] = ("conda-forge",)
    local: tuple[LocalPackage, ...] = ()
    post_install: tuple[PostInstallCommand, ...] = ()
    pixi_lock: bytes | os.PathLike[str] | None = field(default=None, repr=False)
    _lock_bytes: bytes | None = field(init=False, default=None, repr=False, compare=True)

    def __post_init__(self) -> None:
        python = str(self.python).strip()
        if not python:
            raise ValueError("Python constraint cannot be empty")
        conda = _nonempty_strings(self.conda, "Conda dependency")
        conda_names: set[str] = set()
        for dependency in conda:
            if "::" in dependency:
                raise ValueError(
                    f"Channel-qualified Conda dependencies are not supported: {dependency!r}. "
                    "Declare channels with EnvironmentSpec(channels=...)."
                )
            match = re.match(r"^([A-Za-z0-9_.-]+)", dependency)
            if match is None:
                raise ValueError(f"Invalid Conda dependency: {dependency!r}")
            package = canonicalize_name(match.group(1))
            if package in _MANAGED_RUNTIME_PACKAGE_NAMES:
                raise ValueError(f"Conda package {match.group(1)!r} is managed by the Wetlands worker runtime")
            if package in conda_names:
                raise ValueError(f"Duplicate Conda dependency for package {match.group(1)!r}")
            conda_names.add(package)
        pypi = _nonempty_strings(self.pypi, "PyPI dependency")
        pypi_names: set[str] = set()
        for dependency in pypi:
            try:
                requirement = Requirement(dependency)
            except InvalidRequirement as error:
                raise ValueError(f"Invalid PyPI dependency: {dependency!r}") from error
            if requirement.marker is not None:
                raise ValueError(f"PyPI environment markers are not supported in EnvironmentSpec: {dependency!r}")
            if requirement.url is not None:
                parsed_url = urllib.parse.urlsplit(requirement.url)
                if parsed_url.username or parsed_url.password or parsed_url.query:
                    raise ValueError("PyPI direct URLs cannot contain credentials or query parameters")
                if parsed_url.scheme.startswith("git+"):
                    _parse_pinned_git_url(requirement.url)
            package = canonicalize_name(requirement.name)
            if package in _MANAGED_RUNTIME_PACKAGE_NAMES:
                raise ValueError(f"PyPI package {requirement.name!r} is managed by the Wetlands worker runtime")
            if package in pypi_names:
                raise ValueError(f"Duplicate PyPI dependency for package {requirement.name!r}")
            pypi_names.add(package)
        channels = tuple(dict.fromkeys(_nonempty_strings(self.channels, "Channel")))
        if not channels:
            raise ValueError("At least one Pixi channel is required")
        local = tuple(self.local)
        if any(not isinstance(package, LocalPackage) for package in local):
            raise TypeError("local entries must be LocalPackage instances")
        local_names: set[str] = set()
        for local_package in local:
            if local_package.distribution_name in _MANAGED_RUNTIME_PACKAGE_NAMES:
                raise ValueError(
                    f"Local package {local_package.distribution_name!r} is managed by the Wetlands worker runtime"
                )
            if local_package.distribution_name in pypi_names:
                raise ValueError(
                    f"Local package {local_package.distribution_name!r} duplicates a declared PyPI dependency"
                )
            if local_package.distribution_name in local_names:
                raise ValueError(f"Duplicate local package {local_package.distribution_name!r}")
            local_names.add(local_package.distribution_name)
        post_install = tuple(self.post_install)
        if any(not isinstance(command, PostInstallCommand) for command in post_install):
            raise TypeError("post_install entries must be PostInstallCommand instances")
        object.__setattr__(self, "python", python)
        object.__setattr__(self, "conda", conda)
        object.__setattr__(self, "pypi", pypi)
        object.__setattr__(self, "channels", channels)
        object.__setattr__(self, "local", local)
        object.__setattr__(self, "post_install", post_install)
        lock = self.pixi_lock
        if lock is None:
            lock_bytes = None
        elif isinstance(lock, bytes):
            lock_bytes = bytes(lock)
        else:
            lock_bytes = Path(lock).read_bytes()
        object.__setattr__(self, "_lock_bytes", lock_bytes)
        object.__setattr__(self, "pixi_lock", None)

    @property
    def lock_bytes(self) -> bytes | None:
        """Return an independent copy of the supplied lockfile bytes, if any."""
        return self._lock_bytes

    def normalized(self) -> dict[str, Any]:
        """Return the canonical recipe representation used for identity."""
        return {
            "python": self.python.strip(),
            "conda": sorted(set(self.conda)),
            "pypi": sorted(set(self.pypi)),
            "channels": list(self.channels),
            "local": [
                {
                    **(
                        {"content_identity": package.content_identity}
                        if package.content_identity is not None
                        else {"source": str(package.source)}
                    ),
                    "distribution_name": package.distribution_name,
                    "editable": package.editable,
                    "extras": list(package.extras),
                }
                for package in self.local
            ],
            "post_install": [
                {
                    "argv": list(command.argv),
                    "shell": command.shell,
                    "display": command.display,
                }
                for command in self.post_install
            ],
            "pixi_lock_sha256": (
                hashlib.sha256(self._lock_bytes).hexdigest() if self._lock_bytes is not None else None
            ),
            "managed_runtime": {
                "pypi": list(MANAGED_RUNTIME_PYPI),
            },
        }

    @property
    def recipe_hash(self) -> str:
        """Return the SHA-256 identity of the normalized recipe."""
        payload = json.dumps(self.normalized(), sort_keys=True, separators=(",", ":")).encode()
        return hashlib.sha256(payload).hexdigest()

lock_bytes property

Return an independent copy of the supplied lockfile bytes, if any.

recipe_hash property

Return the SHA-256 identity of the normalized recipe.

normalized()

Return the canonical recipe representation used for identity.

Source code in src/wetlands/specs.py

def normalized(self) -> dict[str, Any]:
    """Return the canonical recipe representation used for identity."""
    return {
        "python": self.python.strip(),
        "conda": sorted(set(self.conda)),
        "pypi": sorted(set(self.pypi)),
        "channels": list(self.channels),
        "local": [
            {
                **(
                    {"content_identity": package.content_identity}
                    if package.content_identity is not None
                    else {"source": str(package.source)}
                ),
                "distribution_name": package.distribution_name,
                "editable": package.editable,
                "extras": list(package.extras),
            }
            for package in self.local
        ],
        "post_install": [
            {
                "argv": list(command.argv),
                "shell": command.shell,
                "display": command.display,
            }
            for command in self.post_install
        ],
        "pixi_lock_sha256": (
            hashlib.sha256(self._lock_bytes).hexdigest() if self._lock_bytes is not None else None
        ),
        "managed_runtime": {
            "pypi": list(MANAGED_RUNTIME_PYPI),
        },
    }

LocalPackage dataclass

An installable local Python package included in an environment recipe.

The source must contain a PEP 621 pyproject.toml with [project].name.

Source code in src/wetlands/specs.py

@dataclass(frozen=True)
class LocalPackage:
    """An installable local Python package included in an environment recipe.

    The source must contain a PEP 621 ``pyproject.toml`` with ``[project].name``.
    """

    source: Path
    editable: bool = False
    extras: tuple[str, ...] = ()
    content_identity: str | None = None
    distribution_name: str = field(init=False)

    def __post_init__(self) -> None:
        requested_source = Path(self.source).expanduser()
        if self.content_identity is not None:
            try:
                requested_metadata = requested_source.lstat()
            except OSError as error:
                raise LocalPackageValidationError(
                    f"Could not inspect local package source {requested_source}: {error}"
                ) from error
            if _is_link_or_reparse(requested_metadata):
                raise LocalPackageValidationError(
                    f"A content-identified local package source cannot be a link or reparse point: {requested_source}"
                )
        source = requested_source.resolve()
        pyproject = source / "pyproject.toml"
        if not source.is_dir():
            raise LocalPackageValidationError(f"Local package source must be an existing directory: {source}")
        if not pyproject.is_file():
            raise LocalPackageValidationError(f"Local package {source} must contain pyproject.toml with [project].name")
        try:
            document = tomllib.loads(pyproject.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, tomllib.TOMLDecodeError) as error:
            raise LocalPackageValidationError(
                f"Could not read valid TOML from local package {pyproject}: {error}"
            ) from error
        project = document.get("project")
        declared_name = project.get("name") if isinstance(project, dict) else None
        if not isinstance(declared_name, str) or not declared_name:
            raise LocalPackageValidationError(f"Local package {pyproject} must declare a non-empty [project].name")
        try:
            distribution_name = canonicalize_name(declared_name, validate=True)
        except InvalidName as error:
            raise LocalPackageValidationError(
                f"Local package {pyproject} has invalid [project].name {declared_name!r}"
            ) from error
        object.__setattr__(self, "source", source)
        object.__setattr__(self, "distribution_name", distribution_name)
        content_identity = self.content_identity
        if content_identity is not None:
            if (
                not isinstance(content_identity, str)
                or _LOCAL_PACKAGE_IDENTITY_PATTERN.fullmatch(content_identity) is None
            ):
                raise ValueError("Local package content_identity must be 'sha256:' followed by 64 hexadecimal digits")
            if self.editable:
                raise ValueError("A content-identified local package cannot be editable")
            object.__setattr__(self, "content_identity", content_identity.lower())
        if isinstance(self.extras, str):
            raise TypeError("Local package extras must be a sequence of names, not a string")
        extras = tuple(str(extra) for extra in self.extras)
        if any(not _PORTABLE_EXTRA.fullmatch(extra) for extra in extras):
            raise ValueError("Local package extras must be valid Python distribution extras")
        object.__setattr__(self, "extras", extras)

ManagedEnvironment

A verified, ready Pixi environment managed by one manager root.

Instances are returned by :meth:EnvironmentManager.provision or :meth:EnvironmentManager.environment; applications do not construct them directly.

Methods:

Name	Description
spawn	Launch an independently supervised command in this generation.
run	Run a command to completion and return its Wetlands-owned result.
start	Start a new warm worker pool for this environment generation.
close_pools	Close every worker pool started through this environment handle.
attach_pool	Exclusively attach to this generation's detached persistent pool.

Source code in src/wetlands/managed_environment.py

class ManagedEnvironment:
    """A verified, ready Pixi environment managed by one manager root.

    Instances are returned by :meth:`EnvironmentManager.provision` or
    :meth:`EnvironmentManager.environment`; applications do not construct them directly.
    """

    def __init__(
        self,
        manager: EnvironmentManager,
        name: str,
        path: Path,
        metadata: dict[str, Any],
    ) -> None:
        self._manager = manager
        self._name = name
        self._path = path.resolve()
        self._metadata = dict(metadata)
        self._pools: list[WorkerPool] = []
        self._pool_close_attempts: dict[int, _PoolCloseAttempt] = {}
        self._processes: list[ManagedProcess] = []
        self._process_close_attempts: dict[int, _ProcessCloseAttempt] = {}
        self._lock = threading.RLock()

    @classmethod
    def _from_ready(
        cls,
        manager: EnvironmentManager,
        name: str,
        path: Path,
        metadata: dict[str, Any],
    ) -> ManagedEnvironment:
        return cls(manager, name, path, metadata)

    @property
    def name(self) -> str:
        """Return the managed environment name."""
        return self._name

    @property
    def path(self) -> Path:
        """Return the canonical Pixi project directory."""
        return self._path

    @property
    def pixi_manifest_path(self) -> Path:
        """Return the generated ``pixi.toml`` path."""
        return self._path / "pixi.toml"

    @property
    def pixi_lock_path(self) -> Path:
        """Return the resolved or supplied ``pixi.lock`` path."""
        return self._path / "pixi.lock"

    @property
    def pixi_version(self) -> str:
        """Return the Pixi version recorded when this generation was provisioned."""
        return str(self._metadata["pixi_version"])

    @property
    def pixi_executable_path(self) -> Path:
        """Return the Pixi executable used to provision this generation."""
        return Path(self._metadata["pixi_executable"])

    @property
    def generation_id(self) -> str:
        """Return the unique identifier for this published environment generation."""
        return str(self._metadata["generation_id"])

    @property
    def recipe_hash(self) -> str:
        """Return the hash of the complete normalized environment recipe."""
        return str(self._metadata["recipe_hash"])

    @property
    def lockfile_hash(self) -> str:
        """Return the SHA-256 hash of this generation's lockfile."""
        return str(self._metadata["lock_sha256"])

    def spawn(
        self,
        argv: Sequence[str],
        *,
        cwd: str | Path | None = None,
        env: Mapping[str, str | None] | None = None,
        output_limit: int = 1_048_576,
    ) -> ManagedProcess:
        """Launch an independently supervised command in this generation."""
        from wetlands.managed_process import ManagedProcess, _validate_launch_options

        options = _validate_launch_options(
            argv=argv,
            cwd=cwd,
            env=env,
            output_limit=output_limit,
            default_cwd=self.path,
        )
        with self._manager._manager_work():
            with environment_lifecycle_gate(self._manager, self.name):
                self._require_current_generation()
                runtime_state.reconcile_persistent_pool(
                    self._manager.root,
                    self.name,
                    grace=self._manager.termination_grace,
                )
                return ManagedProcess._launch_validated(
                    environment=self,
                    options=options,
                )

    def run(
        self,
        argv: Sequence[str],
        *,
        cwd: str | Path | None = None,
        env: Mapping[str, str | None] | None = None,
        timeout: float | None = None,
        output_limit: int = 1_048_576,
        check: bool = True,
    ) -> ManagedProcessResult:
        """Run a command to completion and return its Wetlands-owned result."""
        from wetlands.managed_process import _validate_check, _validate_timeout

        normalized_timeout = _validate_timeout(timeout)
        normalized_check = _validate_check(check)
        process = self.spawn(argv, cwd=cwd, env=env, output_limit=output_limit)
        try:
            return process.wait(timeout=normalized_timeout, check=normalized_check)
        finally:
            process.close()

    def start(
        self,
        *,
        workers: int = 1,
        persistent: bool = False,
        worker_environment: Callable[[int], Mapping[str, str]] | None = None,
        worker_timeout: float | None = None,
    ) -> WorkerPool:
        """Start a new warm worker pool for this environment generation.

        Args:
            workers: Number of worker processes in the pool.
            persistent: Keep workers alive when the controller deliberately detaches.
            worker_environment: Optional callable receiving each zero-based worker
                index and returning environment variables for that worker. Wetlands
                snapshots the mappings before launch and reuses the mapping for the
                same index when replacing a worker. This cannot be combined with
                ``persistent=True``.
            worker_timeout: Optional worker inactivity timeout in seconds.
                Each IPC message resets the timer, so this is a health check rather
                than a maximum task execution time.
        """
        with self._manager._manager_work():
            if workers < 1:
                raise ValueError("workers must be at least one")
            if persistent and worker_environment is not None:
                raise ValueError("worker_environment cannot be combined with persistent=True")
            worker_environments = _validate_worker_environments(workers, worker_environment)
            snapshotted_worker_environment = worker_environments.__getitem__ if worker_environment is not None else None
            with self._lock:
                existing_pools = tuple(self._pools)
            for existing_pool in existing_pools:
                if not existing_pool._closed:
                    existing_pool._runtime._raise_if_failed()
            with environment_lifecycle_gate(self._manager, self.name):
                self._require_current_generation()
                runtime_state.reconcile_persistent_pool(
                    self._manager.root,
                    self.name,
                    grace=self._manager.termination_grace,
                )
                runtime = ExternalEnvironment(
                    self.name,
                    self.pixi_manifest_path,
                    self._manager,
                    expected_generation_id=self.generation_id,
                    expected_recipe_hash=self.recipe_hash,
                )
                pool = WorkerPool(self, runtime)
                with self._lock:
                    self._pools.append(pool)
                try:
                    runtime.launch(
                        max_workers=workers,
                        persistent=persistent,
                        worker_environment=snapshotted_worker_environment,
                        worker_timeout=worker_timeout,
                    )
                except BaseException:
                    if not runtime._workers:
                        with self._lock:
                            self._pools.remove(pool)
                        pool._closed = True
                    raise
            return pool

    def close_pools(self) -> None:
        """Close every worker pool started through this environment handle."""
        errors = self._close_pools()
        if errors:
            raise ManagerCloseError(errors)

    def _register_process(self, process: ManagedProcess) -> None:
        """Retain generation ownership until process cleanup is proven."""
        with self._lock:
            if not any(existing is process for existing in self._processes):
                self._processes.append(process)

    def _release_process(self, process: ManagedProcess) -> None:
        """Release a process after its supervisor proves the owned tree clean."""
        with self._lock:
            self._processes[:] = [existing for existing in self._processes if existing is not process]
            self._process_close_attempts.pop(id(process), None)

    def _close_pool(self, pool: WorkerPool, attempt: _PoolCloseAttempt) -> None:
        try:
            pool.close()
        except BaseException as error:
            attempt.error = error
        finally:
            attempt.done.set()

    def _close_pools(
        self,
        *,
        deadline: float | None = None,
        timeout: float | None = None,
    ) -> tuple[BaseException, ...]:
        attempts = self._start_pool_close_attempts()
        return self._collect_pool_close_attempts(attempts, deadline=deadline, timeout=timeout)

    def _start_pool_close_attempts(self) -> tuple[tuple[WorkerPool, _PoolCloseAttempt], ...]:
        with self._lock:
            attempts: list[tuple[WorkerPool, _PoolCloseAttempt]] = []
            for pool in self._pools:
                if pool._closed is True:
                    continue
                key = id(pool)
                attempt = self._pool_close_attempts.get(key)
                if attempt is None:
                    attempt = _PoolCloseAttempt()
                    self._pool_close_attempts[key] = attempt
                    threading.Thread(
                        target=self._close_pool,
                        args=(pool, attempt),
                        name=f"wetlands-pool-close-{self.name}",
                        daemon=True,
                    ).start()
                attempts.append((pool, attempt))
        return tuple(attempts)

    def _collect_pool_close_attempts(
        self,
        attempts: tuple[tuple[WorkerPool, _PoolCloseAttempt], ...],
        *,
        deadline: float | None,
        timeout: float | None,
    ) -> tuple[BaseException, ...]:
        errors: list[BaseException] = []
        for pool, attempt in attempts:
            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
            if not attempt.done.wait(remaining):
                assert timeout is not None
                errors.append(ManagerCloseTimeoutError(f"worker pools for {self.name}", timeout))
                continue
            with self._lock:
                if self._pool_close_attempts.get(id(pool)) is attempt:
                    self._pool_close_attempts.pop(id(pool), None)
            if attempt.error is not None:
                errors.append(attempt.error)
        return tuple(errors)

    def _close_process(self, process: ManagedProcess, attempt: _ProcessCloseAttempt) -> None:
        try:
            process.close()
        except BaseException as error:
            attempt.error = error
        finally:
            attempt.done.set()

    def _start_process_close_attempts(self) -> tuple[tuple[ManagedProcess, _ProcessCloseAttempt], ...]:
        with self._lock:
            attempts: list[tuple[ManagedProcess, _ProcessCloseAttempt]] = []
            for process in self._processes:
                key = id(process)
                attempt = self._process_close_attempts.get(key)
                if attempt is None:
                    attempt = _ProcessCloseAttempt()
                    self._process_close_attempts[key] = attempt
                    threading.Thread(
                        target=self._close_process,
                        args=(process, attempt),
                        name=f"wetlands-process-close-{self.name}",
                        daemon=True,
                    ).start()
                attempts.append((process, attempt))
        return tuple(attempts)

    def _collect_process_close_attempts(
        self,
        attempts: tuple[tuple[ManagedProcess, _ProcessCloseAttempt], ...],
        *,
        deadline: float | None,
        timeout: float | None,
    ) -> tuple[BaseException, ...]:
        errors: list[BaseException] = []
        for process, attempt in attempts:
            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
            if not attempt.done.wait(remaining):
                assert timeout is not None
                errors.append(ManagerCloseTimeoutError(f"managed processes for {self.name}", timeout))
                continue
            with self._lock:
                if self._process_close_attempts.get(id(process)) is attempt:
                    self._process_close_attempts.pop(id(process), None)
            if attempt.error is not None:
                errors.append(attempt.error)
        return tuple(errors)

    def _has_open_pools(self) -> bool:
        with self._lock:
            return any(not pool._closed for pool in self._pools)

    def _has_live_resources(self) -> bool:
        with self._lock:
            return bool(self._processes) or any(not pool._closed for pool in self._pools)

    def attach_pool(self, *, timeout: float = 5.0) -> WorkerPool:
        """Exclusively attach to this generation's detached persistent pool."""
        with self._manager._manager_work():
            if type(timeout) not in {int, float} or not math.isfinite(timeout) or timeout <= 0:
                raise ValueError("timeout must be a positive finite number")
            with environment_lifecycle_gate(self._manager, self.name):
                self._require_current_generation()
                runtime_state.reconcile_persistent_pool(
                    self._manager.root,
                    self.name,
                    grace=self._manager.termination_grace,
                )
                entries = runtime_state.live_workers_for_env(
                    self._manager.root,
                    self.name,
                    expected_identity={
                        "env_path": str(self.path),
                        "generation_id": self.generation_id,
                        "recipe_hash": self.recipe_hash,
                        "worker_runtime_version": WORKER_RUNTIME_VERSION,
                        "protocol_version": EXECUTION_PROTOCOL_VERSION,
                    },
                )
                if not entries:
                    raise RuntimeError(f"No detached persistent pool exists for environment {self.name!r}")
                runtime = ExternalEnvironment(
                    self.name,
                    self.pixi_manifest_path,
                    self._manager,
                    expected_generation_id=self.generation_id,
                    expected_recipe_hash=self.recipe_hash,
                )
                pool = WorkerPool(self, runtime)
                authkey = runtime_state.load_or_create_root_authkey(self._manager.root)
                runtime.attach_workers(entries, authkey, timeout=timeout)
            with self._lock:
                self._pools.append(pool)
            return pool

    def _require_current_generation(self) -> None:
        ready = _read_ready(self.path)
        actual_generation_id = str(ready.get("generation_id")) if ready is not None else None
        actual_recipe_hash = str(ready.get("recipe_hash")) if ready is not None else None
        if actual_generation_id != self.generation_id or actual_recipe_hash != self.recipe_hash:
            raise EnvironmentGenerationChangedError(
                self.name,
                expected_generation_id=self.generation_id,
                expected_recipe_hash=self.recipe_hash,
                actual_generation_id=actual_generation_id,
                actual_recipe_hash=actual_recipe_hash,
            )

name property

Return the managed environment name.

path property

Return the canonical Pixi project directory.

pixi_manifest_path property

Return the generated pixi.toml path.

pixi_lock_path property

Return the resolved or supplied pixi.lock path.

pixi_version property

Return the Pixi version recorded when this generation was provisioned.

pixi_executable_path property

Return the Pixi executable used to provision this generation.

generation_id property

Return the unique identifier for this published environment generation.

recipe_hash property

Return the hash of the complete normalized environment recipe.

lockfile_hash property

Return the SHA-256 hash of this generation's lockfile.

spawn(argv, *, cwd=None, env=None, output_limit=1048576)

Launch an independently supervised command in this generation.

Source code in src/wetlands/managed_environment.py

def spawn(
    self,
    argv: Sequence[str],
    *,
    cwd: str | Path | None = None,
    env: Mapping[str, str | None] | None = None,
    output_limit: int = 1_048_576,
) -> ManagedProcess:
    """Launch an independently supervised command in this generation."""
    from wetlands.managed_process import ManagedProcess, _validate_launch_options

    options = _validate_launch_options(
        argv=argv,
        cwd=cwd,
        env=env,
        output_limit=output_limit,
        default_cwd=self.path,
    )
    with self._manager._manager_work():
        with environment_lifecycle_gate(self._manager, self.name):
            self._require_current_generation()
            runtime_state.reconcile_persistent_pool(
                self._manager.root,
                self.name,
                grace=self._manager.termination_grace,
            )
            return ManagedProcess._launch_validated(
                environment=self,
                options=options,
            )

run(argv, *, cwd=None, env=None, timeout=None, output_limit=1048576, check=True)

Run a command to completion and return its Wetlands-owned result.

Source code in src/wetlands/managed_environment.py

def run(
    self,
    argv: Sequence[str],
    *,
    cwd: str | Path | None = None,
    env: Mapping[str, str | None] | None = None,
    timeout: float | None = None,
    output_limit: int = 1_048_576,
    check: bool = True,
) -> ManagedProcessResult:
    """Run a command to completion and return its Wetlands-owned result."""
    from wetlands.managed_process import _validate_check, _validate_timeout

    normalized_timeout = _validate_timeout(timeout)
    normalized_check = _validate_check(check)
    process = self.spawn(argv, cwd=cwd, env=env, output_limit=output_limit)
    try:
        return process.wait(timeout=normalized_timeout, check=normalized_check)
    finally:
        process.close()

start(*, workers=1, persistent=False, worker_environment=None, worker_timeout=None)

Start a new warm worker pool for this environment generation.

Parameters:

Name	Type	Description	Default
workers	int	Number of worker processes in the pool.	1
persistent	bool	Keep workers alive when the controller deliberately detaches.	False
worker_environment	Callable[[int], Mapping[str, str]] | None	Optional callable receiving each zero-based worker index and returning environment variables for that worker. Wetlands snapshots the mappings before launch and reuses the mapping for the same index when replacing a worker. This cannot be combined with persistent=True.	None
worker_timeout	float | None	Optional worker inactivity timeout in seconds. Each IPC message resets the timer, so this is a health check rather than a maximum task execution time.	None

Source code in src/wetlands/managed_environment.py

def start(
    self,
    *,
    workers: int = 1,
    persistent: bool = False,
    worker_environment: Callable[[int], Mapping[str, str]] | None = None,
    worker_timeout: float | None = None,
) -> WorkerPool:
    """Start a new warm worker pool for this environment generation.

    Args:
        workers: Number of worker processes in the pool.
        persistent: Keep workers alive when the controller deliberately detaches.
        worker_environment: Optional callable receiving each zero-based worker
            index and returning environment variables for that worker. Wetlands
            snapshots the mappings before launch and reuses the mapping for the
            same index when replacing a worker. This cannot be combined with
            ``persistent=True``.
        worker_timeout: Optional worker inactivity timeout in seconds.
            Each IPC message resets the timer, so this is a health check rather
            than a maximum task execution time.
    """
    with self._manager._manager_work():
        if workers < 1:
            raise ValueError("workers must be at least one")
        if persistent and worker_environment is not None:
            raise ValueError("worker_environment cannot be combined with persistent=True")
        worker_environments = _validate_worker_environments(workers, worker_environment)
        snapshotted_worker_environment = worker_environments.__getitem__ if worker_environment is not None else None
        with self._lock:
            existing_pools = tuple(self._pools)
        for existing_pool in existing_pools:
            if not existing_pool._closed:
                existing_pool._runtime._raise_if_failed()
        with environment_lifecycle_gate(self._manager, self.name):
            self._require_current_generation()
            runtime_state.reconcile_persistent_pool(
                self._manager.root,
                self.name,
                grace=self._manager.termination_grace,
            )
            runtime = ExternalEnvironment(
                self.name,
                self.pixi_manifest_path,
                self._manager,
                expected_generation_id=self.generation_id,
                expected_recipe_hash=self.recipe_hash,
            )
            pool = WorkerPool(self, runtime)
            with self._lock:
                self._pools.append(pool)
            try:
                runtime.launch(
                    max_workers=workers,
                    persistent=persistent,
                    worker_environment=snapshotted_worker_environment,
                    worker_timeout=worker_timeout,
                )
            except BaseException:
                if not runtime._workers:
                    with self._lock:
                        self._pools.remove(pool)
                    pool._closed = True
                raise
        return pool

close_pools()

Close every worker pool started through this environment handle.

Source code in src/wetlands/managed_environment.py

def close_pools(self) -> None:
    """Close every worker pool started through this environment handle."""
    errors = self._close_pools()
    if errors:
        raise ManagerCloseError(errors)

attach_pool(*, timeout=5.0)

Exclusively attach to this generation's detached persistent pool.

Source code in src/wetlands/managed_environment.py

def attach_pool(self, *, timeout: float = 5.0) -> WorkerPool:
    """Exclusively attach to this generation's detached persistent pool."""
    with self._manager._manager_work():
        if type(timeout) not in {int, float} or not math.isfinite(timeout) or timeout <= 0:
            raise ValueError("timeout must be a positive finite number")
        with environment_lifecycle_gate(self._manager, self.name):
            self._require_current_generation()
            runtime_state.reconcile_persistent_pool(
                self._manager.root,
                self.name,
                grace=self._manager.termination_grace,
            )
            entries = runtime_state.live_workers_for_env(
                self._manager.root,
                self.name,
                expected_identity={
                    "env_path": str(self.path),
                    "generation_id": self.generation_id,
                    "recipe_hash": self.recipe_hash,
                    "worker_runtime_version": WORKER_RUNTIME_VERSION,
                    "protocol_version": EXECUTION_PROTOCOL_VERSION,
                },
            )
            if not entries:
                raise RuntimeError(f"No detached persistent pool exists for environment {self.name!r}")
            runtime = ExternalEnvironment(
                self.name,
                self.pixi_manifest_path,
                self._manager,
                expected_generation_id=self.generation_id,
                expected_recipe_hash=self.recipe_hash,
            )
            pool = WorkerPool(self, runtime)
            authkey = runtime_state.load_or_create_root_authkey(self._manager.root)
            runtime.attach_workers(entries, authkey, timeout=timeout)
        with self._lock:
            self._pools.append(pool)
        return pool

ManagedEnvironmentInfo dataclass

A side-effect-free snapshot of one environment managed by a root.

Source code in src/wetlands/environment_info.py

@dataclass(frozen=True)
class ManagedEnvironmentInfo:
    """A side-effect-free snapshot of one environment managed by a root."""

    name: str
    path: Path
    state: ManagedEnvironmentState
    generation_id: str | None = None
    recipe_hash: str | None = None
    pixi_version: str | None = None

    @property
    def ready(self) -> bool:
        """Return whether complete managed metadata has been published."""

        return self.state is ManagedEnvironmentState.READY

ready property

Return whether complete managed metadata has been published.

ManagedEnvironmentState

Bases: Enum

The publication state of a discovered managed environment.

Source code in src/wetlands/environment_info.py

class ManagedEnvironmentState(enum.Enum):
    """The publication state of a discovered managed environment."""

    READY = "ready"
    INCOMPLETE = "incomplete"

ManagedProcess

A supervised command and its owned process tree.

Instances are returned by :meth:ManagedEnvironment.spawn; applications do not construct them directly.

Source code in src/wetlands/managed_process.py

class ManagedProcess:
    """A supervised command and its owned process tree.

    Instances are returned by :meth:`ManagedEnvironment.spawn`; applications do
    not construct them directly.
    """

    def __init__(
        self,
        *,
        environment: ManagedEnvironment,
        argv: tuple[str, ...],
        process: subprocess.Popen[bytes],
        output_limit: int,
        started_at: float,
    ) -> None:
        self._environment_handle = environment
        self._environment = environment.name
        self._generation_id = environment.generation_id
        self._argv = argv
        self._process = process
        self._output_limit = output_limit
        self._started_at = started_at
        self._identity: ProcessIdentity | None = None

        self._condition = threading.Condition(threading.RLock())
        self._done = threading.Event()
        self._cleanup_lock = threading.Lock()
        self._cause: tuple[str, float | None] | None = None
        self._result: ManagedProcessResult | None = None
        self._terminal_error: ProcessError | None = None
        self._tree_clean = False
        self._ownership_clean = False
        self._registered = False
        self._released = False

        self._output_lock = threading.Lock()
        self._captured = 0
        self._stdout = bytearray()
        self._stderr = bytearray()
        self._truncated_streams: set[OutputStream] = set()

        self._event_condition = threading.Condition(threading.RLock())
        self._events: deque[OutputEvent] = deque(maxlen=_EVENT_CAPACITY)
        self._next_sequence = 0
        self._streams_closed = False
        self._readers: list[threading.Thread] = []
        self._reader_errors: list[BaseException] = []
        self._supervisor: threading.Thread | None = None

    _validate_launch_options = staticmethod(_validate_launch_options)

    @classmethod
    def _launch(
        cls,
        *,
        environment: ManagedEnvironment,
        argv: Sequence[str],
        cwd: str | Path | None = None,
        env: Mapping[str, str | None] | None = None,
        output_limit: int = 1_048_576,
    ) -> ManagedProcess:
        options = _validate_launch_options(
            argv=argv,
            cwd=cwd,
            env=env,
            output_limit=output_limit,
            default_cwd=environment.path,
        )
        return cls._launch_validated(environment=environment, options=options)

    @classmethod
    def _launch_validated(cls, *, environment: ManagedEnvironment, options: _LaunchOptions) -> ManagedProcess:
        command = [
            str(environment.pixi_executable_path),
            "run",
            "--manifest-path",
            str(environment.pixi_manifest_path),
            "--locked",
            "--",
            *options.argv,
        ]
        launch_env = dict(os.environ)
        launch_env.update(
            {
                "PIXI_HOME": str(environment._manager.state_root / "pixi-home"),
                "PIXI_CACHE_DIR": str(environment._manager.state_root / "pixi-cache"),
                **{
                    ("NO_PROXY" if scheme == "no_proxy" else f"{scheme.upper()}_PROXY"): value
                    for scheme, value in (environment._manager.network or {}).items()
                },
            }
        )
        if os.name == "nt":
            launch_env.update(_windows_git_long_paths_overrides(os.environ))
        cls._apply_environment_overlay(launch_env, options.env_overlay)

        popen_options: dict[str, Any] = {
            "cwd": options.cwd,
            "env": launch_env,
            "stdin": subprocess.DEVNULL,
            "stdout": subprocess.PIPE,
            "stderr": subprocess.PIPE,
            "text": False,
        }
        if os.name == "nt":
            popen_options["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP") | getattr(
                subprocess, "CREATE_SUSPENDED", 0x00000004
            )
        else:
            popen_options["start_new_session"] = True

        started_at = time.time()
        try:
            process = cast(subprocess.Popen[bytes], subprocess.Popen(command, **popen_options))
        except OSError as error:
            message = (
                f"Could not launch command {options.argv!r} in environment {environment.name!r} "
                f"generation {environment.generation_id!r} through Pixi: {error}"
            )
            try:
                contextual_error = type(error)(error.errno, message, error.filename)
            except (TypeError, ValueError):
                contextual_error = OSError(error.errno, message, error.filename)
            raise contextual_error from error
        handle = cls(
            environment=environment,
            argv=options.argv,
            process=process,
            output_limit=options.output_limit,
            started_at=started_at,
        )
        if os.name == "nt":
            process._wetlands_suspended = True  # type: ignore[attr-defined]
        try:
            handle._registered = True
            environment._register_process(handle)
            identity = capture_process_identity(process.pid)
            handle._identity = identity
            if os.name != "nt" and (identity.process_group_id != process.pid or identity.session_id != process.pid):
                raise ProcessIdentityError(
                    f"Managed command PID {process.pid} has no proven isolated process-session ownership"
                )
            process._wetlands_started_at = identity.started_at  # type: ignore[attr-defined]
            process._wetlands_process_group_id = identity.process_group_id  # type: ignore[attr-defined]
            process._wetlands_session_id = identity.session_id  # type: ignore[attr-defined]
            if os.name == "nt":
                _assign_windows_kill_job(process)
            handle._start_readers()
            if os.name == "nt":
                handle._resume_windows_process()
                process._wetlands_suspended = False  # type: ignore[attr-defined]
            handle._start_supervisor()
        except BaseException as launch_error:
            cleanup_errors = handle._cleanup_failed_launch()
            if cleanup_errors:
                cleanup_error = handle._freeze_cleanup_error(cleanup_errors)
                raise cleanup_error from launch_error
            raise
        return handle

    @staticmethod
    def _apply_environment_overlay(target: dict[str, str], overlay: Mapping[str, str | None]) -> None:
        for key, value in overlay.items():
            actual_key = key
            if os.name == "nt":
                inherited = next((candidate for candidate in target if candidate.casefold() == key.casefold()), None)
                if inherited is not None:
                    actual_key = inherited
            if value is None:
                target.pop(actual_key, None)
            else:
                if actual_key != key:
                    target.pop(actual_key, None)
                target[key] = value

    @property
    def argv(self) -> tuple[str, ...]:
        return self._argv

    @property
    def environment(self) -> str:
        return self._environment

    @property
    def generation_id(self) -> str:
        return self._generation_id

    @property
    def pid(self) -> int:
        return self._process.pid

    @property
    def returncode(self) -> int | None:
        return self._process.poll()

    @property
    def running(self) -> bool:
        return self.returncode is None

    def _start_readers(self) -> None:
        assert self._process.stdout is not None
        assert self._process.stderr is not None
        stdout_reader = threading.Thread(
            target=self._drain,
            args=(self._process.stdout, OutputStream.STDOUT),
            name=f"wetlands-process-{self.pid}-stdout",
            daemon=True,
        )
        stderr_reader = threading.Thread(
            target=self._drain,
            args=(self._process.stderr, OutputStream.STDERR),
            name=f"wetlands-process-{self.pid}-stderr",
            daemon=True,
        )
        for reader, pipe in (
            (stdout_reader, self._process.stdout),
            (stderr_reader, self._process.stderr),
        ):
            try:
                reader.start()
            except BaseException:
                pipe.close()
                raise
            self._readers.append(reader)

    def _resume_windows_process(self) -> None:
        """Resume a Windows child created suspended after mandatory Job assignment."""
        import ctypes
        from ctypes import wintypes

        thread_ids = [thread.id for thread in psutil.Process(self.pid).threads()]
        if not thread_ids:
            raise OSError(f"Could not find the suspended primary thread for PID {self.pid}")
        kernel32 = getattr(ctypes, "WinDLL")("kernel32", use_last_error=True)
        kernel32.OpenThread.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD)
        kernel32.OpenThread.restype = wintypes.HANDLE
        kernel32.ResumeThread.argtypes = (wintypes.HANDLE,)
        kernel32.ResumeThread.restype = wintypes.DWORD
        kernel32.CloseHandle.argtypes = (wintypes.HANDLE,)
        kernel32.CloseHandle.restype = wintypes.BOOL
        for thread_id in thread_ids:
            handle = kernel32.OpenThread(0x0002, False, wintypes.DWORD(thread_id))
            if not handle:
                error = getattr(ctypes, "get_last_error")()
                raise OSError(error, getattr(ctypes, "FormatError")(error))
            try:
                previous_count = kernel32.ResumeThread(handle)
                if previous_count == 0xFFFFFFFF:
                    error = getattr(ctypes, "get_last_error")()
                    raise OSError(error, getattr(ctypes, "FormatError")(error))
                if previous_count == 0:
                    raise ProcessIdentityError(
                        f"Suspended process thread {thread_id} for PID {self.pid} was already running"
                    )
            finally:
                kernel32.CloseHandle(handle)

    def _start_supervisor(self) -> None:
        supervisor = threading.Thread(
            target=self._supervise,
            name=f"wetlands-process-{self.pid}-supervisor",
            daemon=True,
        )
        self._supervisor = supervisor
        supervisor.start()

    def _drain(self, pipe: Any, stream: OutputStream) -> None:
        decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
        pending = ""
        try:
            while True:
                chunk = pipe.read1(_READ_SIZE) if hasattr(pipe, "read1") else pipe.read(_READ_SIZE)
                if not chunk:
                    break
                accepted = b""
                breach = False
                with self._output_lock:
                    if not self._truncated_streams:
                        available = self._output_limit - self._captured
                        accepted = chunk[:available]
                        self._captured += len(accepted)
                        target = self._stdout if stream is OutputStream.STDOUT else self._stderr
                        target.extend(accepted)
                        breach = len(chunk) > len(accepted)
                        if breach:
                            self._truncated_streams.add(stream)
                    elif chunk:
                        self._truncated_streams.add(stream)
                if accepted:
                    pending += decoder.decode(accepted, final=False)
                    pending = self._publish_complete_lines(stream, pending)
                if breach:
                    self._request("output_limit", None)
            pending += decoder.decode(b"", final=True)
            if pending:
                self._publish_event(stream, pending)
        except BaseException as error:
            with self._condition:
                self._reader_errors.append(error)
            self._request("cleanup", None)
        finally:
            try:
                pipe.close()
            except OSError:
                pass

    def _publish_complete_lines(self, stream: OutputStream, pending: str) -> str:
        while True:
            newline = pending.find("\n")
            if newline < 0:
                return pending
            end = newline + 1
            self._publish_event(stream, pending[:end])
            pending = pending[end:]

    def _publish_event(self, stream: OutputStream, text: str) -> None:
        with self._event_condition:
            event = OutputEvent(self._next_sequence, time.time(), stream, text)
            self._next_sequence += 1
            self._events.append(event)
            self._event_condition.notify_all()

    def _request(self, cause: str, value: float | None) -> None:
        with self._condition:
            if self._cause is None:
                self._cause = (cause, value)
            self._condition.notify_all()

    def _supervise(self) -> None:
        cleanup_errors: list[BaseException] = []
        cause: tuple[str, float | None] | None
        try:
            while True:
                with self._condition:
                    cause = self._cause
                if cause is not None or self._process.poll() is not None:
                    break
                with self._condition:
                    self._condition.wait(0.02)

            grace = self._environment_handle._manager.termination_grace
            if cause is not None and cause[0] == "kill":
                try:
                    self._kill_tree()
                except BaseException as error:
                    cleanup_errors.append(error)
            elif cause is not None and cause[0] == "terminate" and cause[1] is not None:
                grace = cause[1]
            if cause is None or cause[0] != "kill":
                try:
                    self._terminate_tree(grace)
                except BaseException as error:
                    cleanup_errors.append(error)
            self._join_readers(cleanup_errors)
            with self._condition:
                cleanup_errors.extend(self._reader_errors)
        except BaseException as error:
            cleanup_errors.append(error)

        result = self._make_result()
        if not cleanup_errors:
            try:
                self._release_once()
            except BaseException as error:
                cleanup_errors.append(error)

        with self._condition:
            cause = self._cause
            if cleanup_errors:
                self._terminal_error = self._new_cleanup_error(cleanup_errors, result)
            elif self._terminal_error is None and cause is not None and cause[0] == "output_limit":
                self._terminal_error = ProcessOutputLimitError(
                    self._output_limit,
                    result,
                    frozenset(self._truncated_streams),
                    environment=self.environment,
                    generation_id=self.generation_id,
                )
            elif self._terminal_error is None and cause is not None and cause[0] == "timeout":
                assert cause[1] is not None
                self._terminal_error = ProcessTimeoutError(
                    cause[1],
                    result,
                    environment=self.environment,
                    generation_id=self.generation_id,
                )
            self._result = result
            self._ownership_clean = not cleanup_errors
            self._done.set()
            self._condition.notify_all()
        with self._event_condition:
            self._streams_closed = True
            self._event_condition.notify_all()

    def _terminate_tree(self, grace: float) -> None:
        if os.name == "nt":
            self._terminate_windows_job(grace)
        else:
            terminate_launched_process_tree(
                self._process,
                grace=grace,
                close_windows_job=_close_windows_job,
            )
        with self._condition:
            self._tree_clean = True

    def _kill_tree(self) -> None:
        if os.name == "nt":
            self._terminate_windows_job(0.0, force=True)
            with self._condition:
                self._tree_clean = True
            return
        if self._identity is None:
            raise ProcessIdentityError(f"Managed command PID {self.pid} has no recorded process identity")
        process_group_id = self._identity.process_group_id
        session_id = self._identity.session_id
        if process_group_id != self.pid or session_id != self.pid:
            raise ProcessIdentityError(
                f"Managed command PID {self.pid} has no proven isolated process-session ownership"
            )
        assert process_group_id is not None
        if self._process.poll() is None:
            if not identity_matches(self.pid, self._identity.started_at):
                raise ProcessIdentityError(f"Refusing to signal PID {self.pid}: its process start identity changed")
            if os.getpgid(self.pid) != process_group_id or os.getsid(self.pid) != session_id:
                raise ProcessIdentityError(f"Refusing to signal PID {self.pid}: its process-session identity changed")
        try:
            os.killpg(process_group_id, signal.SIGKILL)
        except ProcessLookupError:
            with contextlib.suppress(subprocess.TimeoutExpired, OSError):
                self._process.wait(timeout=0)
            with self._condition:
                self._tree_clean = True
            return
        verification_timeout = max(1.0, self._environment_handle._manager.termination_grace)
        if _wait_for_posix_group_exit(
            process_group_id,
            process=self._process,
            timeout=verification_timeout,
        ):
            with self._condition:
                self._tree_clean = True
            return
        survivors = _posix_group_members(process_group_id)
        details = f"; surviving PIDs: {survivors}" if survivors else ""
        raise ProcessTerminationError(f"Managed command process group {process_group_id} survived SIGKILL{details}")

    def _terminate_windows_job(self, grace: float, *, force: bool = False) -> None:
        """Gracefully signal, then terminate and verify the mandatory Windows Job."""
        handle = getattr(self._process, "_wetlands_job_handle", None)
        if handle is None:
            raise ProcessIdentityError(f"Managed command PID {self.pid} has no assigned Windows Job Object")
        if self._identity is None:
            raise ProcessIdentityError(f"Managed command PID {self.pid} has no recorded process identity")
        leader_running = self._process.poll() is None
        suspended = bool(getattr(self._process, "_wetlands_suspended", False))
        if leader_running and not identity_matches(self.pid, self._identity.started_at):
            raise ProcessIdentityError(f"Refusing to signal PID {self.pid}: its process start identity changed")

        deadline = time.monotonic() + max(0.0, grace)
        if leader_running and not suspended and not force:
            try:
                self._process.send_signal(getattr(signal, "CTRL_BREAK_EVENT"))
            except OSError:
                pass
            while self._windows_job_active_processes(handle) and time.monotonic() < deadline:
                time.sleep(min(0.02, max(0.0, deadline - time.monotonic())))

        if self._windows_job_active_processes(handle):
            self._terminate_windows_job_object(handle)
            force_deadline = time.monotonic() + max(1.0, grace)
            while self._windows_job_active_processes(handle) and time.monotonic() < force_deadline:
                time.sleep(min(0.02, max(0.0, force_deadline - time.monotonic())))
        survivors = self._windows_job_active_processes(handle)
        if survivors:
            raise ProcessTerminationError(
                f"Managed command Windows Job for PID {self.pid} retained {survivors} active processes after termination"
            )
        try:
            self._process.wait(timeout=max(1.0, grace))
        except (subprocess.TimeoutExpired, OSError) as error:
            raise ProcessTerminationError(f"Could not reap managed command PID {self.pid}") from error
        _close_windows_job(self._process)

    @staticmethod
    def _windows_job_active_processes(handle: Any) -> int:
        import ctypes
        from ctypes import wintypes

        class JOBOBJECT_BASIC_ACCOUNTING_INFORMATION(ctypes.Structure):
            _fields_ = [
                ("TotalUserTime", ctypes.c_longlong),
                ("TotalKernelTime", ctypes.c_longlong),
                ("ThisPeriodTotalUserTime", ctypes.c_longlong),
                ("ThisPeriodTotalKernelTime", ctypes.c_longlong),
                ("TotalPageFaultCount", wintypes.DWORD),
                ("TotalProcesses", wintypes.DWORD),
                ("ActiveProcesses", wintypes.DWORD),
                ("TotalTerminatedProcesses", wintypes.DWORD),
            ]

        kernel32 = getattr(ctypes, "WinDLL")("kernel32", use_last_error=True)
        kernel32.QueryInformationJobObject.argtypes = (
            wintypes.HANDLE,
            ctypes.c_int,
            ctypes.c_void_p,
            wintypes.DWORD,
            ctypes.POINTER(wintypes.DWORD),
        )
        kernel32.QueryInformationJobObject.restype = wintypes.BOOL
        information = JOBOBJECT_BASIC_ACCOUNTING_INFORMATION()
        returned = wintypes.DWORD()
        queried = kernel32.QueryInformationJobObject(
            handle,
            1,
            ctypes.byref(information),
            ctypes.sizeof(information),
            ctypes.byref(returned),
        )
        if not queried:
            error = getattr(ctypes, "get_last_error")()
            raise OSError(error, getattr(ctypes, "FormatError")(error))
        return int(information.ActiveProcesses)

    @staticmethod
    def _terminate_windows_job_object(handle: Any) -> None:
        import ctypes

        kernel32 = getattr(ctypes, "WinDLL")("kernel32", use_last_error=True)
        kernel32.TerminateJobObject.argtypes = (ctypes.c_void_p, ctypes.c_uint)
        kernel32.TerminateJobObject.restype = ctypes.c_int
        if not kernel32.TerminateJobObject(handle, 1):
            error = getattr(ctypes, "get_last_error")()
            raise OSError(error, getattr(ctypes, "FormatError")(error))

    def _join_readers(self, errors: list[BaseException]) -> None:
        timeout = max(1.0, self._environment_handle._manager.termination_grace * 2)
        for reader in self._readers:
            reader.join(timeout)
            if reader.is_alive():
                errors.append(RuntimeError(f"Output reader {reader.name} did not finish"))

    def _make_result(self) -> ManagedProcessResult:
        returncode = self._process.poll()
        if returncode is None:
            returncode = -1
        with self._output_lock:
            stdout = bytes(self._stdout).decode("utf-8", errors="replace")
            stderr = bytes(self._stderr).decode("utf-8", errors="replace")
        return ManagedProcessResult(
            argv=self.argv,
            returncode=returncode,
            stdout=stdout,
            stderr=stderr,
            started_at=self._started_at,
            ended_at=max(self._started_at, time.time()),
        )

    def _new_cleanup_error(
        self,
        failures: Sequence[BaseException],
        result: ManagedProcessResult | None,
    ) -> ProcessCleanupError:
        initiating_error: ProcessError | None = None
        with self._condition:
            cause = self._cause
        if result is not None and cause is not None and cause[0] == "timeout" and cause[1] is not None:
            initiating_error = ProcessTimeoutError(
                cause[1],
                result,
                environment=self.environment,
                generation_id=self.generation_id,
            )
        return ProcessCleanupError(
            failures,
            result,
            argv=self.argv,
            environment=self.environment,
            generation_id=self.generation_id,
            initiating_error=initiating_error,
        )

    def _freeze_cleanup_error(self, failures: Sequence[BaseException]) -> ProcessCleanupError:
        result = self._make_result()
        error = self._new_cleanup_error(failures, result)
        with self._condition:
            self._terminal_error = error
            self._result = result
            self._done.set()
            self._condition.notify_all()
        with self._event_condition:
            self._streams_closed = True
            self._event_condition.notify_all()
        return error

    def _cleanup_failed_launch(self) -> tuple[BaseException, ...]:
        errors: list[BaseException] = []
        try:
            self._terminate_tree(self._environment_handle._manager.termination_grace)
        except BaseException as primary_error:
            try:
                self._terminate_uninitialized_process()
            except BaseException as fallback_error:
                errors.extend((primary_error, fallback_error))
        self._join_readers(errors)
        with self._condition:
            errors.extend(self._reader_errors)
        if not errors:
            try:
                self._release_once()
            except BaseException as error:
                errors.append(error)
        self._ownership_clean = not errors
        return tuple(errors)

    def _terminate_uninitialized_process(self) -> None:
        """Best-effort containment cleanup when post-Popen identity setup failed."""
        grace = self._environment_handle._manager.termination_grace
        if os.name != "nt":
            _terminate_posix_group(self.pid, grace=grace, process=self._process)
        else:
            try:
                _close_windows_job(self._process)
            finally:
                if self._process.poll() is None:
                    self._process.kill()
                self._process.wait(timeout=max(1.0, grace))
        with self._condition:
            self._tree_clean = True

    def _release_once(self) -> None:
        with self._condition:
            if self._released or not self._registered:
                return
            self._environment_handle._release_process(self)
            self._released = True

    def _retry_cleanup(self) -> None:
        with self._cleanup_lock:
            if self._ownership_clean:
                return
            failures: list[BaseException] = []
            with self._condition:
                tree_clean = self._tree_clean
            if not tree_clean:
                try:
                    self._terminate_tree(self._environment_handle._manager.termination_grace)
                except BaseException as error:
                    failures.append(error)
            self._join_readers(failures)
            if not failures:
                try:
                    self._release_once()
                except BaseException as error:
                    failures.append(error)
            if failures:
                raise self._terminal_error or self._new_cleanup_error(failures, self._make_result())
            self._ownership_clean = True

    def wait(self, timeout: float | None = None, *, check: bool = True) -> ManagedProcessResult:
        normalized_timeout = _validate_timeout(timeout)
        normalized_check = _validate_check(check)
        if normalized_timeout is None:
            self._done.wait()
        elif not self._done.wait(normalized_timeout):
            with self._condition:
                if self._process.poll() is None and not self._done.is_set():
                    self._request("timeout", normalized_timeout)
            self._done.wait()
        return self._outcome(normalized_check)

    async def wait_async(self, timeout: float | None = None, *, check: bool = True) -> ManagedProcessResult:
        normalized_timeout = _validate_timeout(timeout)
        normalized_check = _validate_check(check)
        try:
            return await _run_blocking(self.wait, normalized_timeout, check=normalized_check)
        except asyncio.CancelledError:
            cleanup = asyncio.ensure_future(_run_blocking(self.close))
            while not cleanup.done():
                try:
                    await asyncio.shield(cleanup)
                except asyncio.CancelledError:
                    continue
            try:
                cleanup.result()
            except BaseException:
                pass
            raise

    def __await__(self) -> Any:
        return self.wait_async().__await__()

    def _outcome(self, check: bool) -> ManagedProcessResult:
        assert self._result is not None
        if self._terminal_error is not None:
            raise self._terminal_error
        if check and self._result.returncode != 0:
            raise ProcessExitError(
                self._result,
                environment=self.environment,
                generation_id=self.generation_id,
            )
        return self._result

    async def events(self, *, replay: bool = True) -> AsyncIterator[OutputEvent]:
        if not isinstance(replay, bool):
            raise TypeError("replay must be a bool")
        with self._event_condition:
            cursor = self._events[0].sequence if replay and self._events else self._next_sequence
        while True:
            item = await _run_blocking(self._next_event, cursor, 0.1)
            if item is None:
                with self._event_condition:
                    if self._streams_closed:
                        return
                continue
            event, cursor = item
            yield event

    def wait_for_line(
        self,
        predicate: Callable[[OutputEvent], bool],
        timeout: float | None = None,
        *,
        replay: bool = True,
    ) -> OutputEvent:
        if not callable(predicate):
            raise TypeError("predicate must be callable")
        normalized_timeout = _validate_timeout(timeout)
        if not isinstance(replay, bool):
            raise TypeError("replay must be a bool")
        deadline = None if normalized_timeout is None else time.monotonic() + normalized_timeout
        with self._event_condition:
            cursor = self._events[0].sequence if replay and self._events else self._next_sequence
        while True:
            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
            item = self._next_event(cursor, remaining)
            if item is None:
                if self._streams_closed:
                    raise EOFError(f"Command {self.argv!r} closed its output without a matching event")
                assert normalized_timeout is not None
                raise ProcessLineTimeoutError(
                    normalized_timeout,
                    argv=self.argv,
                    environment=self.environment,
                    generation_id=self.generation_id,
                )
            event, cursor = item
            if predicate(event):
                return event

    def _next_event(self, cursor: int, timeout: float | None) -> tuple[OutputEvent, int] | None:
        deadline = None if timeout is None else time.monotonic() + timeout
        with self._event_condition:
            while True:
                oldest = self._events[0].sequence if self._events else self._next_sequence
                if cursor < oldest:
                    raise ProcessEventLagError(
                        cursor,
                        oldest,
                        argv=self.argv,
                        environment=self.environment,
                        generation_id=self.generation_id,
                    )
                if cursor < self._next_sequence:
                    event = self._events[cursor - oldest]
                    return event, cursor + 1
                if self._streams_closed:
                    return None
                if deadline is not None:
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        return None
                else:
                    remaining = None
                self._event_condition.wait(remaining)

    def terminate(self, timeout: float | None = None) -> None:
        grace = _validate_timeout(timeout)
        self._request("terminate", grace)
        self._done.wait()
        if isinstance(self._terminal_error, ProcessCleanupError):
            raise self._terminal_error

    def kill(self) -> None:
        self._request("kill", 0.0)
        self._done.wait()
        if isinstance(self._terminal_error, ProcessCleanupError):
            raise self._terminal_error

    def close(self) -> None:
        if not self._done.is_set():
            self._request("close", None)
            self._done.wait()
        if not self._ownership_clean:
            self._retry_cleanup()

    def __enter__(self) -> ManagedProcess:
        return self

    def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> Literal[False]:
        self.close()
        return False

ManagedProcessResult dataclass

The immutable terminal result of a managed command.

Source code in src/wetlands/managed_process.py

@dataclass(frozen=True)
class ManagedProcessResult:
    """The immutable terminal result of a managed command."""

    argv: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str
    started_at: float
    ended_at: float

OutputEvent dataclass

One decoded line or trailing partial line from a managed command.

Source code in src/wetlands/managed_process.py

@dataclass(frozen=True)
class OutputEvent:
    """One decoded line or trailing partial line from a managed command."""

    sequence: int
    timestamp: float
    stream: OutputStream
    text: str

OutputStream

Bases: str, Enum

The output pipe that produced an event.

Source code in src/wetlands/managed_process.py

class OutputStream(str, Enum):
    """The output pipe that produced an event."""

    STDOUT = "stdout"
    STDERR = "stderr"

PostInstallCommand dataclass

A command run after Pixi installs the environment dependencies.

Source code in src/wetlands/specs.py

@dataclass(frozen=True)
class PostInstallCommand:
    """A command run after Pixi installs the environment dependencies."""

    argv: tuple[str, ...]
    shell: bool = False
    display: str | None = None

    def __post_init__(self) -> None:
        if isinstance(self.argv, str):
            raise TypeError("Post-install argv must be a sequence of arguments, not a string")
        argv = tuple(str(item) for item in self.argv)
        if not argv:
            raise ValueError("Post-install command argv cannot be empty")
        if self.shell and self.display is None:
            raise ValueError("Shell post-install commands require an explicit safe display string")
        object.__setattr__(self, "argv", argv)

PixiInfo dataclass

Information about the validated Pixi executable used by Wetlands.

Source code in src/wetlands/specs.py

@dataclass(frozen=True)
class PixiInfo:
    """Information about the validated Pixi executable used by Wetlands."""

    executable: Path
    version: str
    managed: bool

ProvisioningStage

Bases: Enum

A stable provisioning stage identifier used by operation events.

Source code in src/wetlands/specs.py

class ProvisioningStage(enum.Enum):
    """A stable provisioning stage identifier used by operation events."""

    LOCK_WAIT = "lock_wait"
    PIXI_DISCOVERY = "pixi_discovery"
    PIXI_DOWNLOAD = "pixi_download"
    PIXI_VERIFY = "pixi_verify"
    PIXI_INSTALL = "pixi_install"
    TARGET_INSPECTION = "target_inspection"
    INCOMPLETE_REMOVAL = "incomplete_removal"
    PROJECT_MATERIALIZATION = "project_materialization"
    LOCK_RESOLUTION = "lock_resolution"
    CONDA_INSTALL = "conda_install"
    PYPI_INSTALL = "pypi_install"
    LOCAL_INSTALL = "local_install"
    POST_INSTALL = "post_install"
    VALIDATION = "validation"
    METADATA_PUBLICATION = "metadata_publication"
    CLEANUP = "cleanup"

WorkerPool

A group of warm worker processes for one managed environment generation.

Methods:

Name	Description
submit_import	Submit an installed module:qualified.callable target for execution.
submit_path	Submit a callable from an explicit local source path.
execute_import	Execute an installed target and block until it finishes.
execute_path	Execute a local path target and block until it finishes.
detach	Release control of a persistent pool without stopping its workers.
close	Stop all workers using the manager's bounded termination grace.

Source code in src/wetlands/managed_environment.py

class WorkerPool:
    """A group of warm worker processes for one managed environment generation."""

    def __init__(self, environment: ManagedEnvironment, runtime: ExternalEnvironment) -> None:
        self.environment = environment
        self._runtime = runtime
        self._closed = False

    @property
    def worker_count(self) -> int:
        """Return the number of workers in the pool."""
        return self._runtime.worker_count

    def submit_import(
        self,
        target: str,
        *,
        args: tuple[Any, ...] = (),
        kwargs: dict[str, Any] | None = None,
        context_keyword: str | None = None,
    ) -> ExecutionTask[Any]:
        """Submit an installed ``module:qualified.callable`` target for execution."""
        self._ensure_open()
        return self._runtime.submit_import(
            target,
            args=args,
            kwargs=kwargs,
            context_keyword=context_keyword,
        )

    def submit_path(
        self,
        path: str | Path,
        qualname: str,
        *,
        args: tuple[Any, ...] = (),
        kwargs: dict[str, Any] | None = None,
        cache: bool = True,
        context_keyword: str | None = None,
    ) -> ExecutionTask[Any]:
        """Submit a callable from an explicit local source path.

        Path execution is intended for local development; installed packages should use
        :meth:`submit_import`.
        """
        self._ensure_open()
        return self._runtime.submit_path(
            path,
            qualname,
            args=args,
            kwargs=kwargs,
            cache=cache,
            context_keyword=context_keyword,
        )

    def execute_import(
        self,
        target: str,
        *,
        args: tuple[Any, ...] = (),
        kwargs: dict[str, Any] | None = None,
        timeout: float | None = None,
        context_keyword: str | None = None,
    ) -> Any:
        """Execute an installed target and block until it finishes."""
        return self.submit_import(
            target,
            args=args,
            kwargs=kwargs,
            context_keyword=context_keyword,
        ).wait_for(timeout)

    def execute_path(
        self,
        path: str | Path,
        qualname: str,
        *,
        args: tuple[Any, ...] = (),
        kwargs: dict[str, Any] | None = None,
        cache: bool = True,
        timeout: float | None = None,
        context_keyword: str | None = None,
    ) -> Any:
        """Execute a local path target and block until it finishes."""
        return self.submit_path(
            path,
            qualname,
            args=args,
            kwargs=kwargs,
            cache=cache,
            context_keyword=context_keyword,
        ).wait_for(timeout)

    def detach(self) -> None:
        """Release control of a persistent pool without stopping its workers."""
        self._ensure_open()
        self._runtime.detach()
        self._closed = True

    def close(self) -> None:
        """Stop all workers using the manager's bounded termination grace.

        If cleanup raises, the pool remains open so the caller can retry.
        """
        if self._closed:
            return
        self._runtime._exit()
        self._closed = True

    def _ensure_open(self) -> None:
        self.environment._manager._ensure_open()
        if self._closed:
            raise RuntimeError("WorkerPool is closed")
        self._runtime._raise_if_failed()

    def __enter__(self) -> WorkerPool:
        self._ensure_open()
        return self

    def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
        self.close()

worker_count property

Return the number of workers in the pool.

submit_import(target, *, args=(), kwargs=None, context_keyword=None)

Submit an installed module:qualified.callable target for execution.

Source code in src/wetlands/managed_environment.py

def submit_import(
    self,
    target: str,
    *,
    args: tuple[Any, ...] = (),
    kwargs: dict[str, Any] | None = None,
    context_keyword: str | None = None,
) -> ExecutionTask[Any]:
    """Submit an installed ``module:qualified.callable`` target for execution."""
    self._ensure_open()
    return self._runtime.submit_import(
        target,
        args=args,
        kwargs=kwargs,
        context_keyword=context_keyword,
    )

submit_path(path, qualname, *, args=(), kwargs=None, cache=True, context_keyword=None)

Submit a callable from an explicit local source path.

Path execution is intended for local development; installed packages should use :meth:submit_import.

Source code in src/wetlands/managed_environment.py

def submit_path(
    self,
    path: str | Path,
    qualname: str,
    *,
    args: tuple[Any, ...] = (),
    kwargs: dict[str, Any] | None = None,
    cache: bool = True,
    context_keyword: str | None = None,
) -> ExecutionTask[Any]:
    """Submit a callable from an explicit local source path.

    Path execution is intended for local development; installed packages should use
    :meth:`submit_import`.
    """
    self._ensure_open()
    return self._runtime.submit_path(
        path,
        qualname,
        args=args,
        kwargs=kwargs,
        cache=cache,
        context_keyword=context_keyword,
    )

execute_import(target, *, args=(), kwargs=None, timeout=None, context_keyword=None)

Execute an installed target and block until it finishes.

Source code in src/wetlands/managed_environment.py

def execute_import(
    self,
    target: str,
    *,
    args: tuple[Any, ...] = (),
    kwargs: dict[str, Any] | None = None,
    timeout: float | None = None,
    context_keyword: str | None = None,
) -> Any:
    """Execute an installed target and block until it finishes."""
    return self.submit_import(
        target,
        args=args,
        kwargs=kwargs,
        context_keyword=context_keyword,
    ).wait_for(timeout)

execute_path(path, qualname, *, args=(), kwargs=None, cache=True, timeout=None, context_keyword=None)

Execute a local path target and block until it finishes.

Source code in src/wetlands/managed_environment.py

def execute_path(
    self,
    path: str | Path,
    qualname: str,
    *,
    args: tuple[Any, ...] = (),
    kwargs: dict[str, Any] | None = None,
    cache: bool = True,
    timeout: float | None = None,
    context_keyword: str | None = None,
) -> Any:
    """Execute a local path target and block until it finishes."""
    return self.submit_path(
        path,
        qualname,
        args=args,
        kwargs=kwargs,
        cache=cache,
        context_keyword=context_keyword,
    ).wait_for(timeout)

detach()

Release control of a persistent pool without stopping its workers.

Source code in src/wetlands/managed_environment.py

def detach(self) -> None:
    """Release control of a persistent pool without stopping its workers."""
    self._ensure_open()
    self._runtime.detach()
    self._closed = True

close()

Stop all workers using the manager's bounded termination grace.

If cleanup raises, the pool remains open so the caller can retry.

Source code in src/wetlands/managed_environment.py

def close(self) -> None:
    """Stop all workers using the manager's bounded termination grace.

    If cleanup raises, the pool remains open so the caller can retry.
    """
    if self._closed:
        return
    self._runtime._exit()
    self._closed = True

local_package_content_identity(source)

Return a deterministic content identity for an immutable local package tree.

Only regular files and directories are accepted. Directory entries, file paths, file modes, and file contents are hashed in a canonical order; links, special files, portable path collisions, and concurrent source mutation are rejected.

Source code in src/wetlands/specs.py

def local_package_content_identity(source: str | os.PathLike[str]) -> str:
    """Return a deterministic content identity for an immutable local package tree.

    Only regular files and directories are accepted. Directory entries, file
    paths, file modes, and file contents are hashed in a canonical order; links,
    special files, portable path collisions, and concurrent source mutation are
    rejected.
    """

    path = Path(os.path.abspath(Path(source).expanduser()))
    return _local_package_tree_identity(path)