Errors and diagnostics

EnvironmentGenerationChangedError

Bases: RuntimeError

A managed handle or pool no longer matches the ready environment.

Source code in src/wetlands/lifecycle.py

class EnvironmentGenerationChangedError(RuntimeError):
    """A managed handle or pool no longer matches the ready environment."""

    def __init__(
        self,
        environment: str,
        *,
        expected_generation_id: str,
        expected_recipe_hash: str,
        actual_generation_id: str | None,
        actual_recipe_hash: str | None,
    ) -> None:
        self.environment = environment
        self.expected_generation_id = expected_generation_id
        self.expected_recipe_hash = expected_recipe_hash
        self.actual_generation_id = actual_generation_id
        self.actual_recipe_hash = actual_recipe_hash
        super().__init__(
            f"Environment {environment!r} changed generation or recipe "
            f"(expected generation {expected_generation_id!r}, recipe {expected_recipe_hash!r}; "
            f"found generation {actual_generation_id!r}, recipe {actual_recipe_hash!r})"
        )

EnvironmentInUseError

Bases: RuntimeError

A managed environment cannot be changed while managed resources are alive.

Source code in src/wetlands/lifecycle.py

class EnvironmentInUseError(RuntimeError):
    """A managed environment cannot be changed while managed resources are alive."""

    def __init__(self, environment: str, generation_id: str | None = None) -> None:
        self.environment = environment
        self.generation_id = generation_id
        generation = f" generation {generation_id!r}" if generation_id is not None else ""
        super().__init__(f"Environment {environment!r}{generation} is in use by a live managed resource")

EnvironmentNotFoundError

Bases: LookupError

A requested managed environment target does not exist.

Source code in src/wetlands/lifecycle.py

class EnvironmentNotFoundError(LookupError):
    """A requested managed environment target does not exist."""

    def __init__(self, environment: str, *, alias: str | None = None) -> None:
        self.environment = environment
        self.alias = alias
        if alias is None:
            message = f"Managed environment {environment!r} does not exist"
        else:
            message = f"Managed environment {environment!r} does not exist; found portable name alias {alias!r}"
        super().__init__(message)

EnvironmentRecipeConflictError

Bases: RuntimeError

A ready environment has a different recipe and replacement was not requested.

Source code in src/wetlands/lifecycle.py

class EnvironmentRecipeConflictError(RuntimeError):
    """A ready environment has a different recipe and replacement was not requested."""

    def __init__(
        self,
        environment: str,
        *,
        existing_recipe_hash: str,
        requested_recipe_hash: str,
    ) -> None:
        self.environment = environment
        self.existing_recipe_hash = existing_recipe_hash
        self.requested_recipe_hash = requested_recipe_hash
        super().__init__(
            f"Environment {environment!r} has recipe {existing_recipe_hash!r}, "
            f"not requested recipe {requested_recipe_hash!r}; "
            "pass replace_existing=True to rebuild it"
        )

EnvironmentNotReadyError

Bases: RuntimeError

Source code in src/wetlands/environment_manager.py

class EnvironmentNotReadyError(RuntimeError):
    pass

ExecutionError

Bases: OperationError

Source code in src/wetlands/operation.py

class ExecutionError(OperationError):
    def __init__(self, failure: Any):
        message = failure.summary() if hasattr(failure, "summary") else failure.message
        RuntimeError.__init__(self, message)
        self.failure = failure

ExecutionFailure dataclass

Structured details for a failed worker execution.

Source code in src/wetlands/diagnostics.py

@dataclass(frozen=True)
class ExecutionFailure:
    """Structured details for a failed worker execution."""

    category: ExecutionFailureCategory
    message: str
    task_id: str | None = None
    call_target: str | None = None
    traceback: str | None = None
    traceback_frames: list[str] = field(default_factory=list)
    remote_exception: RemoteExceptionInfo | None = None
    worker: WorkerInfo | None = None
    exit_code: int | None = None
    signal: int | None = None
    timeout: float | None = None
    elapsed: float | None = None
    serialization_context: str | None = None
    raw: dict[str, Any] | None = None

    @classmethod
    def normalize(
        cls,
        value: "ExecutionFailure | dict[str, Any] | BaseException | str",
        *,
        traceback: list[str] | str | None = None,
        task_id: str | None = None,
        call_target: str | None = None,
    ) -> "ExecutionFailure":
        if isinstance(value, ExecutionFailure):
            return value.with_defaults(task_id=task_id, call_target=call_target)
        if isinstance(value, BaseException):
            return cls.from_exception(value, task_id=task_id, call_target=call_target)
        if isinstance(value, dict):
            return cls.from_payload(value, task_id=task_id, call_target=call_target)
        traceback_string = _traceback_to_string(traceback)
        return cls(
            category=ExecutionFailureCategory.UNKNOWN,
            message=str(value),
            task_id=task_id,
            call_target=call_target,
            traceback=traceback_string,
            traceback_frames=_traceback_to_frames(traceback),
        )

    @classmethod
    def from_exception(
        cls,
        exc: BaseException,
        *,
        category: ExecutionFailureCategory = ExecutionFailureCategory.REMOTE_EXCEPTION,
        task_id: str | None = None,
        call_target: str | None = None,
        serialization_context: str | None = None,
    ) -> "ExecutionFailure":
        exc_type = type(exc)
        return cls(
            category=category,
            message=str(exc),
            task_id=task_id,
            call_target=call_target,
            traceback="".join(traceback_module.format_exception(exc_type, exc, exc.__traceback__, chain=True)),
            traceback_frames=traceback_module.format_tb(exc.__traceback__),
            remote_exception=RemoteExceptionInfo.from_exception(exc),
            serialization_context=serialization_context,
        )

    @classmethod
    def from_payload(
        cls,
        payload: dict[str, Any],
        *,
        task_id: str | None = None,
        call_target: str | None = None,
    ) -> "ExecutionFailure":
        failure_payload = payload.get("failure") if "failure" in payload else payload
        if not isinstance(failure_payload, dict):
            return cls.normalize(str(failure_payload), task_id=task_id, call_target=call_target)

        legacy_traceback = failure_payload.get("traceback")
        category_value = failure_payload.get("category")
        try:
            category = (
                ExecutionFailureCategory(category_value)
                if category_value is not None
                else ExecutionFailureCategory.REMOTE_EXCEPTION
            )
        except ValueError:
            category = ExecutionFailureCategory.UNKNOWN

        remote_exception = RemoteExceptionInfo.from_payload(failure_payload.get("remote_exception"))
        message = failure_payload.get("message")
        if message is None:
            message = failure_payload.get("exception")
        if message is None and remote_exception is not None:
            message = remote_exception.message
        if message is None:
            message = "Unknown task failure"

        return cls(
            category=category,
            message=str(message),
            task_id=failure_payload.get("task_id") or payload.get("task_id") or task_id,
            call_target=failure_payload.get("call_target") or payload.get("_call_target") or call_target,
            traceback=_traceback_to_string(legacy_traceback),
            traceback_frames=_traceback_to_frames(failure_payload.get("traceback_frames", legacy_traceback)),
            remote_exception=remote_exception,
            worker=WorkerInfo.from_payload(failure_payload.get("worker")),
            exit_code=failure_payload.get("exit_code"),
            signal=failure_payload.get("signal"),
            timeout=failure_payload.get("timeout"),
            elapsed=failure_payload.get("elapsed"),
            serialization_context=failure_payload.get("serialization_context"),
            raw=failure_payload,
        )

    @classmethod
    def environment(
        cls,
        message: str,
        *,
        task_id: str | None = None,
        call_target: str | None = None,
    ) -> "ExecutionFailure":
        return cls(ExecutionFailureCategory.ENVIRONMENT, message, task_id=task_id, call_target=call_target)

    @classmethod
    def serialization(
        cls,
        message: str,
        *,
        task_id: str | None = None,
        call_target: str | None = None,
        context: str | None = None,
        worker: WorkerInfo | None = None,
    ) -> "ExecutionFailure":
        return cls(
            ExecutionFailureCategory.SERIALIZATION,
            message,
            task_id=task_id,
            call_target=call_target,
            serialization_context=context,
            worker=worker,
        )

    @classmethod
    def worker_connection(
        cls,
        message: str,
        *,
        task_id: str | None = None,
        call_target: str | None = None,
        worker: WorkerInfo | None = None,
    ) -> "ExecutionFailure":
        return cls(
            ExecutionFailureCategory.WORKER_CONNECTION,
            message,
            task_id=task_id,
            call_target=call_target,
            worker=worker,
        )

    @classmethod
    def worker_died(
        cls,
        *,
        task_id: str | None = None,
        call_target: str | None = None,
        worker: WorkerInfo | None = None,
        returncode: int | None = None,
    ) -> "ExecutionFailure":
        exit_code = returncode if returncode is not None and returncode >= 0 else None
        signal = -returncode if returncode is not None and returncode < 0 else None
        if signal is not None:
            message = f"Worker process died with signal {signal}"
        elif exit_code is not None:
            message = f"Worker process died with exit code {exit_code}"
        else:
            message = "Worker process died"
        return cls(
            ExecutionFailureCategory.WORKER_DIED,
            message,
            task_id=task_id,
            call_target=call_target,
            worker=worker,
            exit_code=exit_code,
            signal=signal,
        )

    @classmethod
    def timeout_failure(
        cls,
        *,
        task_id: str | None = None,
        call_target: str | None = None,
        worker: WorkerInfo | None = None,
        timeout: float | None = None,
        elapsed: float | None = None,
    ) -> "ExecutionFailure":
        message = (
            f"Task timed out after {elapsed:.1f}s without worker activity"
            if elapsed is not None
            else "Task timed out without worker activity"
        )
        return cls(
            ExecutionFailureCategory.TIMEOUT,
            message,
            task_id=task_id,
            call_target=call_target,
            worker=worker,
            timeout=timeout,
            elapsed=elapsed,
        )

    def with_defaults(self, *, task_id: str | None = None, call_target: str | None = None) -> "ExecutionFailure":
        if (task_id is None or self.task_id is not None) and (call_target is None or self.call_target is not None):
            return self
        return ExecutionFailure(
            category=self.category,
            message=self.message,
            task_id=self.task_id or task_id,
            call_target=self.call_target or call_target,
            traceback=self.traceback,
            traceback_frames=list(self.traceback_frames),
            remote_exception=self.remote_exception,
            worker=self.worker,
            exit_code=self.exit_code,
            signal=self.signal,
            timeout=self.timeout,
            elapsed=self.elapsed,
            serialization_context=self.serialization_context,
            raw=self.raw,
        )

    def to_payload(self) -> dict[str, Any]:
        return {
            "category": self.category.value,
            "message": self.message,
            "task_id": self.task_id,
            "call_target": self.call_target,
            "traceback": self.traceback,
            "traceback_frames": list(self.traceback_frames),
            "remote_exception": self.remote_exception.to_payload() if self.remote_exception else None,
            "worker": self.worker.to_payload() if self.worker else None,
            "exit_code": self.exit_code,
            "signal": self.signal,
            "timeout": self.timeout,
            "elapsed": self.elapsed,
            "serialization_context": self.serialization_context,
        }

    def summary(self) -> str:
        if self.category in (
            ExecutionFailureCategory.REMOTE_EXCEPTION,
            ExecutionFailureCategory.INTERNAL_EXCEPTION,
        ):
            prefix = "Remote" if self.category == ExecutionFailureCategory.REMOTE_EXCEPTION else "Local"
            if self.remote_exception is not None:
                exc_name = self.remote_exception.qualified_name or self.remote_exception.type_name or "Exception"
                module = self.remote_exception.module
                source = f" from {module}" if module else ""
                message = self.remote_exception.message or self.message
                return f"{prefix} {exc_name}{source}: {message}"
        if self.category == ExecutionFailureCategory.WORKER_DIED:
            worker = _worker_label(self.worker)
            if self.signal is not None:
                return f"{worker}died with signal {self.signal}"
            if self.exit_code is not None:
                return f"{worker}died with exit code {self.exit_code}"
        if self.category == ExecutionFailureCategory.TIMEOUT:
            if self.elapsed is not None:
                return f"Task timed out after {self.elapsed:.1f}s without worker activity"
            return "Task timed out without worker activity"
        if self.category == ExecutionFailureCategory.SERIALIZATION:
            context = f" while serializing {self.serialization_context}" if self.serialization_context else ""
            return f"Task serialization failure{context}: {self.message}"
        return self.message

ExecutionFailureCategory

Bases: str, Enum

Stable categories for failures observed at the execution boundary.

Source code in src/wetlands/diagnostics.py

class ExecutionFailureCategory(str, enum.Enum):
    """Stable categories for failures observed at the execution boundary."""

    REMOTE_EXCEPTION = "remote_exception"
    INTERNAL_EXCEPTION = "internal_exception"
    SERIALIZATION = "serialization"
    WORKER_CONNECTION = "worker_connection"
    WORKER_DIED = "worker_died"
    TIMEOUT = "timeout"
    ENVIRONMENT = "environment"
    UNKNOWN = "unknown"

InvalidStateError

Bases: Exception

Raised when accessing task result in an invalid state.

Source code in src/wetlands/task.py

class InvalidStateError(Exception):
    """Raised when accessing task result in an invalid state."""

LocalPackageValidationError

Bases: ValueError

A local package cannot be represented as a deterministic Pixi requirement.

Source code in src/wetlands/specs.py

class LocalPackageValidationError(ValueError):
    """A local package cannot be represented as a deterministic Pixi requirement."""

ManagerCloseError

Bases: RuntimeError

One or more resources could not be cleaned up during manager shutdown.

Source code in src/wetlands/lifecycle.py

class ManagerCloseError(RuntimeError):
    """One or more resources could not be cleaned up during manager shutdown."""

    def __init__(self, errors: tuple[BaseException, ...]) -> None:
        if not errors:
            raise ValueError("errors must not be empty")
        self.errors = errors
        details = "; ".join(f"{type(error).__name__}: {error}" for error in errors)
        super().__init__(f"EnvironmentManager cleanup did not complete: {details}")

ManagerCloseTimeoutError

Bases: TimeoutError

A manager shutdown phase exceeded its shared close deadline.

Source code in src/wetlands/lifecycle.py

class ManagerCloseTimeoutError(TimeoutError):
    """A manager shutdown phase exceeded its shared close deadline."""

    def __init__(self, phase: str, timeout: float) -> None:
        self.phase = phase
        self.timeout = timeout
        super().__init__(f"EnvironmentManager close timed out during {phase!r} after {timeout} seconds")

OperationCanceled

Bases: RuntimeError

Source code in src/wetlands/operation.py

class OperationCanceled(RuntimeError):
    def __init__(self, operation_id: str, message: str = "Operation was canceled"):
        super().__init__(message)
        self.operation_id = operation_id

OperationError

Bases: RuntimeError

Source code in src/wetlands/operation.py

class OperationError(RuntimeError):
    def __init__(self, failure: Any):
        super().__init__(failure.message)
        self.failure = failure

OperationFailure dataclass

Methods:

Name	Description
to_payload	Return a stable JSON-compatible diagnostic representation.

Source code in src/wetlands/operation.py

@dataclass(frozen=True)
class OperationFailure:
    operation_id: str
    stage: str
    message: str
    step_id: str | None = None
    command: str | None = None
    returncode: int | None = None
    stdout_tail: tuple[str, ...] = ()
    stderr_tail: tuple[str, ...] = ()
    environment: str | None = None
    cleanup_error: str | None = None

    def to_payload(self) -> dict[str, Any]:
        """Return a stable JSON-compatible diagnostic representation."""

        return {
            "operation_id": self.operation_id,
            "stage": self.stage,
            "message": self.message,
            "step_id": self.step_id,
            "command": self.command,
            "returncode": self.returncode,
            "stdout_tail": list(self.stdout_tail),
            "stderr_tail": list(self.stderr_tail),
            "environment": self.environment,
            "cleanup_error": self.cleanup_error,
        }

to_payload()

Return a stable JSON-compatible diagnostic representation.

Source code in src/wetlands/operation.py

def to_payload(self) -> dict[str, Any]:
    """Return a stable JSON-compatible diagnostic representation."""

    return {
        "operation_id": self.operation_id,
        "stage": self.stage,
        "message": self.message,
        "step_id": self.step_id,
        "command": self.command,
        "returncode": self.returncode,
        "stdout_tail": list(self.stdout_tail),
        "stderr_tail": list(self.stderr_tail),
        "environment": self.environment,
        "cleanup_error": self.cleanup_error,
    }

PreparationError

Bases: OperationError

Source code in src/wetlands/operation.py

class PreparationError(OperationError):
    pass

ProvisioningError

Bases: OperationError

Source code in src/wetlands/operation.py

class ProvisioningError(OperationError):
    pass

ProcessCleanupError

Bases: ProcessError

The complete owned process tree could not be proven terminated.

Source code in src/wetlands/managed_process.py

class ProcessCleanupError(ProcessError):
    """The complete owned process tree could not be proven terminated."""

    def __init__(
        self,
        failures: Sequence[BaseException],
        result: ManagedProcessResult | None,
        *,
        argv: tuple[str, ...],
        environment: str,
        generation_id: str,
        initiating_error: ProcessError | None = None,
    ) -> None:
        frozen_failures = tuple(failures)
        details = "; ".join(str(failure) for failure in frozen_failures)
        super().__init__(
            f"Could not completely clean up command {argv!r}: {details}",
            argv=argv,
            environment=environment,
            generation_id=generation_id,
        )
        self.failures = frozen_failures
        self.result = result
        self.initiating_error = initiating_error
        if initiating_error is not None:
            self.__cause__ = initiating_error

ProcessError

Bases: RuntimeError

Base class for errors reported by a managed command.

Source code in src/wetlands/managed_process.py

class ProcessError(RuntimeError):
    """Base class for errors reported by a managed command."""

    def __init__(self, message: str, *, argv: tuple[str, ...], environment: str, generation_id: str) -> None:
        super().__init__(message)
        self.argv = argv
        self.environment = environment
        self.generation_id = generation_id

ProcessEventLagError

Bases: ProcessError

An output observer fell behind the bounded event history.

Source code in src/wetlands/managed_process.py

class ProcessEventLagError(ProcessError):
    """An output observer fell behind the bounded event history."""

    def __init__(
        self,
        first_unavailable_sequence: int,
        oldest_retained_sequence: int,
        *,
        argv: tuple[str, ...],
        environment: str,
        generation_id: str,
    ) -> None:
        super().__init__(
            "Output observer fell behind: "
            f"sequence {first_unavailable_sequence} is unavailable; oldest retained is {oldest_retained_sequence}",
            argv=argv,
            environment=environment,
            generation_id=generation_id,
        )
        self.first_unavailable_sequence = first_unavailable_sequence
        self.oldest_retained_sequence = oldest_retained_sequence

ProcessExitError

Bases: ProcessError

A checked command exited with a non-zero status.

Source code in src/wetlands/managed_process.py

class ProcessExitError(ProcessError):
    """A checked command exited with a non-zero status."""

    def __init__(self, result: ManagedProcessResult, *, environment: str, generation_id: str) -> None:
        super().__init__(
            f"Command {result.argv!r} exited with status {result.returncode}",
            argv=result.argv,
            environment=environment,
            generation_id=generation_id,
        )
        self.result = result

ProcessLineTimeoutError

Bases: ProcessError, TimeoutError

No matching output event arrived before a readiness deadline.

Source code in src/wetlands/managed_process.py

class ProcessLineTimeoutError(ProcessError, TimeoutError):
    """No matching output event arrived before a readiness deadline."""

    def __init__(
        self,
        timeout: float,
        *,
        argv: tuple[str, ...],
        environment: str,
        generation_id: str,
    ) -> None:
        super().__init__(
            f"No matching output from command {argv!r} arrived within {timeout} seconds",
            argv=argv,
            environment=environment,
            generation_id=generation_id,
        )
        self.timeout = timeout

ProcessOutputLimitError

Bases: ProcessError

A command emitted more output than its configured capture limit.

Source code in src/wetlands/managed_process.py

class ProcessOutputLimitError(ProcessError):
    """A command emitted more output than its configured capture limit."""

    def __init__(
        self,
        limit: int,
        result: ManagedProcessResult,
        truncated_streams: frozenset[OutputStream],
        *,
        environment: str,
        generation_id: str,
    ) -> None:
        streams = ", ".join(sorted(stream.value for stream in truncated_streams))
        super().__init__(
            f"Command {result.argv!r} exceeded its {limit}-byte output limit on {streams}",
            argv=result.argv,
            environment=environment,
            generation_id=generation_id,
        )
        self.limit = limit
        self.result = result
        self.truncated_streams = truncated_streams

ProcessTimeoutError

Bases: ProcessError, TimeoutError

A command exceeded the timeout of a wait operation.

Source code in src/wetlands/managed_process.py

class ProcessTimeoutError(ProcessError, TimeoutError):
    """A command exceeded the timeout of a wait operation."""

    def __init__(
        self,
        timeout: float,
        result: ManagedProcessResult,
        *,
        environment: str,
        generation_id: str,
    ) -> None:
        super().__init__(
            f"Command {result.argv!r} did not finish within {timeout} seconds",
            argv=result.argv,
            environment=environment,
            generation_id=generation_id,
        )
        self.timeout = timeout
        self.result = result

RemoteExceptionInfo dataclass

Serializable identity and traceback information for a remote exception.

Source code in src/wetlands/diagnostics.py

@dataclass(frozen=True)
class RemoteExceptionInfo:
    """Serializable identity and traceback information for a remote exception."""

    module: str | None = None
    type_name: str | None = None
    qualified_name: str | None = None
    message: str | None = None
    traceback: str | None = None
    cause: "RemoteExceptionInfo | None" = None
    context: "RemoteExceptionInfo | None" = None
    suppress_context: bool = False

    @classmethod
    def from_exception(cls, exc: BaseException) -> "RemoteExceptionInfo":
        exc_type = type(exc)
        return cls(
            module=exc_type.__module__,
            type_name=exc_type.__name__,
            qualified_name=getattr(exc_type, "__qualname__", exc_type.__name__),
            message=str(exc),
            traceback="".join(traceback_module.format_exception(exc_type, exc, exc.__traceback__, chain=False)),
            cause=cls.from_exception(exc.__cause__) if exc.__cause__ is not None else None,
            context=cls.from_exception(exc.__context__) if exc.__context__ is not None else None,
            suppress_context=bool(getattr(exc, "__suppress_context__", False)),
        )

    @classmethod
    def from_payload(cls, payload: dict[str, Any] | None) -> "RemoteExceptionInfo | None":
        if not payload:
            return None
        return cls(
            module=payload.get("module"),
            type_name=payload.get("type_name"),
            qualified_name=payload.get("qualified_name"),
            message=payload.get("message"),
            traceback=payload.get("traceback"),
            cause=cls.from_payload(payload.get("cause")),
            context=cls.from_payload(payload.get("context")),
            suppress_context=bool(payload.get("suppress_context", False)),
        )

    def to_payload(self) -> dict[str, Any]:
        return {
            "module": self.module,
            "type_name": self.type_name,
            "qualified_name": self.qualified_name,
            "message": self.message,
            "traceback": self.traceback,
            "cause": self.cause.to_payload() if self.cause is not None else None,
            "context": self.context.to_payload() if self.context is not None else None,
            "suppress_context": self.suppress_context,
        }

RemovalError

Bases: OperationError

A managed environment could not be removed cleanly.

Source code in src/wetlands/operation.py

class RemovalError(OperationError):
    """A managed environment could not be removed cleanly."""

UnmanagedTargetError

Bases: RuntimeError

An existing target is not proven to be owned by Wetlands.

Source code in src/wetlands/lifecycle.py

class UnmanagedTargetError(RuntimeError):
    """An existing target is not proven to be owned by Wetlands."""

    def __init__(self, environment: str, path: str | Path) -> None:
        self.environment = environment
        self.path = Path(path)
        super().__init__(f"Environment target {str(self.path)!r} is unmanaged; Wetlands will not modify or remove it")

ValueDecodingError

Bases: ValueError

Source code in src/wetlands/_internal/value_codec.py

class ValueDecodingError(ValueError):
    pass

ValueEncodingError

Bases: TypeError

Source code in src/wetlands/_internal/value_codec.py

class ValueEncodingError(TypeError):
    pass

WorkerInfo dataclass

Worker identity attached to an execution failure.

Source code in src/wetlands/diagnostics.py

@dataclass(frozen=True)
class WorkerInfo:
    """Worker identity attached to an execution failure."""

    environment: str | None = None
    index: int | None = None
    pid: int | None = None
    port: int | None = None
    persistent: bool | None = None

    @classmethod
    def from_payload(cls, payload: dict[str, Any] | None) -> "WorkerInfo | None":
        if not payload:
            return None
        return cls(
            environment=payload.get("environment"),
            index=payload.get("index"),
            pid=payload.get("pid"),
            port=payload.get("port"),
            persistent=payload.get("persistent"),
        )

    def to_payload(self) -> dict[str, Any]:
        return {
            "environment": self.environment,
            "index": self.index,
            "pid": self.pid,
            "port": self.port,
            "persistent": self.persistent,
        }

WorkerStartError

Bases: RuntimeError

A worker pool could not be launched or attached cleanly.

Source code in src/wetlands/lifecycle.py

class WorkerStartError(RuntimeError):
    """A worker pool could not be launched or attached cleanly."""

    def __init__(
        self,
        environment: str,
        message: str,
        *,
        worker_index: int | None = None,
        phase: str = "launch",
        cleanup_errors: tuple[str, ...] = (),
    ) -> None:
        self.environment = environment
        self.worker_index = worker_index
        self.phase = phase
        self.cleanup_errors = cleanup_errors
        worker = f" worker {worker_index}" if worker_index is not None else " worker pool"
        cleanup = f"; cleanup also failed: {'; '.join(cleanup_errors)}" if cleanup_errors else ""
        super().__init__(f"Could not {phase}{worker} for environment {environment!r}: {message}{cleanup}")