Operations and tasks

ExecutionEvent dataclass

Immutable observation snapshot emitted by an execution task.

Source code in src/wetlands/task.py

@dataclass(frozen=True)
class ExecutionEvent:
    """Immutable observation snapshot emitted by an execution task."""

    sequence: int
    timestamp: float
    task_id: str
    kind: ExecutionEventKind
    state: ExecutionState
    message: str
    current: int | None = None
    maximum: int | None = None
    progress: float | None = None
    failure: ExecutionFailure | None = None

ExecutionEventKind

Bases: Enum

Types of events emitted by an execution task.

Source code in src/wetlands/task.py

class ExecutionEventKind(enum.Enum):
    """Types of events emitted by an execution task."""

    STARTED = "started"
    UPDATE = "update"
    COMPLETION = "completion"
    FAILURE = "failure"
    CANCELLATION_REQUESTED = "cancellation_requested"
    CANCELLATION = "cancellation"

ExecutionState

Bases: Enum

Status of a task through its lifecycle.

Source code in src/wetlands/task.py

class ExecutionState(enum.Enum):
    """Status of a task through its lifecycle."""

    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELED = "canceled"

    @property
    def terminal(self) -> bool:
        """Whether the execution reached a terminal state."""
        return self in (ExecutionState.COMPLETED, ExecutionState.FAILED, ExecutionState.CANCELED)

terminal property

Whether the execution reached a terminal state.

ExecutionTask

Bases: Generic[T]

Represents an asynchronous unit of work in a remote environment.

Type parameter T is the return type of the remote function.

Methods:

Name	Description
cancel	Request cooperative cancellation.
wait_for	Block until the task reaches a terminal state.
listen	Register a listener, replaying bounded history by default.
remove_listener	Remove a previously registered listener.
__await__	Return an awaiter which does not bypass cancellation cleanup.
events	Iterate over bounded history and live events until terminal state.

Source code in src/wetlands/task.py

class ExecutionTask(Generic[T]):
    """Represents an asynchronous unit of work in a remote environment.

    Type parameter T is the return type of the remote function.
    """

    _EVENT_HISTORY_LIMIT = 2048

    def __init__(self, task_id: str | None = None) -> None:
        self._id = task_id or str(uuid.uuid4())
        self._status = ExecutionState.PENDING
        self._result: T | None = None
        self._error: ExecutionFailure | None = None
        self._traceback: str | None = None
        self._exception: ExecutionError | None = None
        self._message: str | None = None
        self._current: int | None = None
        self._maximum: int | None = None
        self._outputs: dict[str, Any] = {}
        self._listeners: list[Callable[[ExecutionEvent], None]] = []
        self._events: deque[ExecutionEvent] = deque(maxlen=self._EVENT_HISTORY_LIMIT)
        self._sequence = 0
        self._last_event_timestamp = 0.0
        self._future: Future[T] = Future()
        self._cancellation_requested = False
        self._lock = threading.RLock()
        self._done_event = threading.Event()
        self._payload: dict[str, Any] = {}

        # Set by the environment before dispatch
        self._start_fn: Callable[[], None] | None = None
        self._cancel_fn: Callable[[], None] | None = None

    @property
    def id(self) -> str:
        return self._id

    @property
    def state(self) -> ExecutionState:
        """The execution lifecycle state."""
        with self._lock:
            return self._status

    @property
    def result(self) -> T:
        """The return value. Raises InvalidStateError if not COMPLETED."""
        if self._status != ExecutionState.COMPLETED:
            raise InvalidStateError(f"Task is {self._status.value}, not completed")
        return self._result  # type: ignore[return-value]

    @property
    def error(self) -> ExecutionFailure | None:
        return self._error

    @property
    def traceback(self) -> str | None:
        return self._traceback

    @property
    def exception(self) -> ExecutionError | None:
        return self._exception

    @property
    def message(self) -> str | None:
        return self._message

    @property
    def current(self) -> int | None:
        return self._current

    @property
    def maximum(self) -> int | None:
        return self._maximum

    @property
    def progress(self) -> float | None:
        """current / maximum as a float in [0, 1]. None if unavailable."""
        if self._current is not None and self._maximum is not None and self._maximum > 0:
            return self._current / self._maximum
        return None

    @property
    def outputs(self) -> dict[str, Any]:
        with self._lock:
            return dict(self._outputs)

    # --- Control ---

    def _start(self) -> Self:
        """Dispatch a task after its owning worker pool has configured it."""
        with self._lock:
            if self._status != ExecutionState.PENDING:
                return self
            if self._start_fn is None:
                raise InvalidStateError("Task has no start function. Was it created via submit()?")
            start_fn = self._start_fn
        start_fn()
        return self

    @property
    def cancellation_requested(self) -> bool:
        with self._lock:
            return self._cancellation_requested

    def cancel(self) -> bool:
        """Request cooperative cancellation.
        Sets a flag that the remote code can check via task.cancel_requested.
        Does nothing if the task is already finished.
        """
        with self._lock:
            if self._status.terminal:
                return False
            first_request = not self._cancellation_requested
            self._cancellation_requested = True
            cancel_fn = self._cancel_fn
            event_data = (
                self._record_event_locked(
                    ExecutionEventKind.CANCELLATION_REQUESTED,
                    "Cancellation requested",
                )
                if first_request
                else None
            )
        if event_data is not None:
            self._notify(*event_data)
        if first_request and cancel_fn is not None:
            cancel_fn()
        return True

    def wait_for(self, timeout: float | None = None) -> T:
        """Block until the task reaches a terminal state.
        Raises TimeoutError if timeout (in seconds) is exceeded.
        Does NOT cancel the task on timeout (matches concurrent.futures behavior).
        Returns the task result.
        """
        if not self._done_event.wait(timeout=timeout):
            raise TimeoutError(f"Task {self._id} did not finish within {timeout}s")
        if self._status == ExecutionState.COMPLETED:
            return cast(T, self._result)
        if self._status == ExecutionState.CANCELED:
            raise OperationCanceled(self._id)
        if self._exception is not None:
            raise self._exception
        raise InvalidStateError(f"Task reached unexpected terminal state {self._status.value}")

    # --- Observation ---

    def listen(
        self,
        callback: Callable[[ExecutionEvent], None],
        *,
        replay: bool = True,
    ) -> Self:
        """Register a listener, replaying bounded history by default."""
        with self._lock:
            history = tuple(self._events) if replay else ()
            for event in history:
                self._notify_listener(callback, event)
            if not self._status.terminal:
                self._listeners.append(callback)
        return self

    def remove_listener(self, callback: Callable[[ExecutionEvent], None]) -> None:
        """Remove a previously registered listener."""
        with self._lock:
            if callback in self._listeners:
                self._listeners.remove(callback)

    # --- Awaitable ---

    def __await__(self):
        """Return an awaiter which does not bypass cancellation cleanup."""
        return self._async_result().__await__()

    async def _async_result(self) -> T:
        loop = asyncio.get_running_loop()
        waiter = asyncio.wrap_future(self._future, loop=loop)
        try:
            return await asyncio.shield(waiter)
        except asyncio.CancelledError:
            self.cancel()
            cleanup_waiter = loop.run_in_executor(None, self._done_event.wait)
            while not cleanup_waiter.done():
                try:
                    await asyncio.shield(cleanup_waiter)
                except asyncio.CancelledError:
                    continue
            try:
                await asyncio.shield(waiter)
            except BaseException:
                pass
            raise

    # --- Async event stream ---

    async def events(self, *, replay: bool = True) -> AsyncIterator[ExecutionEvent]:
        """Iterate over bounded history and live events until terminal state."""
        queue: asyncio.Queue[ExecutionEvent] = asyncio.Queue(maxsize=self._EVENT_HISTORY_LIMIT)
        loop = asyncio.get_running_loop()

        def enqueue(event: ExecutionEvent) -> None:
            if queue.full():
                queue.get_nowait()
            queue.put_nowait(event)

        def receive(event: ExecutionEvent) -> None:
            loop.call_soon_threadsafe(enqueue, event)

        with self._lock:
            terminal_without_replay = self._status.terminal and not replay
            if replay:
                for event in self._events:
                    queue.put_nowait(event)
            if not self._status.terminal:
                self._listeners.append(receive)
        if terminal_without_replay:
            return
        try:
            while True:
                event = await queue.get()
                yield event
                if event.state.terminal:
                    return
        finally:
            self.remove_listener(receive)

    # --- Internal methods (called by the environment/IPC reader) ---

    def _set_start_fn(self, fn: Callable[[], None]) -> None:
        self._start_fn = fn

    def _set_cancel_fn(self, fn: Callable[[], None]) -> None:
        self._cancel_fn = fn

    def _set_running(self) -> None:
        with self._lock:
            if self._status is not ExecutionState.PENDING:
                return
            self._status = ExecutionState.RUNNING
            event_data = self._record_event_locked(
                ExecutionEventKind.STARTED,
                "Execution started",
            )
        self._notify(*event_data)

    def _set_completed(self, result: T) -> None:
        with self._lock:
            if self._status.terminal:
                return
            if self._cancellation_requested:
                canceled = True
            else:
                canceled = False
                self._status = ExecutionState.COMPLETED
                self._result = result
                event_data = self._record_event_locked(
                    ExecutionEventKind.COMPLETION,
                    "Execution completed",
                )
        if canceled:
            self._set_canceled()
            return
        self._future.set_result(result)
        self._done_event.set()
        self._notify(*event_data)

    def _set_failed(self, error: Any, traceback: list[str] | str | None = None) -> None:
        call_target = self._payload.get("_call_target") if isinstance(self._payload, dict) else None
        if ExecutionFailure is not None:
            failure = ExecutionFailure.normalize(
                error,
                traceback=traceback,
                task_id=self._id,
                call_target=call_target,
            )
            exception = ExecutionError(failure)
        else:
            failure = error
            exception = ExecutionError(str(error))
        with self._lock:
            if self._status.terminal:
                return
            self._status = ExecutionState.FAILED
            self._error = failure
            self._traceback = failure.traceback if ExecutionFailure is not None else traceback  # type: ignore[union-attr,assignment]
            self._exception = exception
            event_data = self._record_event_locked(
                ExecutionEventKind.FAILURE,
                str(exception) or "Execution failed",
                failure=failure,
            )
        self._future.set_exception(self._exception)
        self._done_event.set()
        self._notify(*event_data)

    def _set_canceled(self) -> None:
        with self._lock:
            if self._status.terminal:
                return
            self._status = ExecutionState.CANCELED
            event_data = self._record_event_locked(
                ExecutionEventKind.CANCELLATION,
                "Execution canceled",
            )
        self._future.set_exception(OperationCanceled(self._id))
        self._done_event.set()
        self._notify(*event_data)

    def _set_update(
        self,
        message: str | None = None,
        current: int | None = None,
        maximum: int | None = None,
        outputs: dict[str, Any] | None = None,
    ) -> None:
        if message is not None and not isinstance(message, str):
            raise TypeError("Progress message must be a string")
        for field, value in {"current": current, "maximum": maximum}.items():
            if value is not None and (type(value) is not int or value < 0):
                raise TypeError(f"Progress {field} must be a nonnegative integer")
        if outputs:
            _validate_intermediate_value(outputs, path="outputs")
        with self._lock:
            if message is not None:
                self._message = message
            if current is not None:
                self._current = current
            if maximum is not None:
                self._maximum = maximum
            if outputs:
                self._outputs.update(outputs)
            event_data = self._record_event_locked(
                ExecutionEventKind.UPDATE,
                message or self._message or "Execution progress updated",
            )
        self._notify(*event_data)

    def _record_event_locked(
        self,
        kind: ExecutionEventKind,
        message: str,
        *,
        failure: Any | None = None,
    ) -> tuple[ExecutionEvent, tuple[Callable[[ExecutionEvent], None], ...]]:
        self._sequence += 1
        timestamp = max(time.time(), self._last_event_timestamp)
        self._last_event_timestamp = timestamp
        progress = (
            self._current / self._maximum
            if self._current is not None and self._maximum is not None and self._maximum > 0
            else None
        )
        event = ExecutionEvent(
            sequence=self._sequence,
            timestamp=timestamp,
            task_id=self._id,
            kind=kind,
            state=self._status,
            message=message,
            current=self._current,
            maximum=self._maximum,
            progress=progress,
            failure=failure,
        )
        self._events.append(event)
        return event, tuple(self._listeners)

    def _notify(
        self,
        event: ExecutionEvent,
        listeners: tuple[Callable[[ExecutionEvent], None], ...],
    ) -> None:
        for listener in listeners:
            self._notify_listener(listener, event)

    def _notify_listener(
        self,
        listener: Callable[[ExecutionEvent], None],
        event: ExecutionEvent,
    ) -> None:
        try:
            listener(event)
        except Exception:
            logging.getLogger(__name__).exception("Execution listener failed")

    def _on_message(self, message: dict[str, Any]) -> None:
        """Handle an IPC message from the remote worker."""
        action = message.get("action")
        if action == "execution finished":
            self._set_completed(cast(T, message.get("result")))
        elif action == "error":
            self._set_failed(message)
        elif action == "update":
            self._set_update(
                message=message.get("message"),
                current=message.get("current"),
                maximum=message.get("maximum"),
                outputs=message.get("outputs"),
            )
        elif action == "canceled":
            self._set_canceled()

state property

The execution lifecycle state.

result property

The return value. Raises InvalidStateError if not COMPLETED.

progress property

current / maximum as a float in [0, 1]. None if unavailable.

cancel()

Request cooperative cancellation. Sets a flag that the remote code can check via task.cancel_requested. Does nothing if the task is already finished.

Source code in src/wetlands/task.py

def cancel(self) -> bool:
    """Request cooperative cancellation.
    Sets a flag that the remote code can check via task.cancel_requested.
    Does nothing if the task is already finished.
    """
    with self._lock:
        if self._status.terminal:
            return False
        first_request = not self._cancellation_requested
        self._cancellation_requested = True
        cancel_fn = self._cancel_fn
        event_data = (
            self._record_event_locked(
                ExecutionEventKind.CANCELLATION_REQUESTED,
                "Cancellation requested",
            )
            if first_request
            else None
        )
    if event_data is not None:
        self._notify(*event_data)
    if first_request and cancel_fn is not None:
        cancel_fn()
    return True

wait_for(timeout=None)

Block until the task reaches a terminal state. Raises TimeoutError if timeout (in seconds) is exceeded. Does NOT cancel the task on timeout (matches concurrent.futures behavior). Returns the task result.

Source code in src/wetlands/task.py

def wait_for(self, timeout: float | None = None) -> T:
    """Block until the task reaches a terminal state.
    Raises TimeoutError if timeout (in seconds) is exceeded.
    Does NOT cancel the task on timeout (matches concurrent.futures behavior).
    Returns the task result.
    """
    if not self._done_event.wait(timeout=timeout):
        raise TimeoutError(f"Task {self._id} did not finish within {timeout}s")
    if self._status == ExecutionState.COMPLETED:
        return cast(T, self._result)
    if self._status == ExecutionState.CANCELED:
        raise OperationCanceled(self._id)
    if self._exception is not None:
        raise self._exception
    raise InvalidStateError(f"Task reached unexpected terminal state {self._status.value}")

listen(callback, *, replay=True)

Register a listener, replaying bounded history by default.

Source code in src/wetlands/task.py

def listen(
    self,
    callback: Callable[[ExecutionEvent], None],
    *,
    replay: bool = True,
) -> Self:
    """Register a listener, replaying bounded history by default."""
    with self._lock:
        history = tuple(self._events) if replay else ()
        for event in history:
            self._notify_listener(callback, event)
        if not self._status.terminal:
            self._listeners.append(callback)
    return self

remove_listener(callback)

Remove a previously registered listener.

Source code in src/wetlands/task.py

def remove_listener(self, callback: Callable[[ExecutionEvent], None]) -> None:
    """Remove a previously registered listener."""
    with self._lock:
        if callback in self._listeners:
            self._listeners.remove(callback)

__await__()

Return an awaiter which does not bypass cancellation cleanup.

Source code in src/wetlands/task.py

def __await__(self):
    """Return an awaiter which does not bypass cancellation cleanup."""
    return self._async_result().__await__()

events(*, replay=True) async

Iterate over bounded history and live events until terminal state.

Source code in src/wetlands/task.py

async def events(self, *, replay: bool = True) -> AsyncIterator[ExecutionEvent]:
    """Iterate over bounded history and live events until terminal state."""
    queue: asyncio.Queue[ExecutionEvent] = asyncio.Queue(maxsize=self._EVENT_HISTORY_LIMIT)
    loop = asyncio.get_running_loop()

    def enqueue(event: ExecutionEvent) -> None:
        if queue.full():
            queue.get_nowait()
        queue.put_nowait(event)

    def receive(event: ExecutionEvent) -> None:
        loop.call_soon_threadsafe(enqueue, event)

    with self._lock:
        terminal_without_replay = self._status.terminal and not replay
        if replay:
            for event in self._events:
                queue.put_nowait(event)
        if not self._status.terminal:
            self._listeners.append(receive)
    if terminal_without_replay:
        return
    try:
        while True:
            event = await queue.get()
            yield event
            if event.state.terminal:
                return
    finally:
        self.remove_listener(receive)

Operation

Bases: Generic[T]

A cleanup-aware unit of asynchronous work.

Subclasses arrange execution by calling _start_runner exactly once. Cancellation is only a request until the runner has completed termination and cleanup and calls _set_canceled.

Source code in src/wetlands/operation.py

class Operation(Generic[T]):
    """A cleanup-aware unit of asynchronous work.

    Subclasses arrange execution by calling ``_start_runner`` exactly once.
    Cancellation is only a request until the runner has completed termination and
    cleanup and calls ``_set_canceled``.
    """

    _EVENT_HISTORY_LIMIT = 2048

    def __init__(self, operation_id: str | None = None, *, environment: str | None = None) -> None:
        self._id = operation_id or str(uuid.uuid4())
        self._environment = environment
        self._state = OperationState.PENDING
        self._result: T | None = None
        self._exception: BaseException | None = None
        self._cancellation_requested = False
        self._cancellation_sealed = False
        self._cancel_callback: Callable[[], None] | None = None
        self._listeners: list[tuple[Callable[[OperationEvent], None], Callable[[OperationEvent], None]]] = []
        self._events: deque[OperationEvent] = deque(maxlen=self._EVENT_HISTORY_LIMIT)
        self._pending_notifications: deque[
            tuple[OperationEvent, tuple[Callable[[OperationEvent], None], ...], threading.Event]
        ] = deque()
        self._notification_thread_id: int | None = None
        self._sequence = 0
        self._lock = threading.RLock()
        self._done = threading.Event()
        self._thread: threading.Thread | None = None

    @property
    def id(self) -> str:
        return self._id

    @property
    def state(self) -> OperationState:
        with self._lock:
            return self._state

    @property
    def cancellation_requested(self) -> bool:
        with self._lock:
            return self._cancellation_requested

    def cancel(self) -> bool:
        with self._lock:
            if self._state.terminal or self._cancellation_sealed:
                return False
            first_request = not self._cancellation_requested
            self._cancellation_requested = True
            callback = self._cancel_callback
            publication = (
                self._queue_event_locked(
                    OperationEventKind.CANCELLATION_REQUESTED,
                    "Cancellation requested",
                )
                if first_request
                else None
            )
        if publication is not None:
            self._finish_event_publication(publication)
        if callback is not None:
            try:
                callback()
            except Exception:
                logger.exception("Operation cancellation callback failed")
        return True

    def wait_for(self, timeout: float | None = None) -> T:
        if not self._done.wait(timeout):
            raise TimeoutError(f"Operation {self.id} did not finish within {timeout} seconds")
        with self._lock:
            state = self._state
            result = self._result
            exception = self._exception
        if state is OperationState.COMPLETED:
            return cast(T, result)
        if state is OperationState.CANCELED:
            if isinstance(exception, OperationCanceled):
                raise exception
            raise OperationCanceled(self.id)
        if exception is not None:
            raise exception
        raise RuntimeError(f"Operation {self.id} reached invalid terminal state {state.value}")

    def __await__(self):
        return self._async_result().__await__()

    async def _async_result(self) -> T:
        loop = asyncio.get_running_loop()
        waiter = loop.run_in_executor(None, self.wait_for)
        try:
            return await asyncio.shield(waiter)
        except asyncio.CancelledError:
            self.cancel()
            while not self._done.is_set():
                try:
                    await asyncio.shield(waiter)
                except asyncio.CancelledError:
                    continue
                except BaseException:
                    break
            raise

    def listen(self, callback: Callable[[OperationEvent], None], *, replay: bool = True) -> Operation[T]:
        delivery_lock = threading.Lock()
        buffered: list[OperationEvent] = []
        replaying = True

        def receive(event: OperationEvent) -> None:
            nonlocal replaying
            with delivery_lock:
                if replaying:
                    buffered.append(event)
                    return
            self._notify_listener(callback, event)

        with self._lock:
            history = tuple(self._events) if replay else ()
            if not self._state.terminal:
                self._listeners.append((callback, receive))
        for event in history:
            self._notify_listener(callback, event)
        while True:
            with delivery_lock:
                if not buffered:
                    replaying = False
                    break
                pending = tuple(buffered)
                buffered.clear()
            for event in pending:
                self._notify_listener(callback, event)
        return self

    def remove_listener(self, callback: Callable[[OperationEvent], None]) -> None:
        with self._lock:
            for entry in self._listeners:
                if entry[0] is callback:
                    self._listeners.remove(entry)
                    break

    async def events(self, *, replay: bool = True) -> AsyncIterator[OperationEvent]:
        loop = asyncio.get_running_loop()
        queue: asyncio.Queue[OperationEvent] = asyncio.Queue(maxsize=self._EVENT_HISTORY_LIMIT)

        def enqueue(event: OperationEvent) -> None:
            if queue.full():
                queue.get_nowait()
            queue.put_nowait(event)

        def receive(event: OperationEvent) -> None:
            loop.call_soon_threadsafe(enqueue, event)

        # Enqueue replay while holding the same lock used by emit(), then install
        # the live listener before releasing it.  This prevents a concurrent event
        # from overtaking replayed events.
        with self._lock:
            if replay:
                for event in self._events:
                    queue.put_nowait(event)
            terminal_without_replay = self._state.terminal and not replay
            if not self._state.terminal:
                self._listeners.append((receive, receive))
        if terminal_without_replay:
            return
        try:
            while True:
                event = await queue.get()
                yield event
                if event.state.terminal:
                    return
        finally:
            self.remove_listener(receive)

    def _emit(
        self,
        kind: OperationEventKind,
        message: str,
        *,
        stage: str | None = None,
        step_id: str | None = None,
        stream: str | None = None,
        line: str | None = None,
        current: int | None = None,
        maximum: int | None = None,
        environment: str | None = None,
    ) -> OperationEvent:
        with self._lock:
            publication = self._queue_event_locked(
                kind,
                message,
                stage=stage,
                step_id=step_id,
                stream=stream,
                line=line,
                current=current,
                maximum=maximum,
                environment=environment,
            )
        return self._finish_event_publication(publication)

    def _queue_event_locked(
        self,
        kind: OperationEventKind,
        message: str,
        *,
        stage: str | None = None,
        step_id: str | None = None,
        stream: str | None = None,
        line: str | None = None,
        current: int | None = None,
        maximum: int | None = None,
        environment: str | None = None,
    ) -> tuple[OperationEvent, threading.Event, bool, bool]:
        self._sequence += 1
        event = OperationEvent(
            sequence=self._sequence,
            timestamp=time.time(),
            operation_id=self._id,
            environment=environment if environment is not None else self._environment,
            kind=kind,
            state=self._state,
            stage=stage,
            message=message,
            step_id=step_id,
            stream=stream,
            line=line,
            current=current,
            maximum=maximum,
        )
        self._events.append(event)
        listeners = tuple(receiver for _, receiver in self._listeners)
        delivered = threading.Event()
        self._pending_notifications.append((event, listeners, delivered))

        current_thread_id = threading.get_ident()
        should_drain = self._notification_thread_id is None
        if should_drain:
            self._notification_thread_id = current_thread_id
        should_wait = not should_drain and self._notification_thread_id != current_thread_id
        return event, delivered, should_drain, should_wait

    def _finish_event_publication(
        self,
        publication: tuple[OperationEvent, threading.Event, bool, bool],
    ) -> OperationEvent:
        event, delivered, should_drain, should_wait = publication
        if should_drain:
            self._drain_notifications()
        elif should_wait:
            delivered.wait()
        return event

    def _drain_notifications(self) -> None:
        while True:
            with self._lock:
                if not self._pending_notifications:
                    self._notification_thread_id = None
                    return
                event, listeners, delivered = self._pending_notifications.popleft()
            try:
                for listener in listeners:
                    self._notify_listener(listener, event)
            finally:
                delivered.set()

    def _notify_listener(self, listener: Callable[[OperationEvent], None], event: OperationEvent) -> None:
        try:
            listener(event)
        except Exception:
            logger.exception("Operation listener failed")

    def _set_cancel_callback(self, callback: Callable[[], None]) -> None:
        with self._lock:
            self._cancel_callback = callback
            requested = self._cancellation_requested
        if requested:
            callback()

    def _seal_cancellation(self) -> bool:
        """Linearize final publication against cancellation.

        Returns ``False`` when cancellation already won.  Once this returns
        ``True``, subsequent cancellation requests are rejected and the runner
        must proceed directly to terminal success or failure.
        """

        with self._lock:
            if self._cancellation_requested:
                return False
            self._cancellation_sealed = True
            return True

    def _start_runner(self, runner: Callable[[], T], *, thread_name: str) -> None:
        with self._lock:
            if self._thread is not None:
                raise RuntimeError("Operation runner already started")
            self._thread = threading.Thread(target=self._run, args=(runner,), name=thread_name, daemon=True)
            self._thread.start()

    def _runs_on_current_thread(self) -> bool:
        with self._lock:
            return self._thread is threading.current_thread()

    def _run(self, runner: Callable[[], T]) -> None:
        if self.cancellation_requested:
            self._set_canceled()
            return
        self._set_running()
        try:
            result = runner()
        except OperationCanceled as error:
            self._set_canceled(error)
        except BaseException as error:
            self._set_failed(error)
        else:
            if self.cancellation_requested and not self._cancellation_sealed:
                self._set_canceled()
            else:
                self._set_completed(result)

    def _set_running(self) -> None:
        with self._lock:
            if self._state is not OperationState.PENDING:
                return
            self._state = OperationState.RUNNING
            publication = self._queue_event_locked(OperationEventKind.STATE, "Operation started")
        self._finish_event_publication(publication)

    def _set_completed(self, result: T) -> None:
        with self._lock:
            if self._state.terminal:
                return
            self._result = result
            self._state = OperationState.COMPLETED
            publication = self._queue_event_locked(OperationEventKind.STATE, "Operation completed")
        self._finish_event_publication(publication)
        self._done.set()

    def _set_failed(self, error: BaseException) -> None:
        with self._lock:
            if self._state.terminal:
                return
            self._exception = error
            self._state = OperationState.FAILED
            publication = self._queue_event_locked(OperationEventKind.STATE, str(error) or type(error).__name__)
        self._finish_event_publication(publication)
        self._done.set()

    def _set_canceled(self, error: OperationCanceled | None = None) -> None:
        with self._lock:
            if self._state.terminal:
                return
            self._exception = error or OperationCanceled(self.id)
            self._state = OperationState.CANCELED
            publication = self._queue_event_locked(OperationEventKind.STATE, "Operation canceled")
        self._finish_event_publication(publication)
        self._done.set()

OperationEvent dataclass

Source code in src/wetlands/operation.py

@dataclass(frozen=True)
class OperationEvent:
    sequence: int
    timestamp: float
    operation_id: str
    environment: str | None
    kind: OperationEventKind
    state: OperationState
    stage: str | None
    message: str
    step_id: str | None = None
    stream: str | None = None
    line: str | None = None
    current: int | None = None
    maximum: int | None = None

OperationEventKind

Bases: Enum

Source code in src/wetlands/operation.py

class OperationEventKind(enum.Enum):
    STATE = "state"
    STEP = "step"
    OUTPUT = "output"
    PROGRESS = "progress"
    CANCELLATION_REQUESTED = "cancellation_requested"
    CLEANUP = "cleanup"

OperationState

Bases: Enum

Source code in src/wetlands/operation.py

class OperationState(enum.Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELED = "canceled"

    @property
    def terminal(self) -> bool:
        return self in {
            OperationState.COMPLETED,
            OperationState.FAILED,
            OperationState.CANCELED,
        }

PreparationOperation

Bases: Operation[T]

Source code in src/wetlands/operation.py

class PreparationOperation(Operation[T]):
    pass

ProvisioningOperation

Bases: Operation[T]

Source code in src/wetlands/operation.py

class ProvisioningOperation(Operation[T]):
    pass

RemovalOperation

Bases: Operation[T]

An asynchronous managed-environment removal.

Source code in src/wetlands/operation.py

class RemovalOperation(Operation[T]):
    """An asynchronous managed-environment removal."""