diff --git a/.release-please-manifest.json b/.release-please-manifest.json index bfaab56f6..0c0c0c357 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -1,3 +1,3 @@ { - ".": "1.24.0" + ".": "1.25.0" } \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index de0a6c03f..ce5304ba8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,23 @@ # Changelog +## 1.25.0 (2026-07-24) + +Full Changelog: [v1.24.0...v1.25.0](https://github.com/runloopai/api-client-python/compare/v1.24.0...v1.25.0) + +### Features + +* **mux:** wait_for_eviction endpoint for flex eviction notifications ([#10325](https://github.com/runloopai/api-client-python/issues/10325)) ([c06794d](https://github.com/runloopai/api-client-python/commit/c06794d413f37be16ee5955fe31b07b92b8adbc8)) + + +### Bug Fixes + +* **mux:** field-level merge on network policy egress updates ([#10338](https://github.com/runloopai/api-client-python/issues/10338)) ([cfc97ef](https://github.com/runloopai/api-client-python/commit/cfc97ef55a65e88abb0d76f733cf51dc33823fb0)) + + +### Chores + +* **stainless:** generate watch_evictions as an SSE stream ([#10404](https://github.com/runloopai/api-client-python/issues/10404)) ([0413e7b](https://github.com/runloopai/api-client-python/commit/0413e7ba4cd10b919a3fc50e67e0238dc95f4b18)) + ## 1.24.0 (2026-07-22) Full Changelog: [v1.23.3...v1.24.0](https://github.com/runloopai/api-client-python/compare/v1.23.3...v1.24.0) diff --git a/api.md b/api.md index fc7f676a3..6becb60c5 100644 --- a/api.md +++ b/api.md @@ -245,7 +245,7 @@ Methods: - client.devboxes.suspend(id) -> DevboxView - client.devboxes.upload_file(id, \*\*params) -> object - client.devboxes.wait_for_command(execution_id, \*, devbox_id, \*\*params) -> DevboxAsyncExecutionDetailView -- client.devboxes.watch_evictions() -> DevboxEvictionEventView +- client.devboxes.watch_evictions() -> DevboxEvictionEventView - client.devboxes.write_file_contents(id, \*\*params) -> DevboxExecutionDetailView diff --git a/pyproject.toml b/pyproject.toml index 77d099e9e..2fe9fc3e2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "runloop_api_client" -version = "1.24.0" +version = "1.25.0" description = "The official Python library for the runloop API" dynamic = ["readme"] license = "MIT" diff --git a/src/runloop_api_client/_version.py b/src/runloop_api_client/_version.py index 5e32f9066..54818c6b0 100644 --- a/src/runloop_api_client/_version.py +++ b/src/runloop_api_client/_version.py @@ -1,4 +1,4 @@ # File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. __title__ = "runloop_api_client" -__version__ = "1.24.0" # x-release-please-version +__version__ = "1.25.0" # x-release-please-version diff --git a/src/runloop_api_client/resources/devboxes/devboxes.py b/src/runloop_api_client/resources/devboxes/devboxes.py index 0a3fdade5..19578d24c 100644 --- a/src/runloop_api_client/resources/devboxes/devboxes.py +++ b/src/runloop_api_client/resources/devboxes/devboxes.py @@ -1762,7 +1762,7 @@ def watch_evictions( and delivery is not guaranteed. """ return self._get( - "/v1/devboxes/watch_evictions", + "/v1/devboxes/evictions/watch", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), @@ -3464,7 +3464,7 @@ async def watch_evictions( and delivery is not guaranteed. """ return await self._get( - "/v1/devboxes/watch_evictions", + "/v1/devboxes/evictions/watch", options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), diff --git a/src/runloop_api_client/sdk/async_.py b/src/runloop_api_client/sdk/async_.py index 6319e02bd..aa2df16ae 100644 --- a/src/runloop_api_client/sdk/async_.py +++ b/src/runloop_api_client/sdk/async_.py @@ -46,6 +46,7 @@ from .async_devbox import AsyncDevbox from .async_scorer import AsyncScorer from .async_secret import AsyncSecret +from .async_eviction import shutdown_monitor_for as shutdown_eviction_monitor from .async_scenario import AsyncScenario from .async_snapshot import AsyncSnapshot from .async_benchmark import AsyncBenchmark @@ -1373,6 +1374,7 @@ def __init__( async def aclose(self) -> None: """Close the underlying HTTP client and release resources.""" + await shutdown_eviction_monitor(self.api) await self.api.close() async def __aenter__(self) -> "AsyncRunloopSDK": diff --git a/src/runloop_api_client/sdk/async_devbox.py b/src/runloop_api_client/sdk/async_devbox.py index bed785a7d..796732006 100644 --- a/src/runloop_api_client/sdk/async_devbox.py +++ b/src/runloop_api_client/sdk/async_devbox.py @@ -35,6 +35,7 @@ from ._helpers import filter_params from .._streaming import AsyncStream from ..lib.polling import PollingConfig +from .async_eviction import AsyncEvictionCallback, monitor_for as _eviction_monitor_for from ..types.devboxes import ExecutionUpdateChunk from .async_execution import AsyncExecution, _AsyncStreamingGroup from .async_execution_result import AsyncExecutionResult @@ -216,6 +217,25 @@ async def await_suspended(self, *, polling_config: PollingConfig | None = None) """ return await self._client.devboxes.await_suspended(self._id, polling_config=polling_config) + async def on_evict(self, callback: AsyncEvictionCallback) -> None: + """Register a callback fired once if this devbox gets a pending infrastructure eviction. + + The first ``on_evict`` across any devbox on this client opens a single + account-wide notification stream; it closes automatically once every + registered devbox has been notified. The callback runs at most once and is + invoked as ``callback(devbox, eviction_deadline_ms)`` — the devbox itself and + the Unix millisecond deadline by which it will be suspended. Use it to run + cleanup before the devbox is suspended. The callback may be sync or async. + + :param callback: Callable invoked with this devbox and its eviction deadline (ms). + :type callback: AsyncEvictionCallback + """ + await _eviction_monitor_for(self._client).register(self, callback) + + async def cancel_on_evict(self) -> None: + """Withdraw this devbox's eviction interest registered via :meth:`on_evict`.""" + await _eviction_monitor_for(self._client).unregister(self._id) + async def shutdown( self, **options: Unpack[LongRequestOptions], diff --git a/src/runloop_api_client/sdk/async_eviction.py b/src/runloop_api_client/sdk/async_eviction.py new file mode 100644 index 000000000..dd384a09c --- /dev/null +++ b/src/runloop_api_client/sdk/async_eviction.py @@ -0,0 +1,155 @@ +"""Async client-side eviction notification monitor. + +Async counterpart of :mod:`runloop_api_client.sdk.eviction`; see that module for the +delivery contract. Callbacks may be plain functions or coroutines. +""" + +from __future__ import annotations + +import asyncio +import inspect +import logging +from typing import TYPE_CHECKING, Dict, Tuple, Union, Callable, Optional, Awaitable +from weakref import WeakKeyDictionary + +if TYPE_CHECKING: + from ..types import DevboxEvictionEventView + from .._client import AsyncRunloop + from .._streaming import AsyncStream + from .async_devbox import AsyncDevbox + +AsyncEvictionCallback = Callable[["AsyncDevbox", int], Union[None, Awaitable[None]]] +"""Sync or async callable invoked once with the devbox and its eviction deadline (ms).""" + +_logger = logging.getLogger(__name__) + + +class AsyncEvictionMonitor: + """Async fan-out of account-wide eviction notifications to per-devbox callbacks.""" + + def __init__(self, client: "AsyncRunloop") -> None: + self._client = client + self._lock = asyncio.Lock() + self._callbacks: Dict[str, Tuple["AsyncDevbox", AsyncEvictionCallback]] = {} + self._task: Optional["asyncio.Task[None]"] = None + self._stream: Optional["AsyncStream[DevboxEvictionEventView]"] = None + + async def register(self, devbox: "AsyncDevbox", callback: AsyncEvictionCallback) -> None: + """Add ``devbox`` to the interest set, starting the stream task if idle.""" + async with self._lock: + self._callbacks[devbox.id] = (devbox, callback) + if self._task is None or self._task.done(): + self._task = asyncio.create_task(self._run(), name="runloop-eviction-monitor") + + async def unregister(self, devbox_id: str) -> None: + """Drop ``devbox_id``; close the stream if it was the last interested devbox.""" + async with self._lock: + self._callbacks.pop(devbox_id, None) + empty = not self._callbacks + if empty: + await self._close_stream() + + async def close(self) -> None: + """Clear all interest and tear down the stream.""" + async with self._lock: + self._callbacks.clear() + await self._close_stream() + + # Reconnect backoff bounds (seconds). The server force-closes the stream on purpose — on a + # leader change (FAILED_PRECONDITION) or a slow consumer (RESOURCE_EXHAUSTED) — and expects the + # client to reconnect and re-read the snapshot, which re-delivers anything missed. So a single + # stream ending is normal, not terminal: reconnect until no devbox is still interested. + _RECONNECT_BACKOFF_INITIAL_S = 0.5 + _RECONNECT_BACKOFF_MAX_S = 30.0 + + async def _run(self) -> None: + backoff = self._RECONNECT_BACKOFF_INITIAL_S + try: + while True: + async with self._lock: + if not self._callbacks: + return + try: + # Force the SSE Accept header: the endpoint only streams for + # text/event-stream; the generated client's default (application/json) gets an + # empty text/plain response, so the feed would silently deliver nothing. + stream = await self._client.devboxes.watch_evictions(extra_headers={"Accept": "text/event-stream"}) + async with self._lock: + self._stream = stream + _logger.debug("async eviction monitor stream connected") + async with stream: + async for event in stream: + _logger.debug("async eviction monitor received event for %s", event.devbox_id) + await self._dispatch(event) + async with self._lock: + if not self._callbacks: + return + # Clean end (server closed the stream): reset backoff and reconnect if still + # interested. The reconnect's snapshot re-delivers still-pending evictions. + backoff = self._RECONNECT_BACKOFF_INITIAL_S + _logger.debug("async eviction monitor stream ended; reconnecting") + except asyncio.CancelledError: + raise + except Exception: + # An intentional teardown (close/unregister clears the interest set, then closes + # the stream) surfaces here as a read error — exit quietly in that case. + async with self._lock: + interested = bool(self._callbacks) + if not interested: + return + # Routine: the server force-closes on leader change / slow consumer, and a + # long-lived stream can drop (e.g. an HTTP/2 disconnect). Reconnecting recovers + # it, so keep this at debug to avoid log spam. + _logger.debug("async eviction monitor stream error; reconnecting", exc_info=True) + async with self._lock: + if not self._callbacks: + return + await asyncio.sleep(backoff) + backoff = min(backoff * 2, self._RECONNECT_BACKOFF_MAX_S) + finally: + async with self._lock: + self._stream = None + self._task = None + + async def _dispatch(self, event: "DevboxEvictionEventView") -> None: + async with self._lock: + entry = self._callbacks.pop(event.devbox_id, None) + if entry is None: + return + devbox, callback = entry + try: + result = callback(devbox, event.eviction_deadline_ms) + if inspect.isawaitable(result): + await result + except Exception: + _logger.exception("error in eviction callback for devbox %s", event.devbox_id) + + async def _close_stream(self) -> None: + async with self._lock: + stream = self._stream + self._stream = None + if stream is not None: + try: + await stream.close() + except Exception: + _logger.debug("error closing eviction stream", exc_info=True) + + +# asyncio is single-threaded, so the registry needs no lock. +_monitors: "WeakKeyDictionary[AsyncRunloop, AsyncEvictionMonitor]" = WeakKeyDictionary() + + +def monitor_for(client: "AsyncRunloop") -> AsyncEvictionMonitor: + """Return the shared :class:`AsyncEvictionMonitor` for ``client``, creating it once.""" + monitor = _monitors.get(client) + if monitor is None: + monitor = AsyncEvictionMonitor(client) + _monitors[client] = monitor + return monitor + + +async def shutdown_monitor_for(client: "AsyncRunloop") -> None: + """Tear down the shared monitor for ``client`` if one exists.""" + monitor = _monitors.pop(client, None) + if monitor is not None: + await monitor.close() diff --git a/src/runloop_api_client/sdk/devbox.py b/src/runloop_api_client/sdk/devbox.py index c97ea2682..3c7d6123a 100644 --- a/src/runloop_api_client/sdk/devbox.py +++ b/src/runloop_api_client/sdk/devbox.py @@ -34,6 +34,7 @@ from .._types import omit from .._client import Runloop from ._helpers import filter_params +from .eviction import EvictionCallback, monitor_for as _eviction_monitor_for from .execution import Execution, _StreamingGroup from .._streaming import Stream from ..lib.polling import PollingConfig @@ -219,6 +220,25 @@ def await_suspended(self, *, polling_config: PollingConfig | None = None) -> Dev """ return self._client.devboxes.await_suspended(self._id, polling_config=polling_config) + def on_evict(self, callback: EvictionCallback) -> None: + """Register a callback fired once if this devbox gets a pending infrastructure eviction. + + The first ``on_evict`` across any devbox on this client opens a single + account-wide notification stream; it closes automatically once every + registered devbox has been notified. The callback runs at most once and is + invoked as ``callback(devbox, eviction_deadline_ms)`` — the devbox itself and + the Unix millisecond deadline by which it will be suspended. Use it to run + cleanup before the devbox is suspended. + + :param callback: Callable invoked with this devbox and its eviction deadline (ms). + :type callback: EvictionCallback + """ + _eviction_monitor_for(self._client).register(self, callback) + + def cancel_on_evict(self) -> None: + """Withdraw this devbox's eviction interest registered via :meth:`on_evict`.""" + _eviction_monitor_for(self._client).unregister(self._id) + def shutdown( self, **options: Unpack[LongRequestOptions], diff --git a/src/runloop_api_client/sdk/eviction.py b/src/runloop_api_client/sdk/eviction.py new file mode 100644 index 000000000..476062f6b --- /dev/null +++ b/src/runloop_api_client/sdk/eviction.py @@ -0,0 +1,177 @@ +"""Client-side eviction notification monitor. + +Backs :meth:`runloop_api_client.sdk.devbox.Devbox.on_evict`. A single account-wide +SSE stream (the generated ``devboxes.watch_evictions`` endpoint) is opened lazily +the moment the first devbox registers interest, and torn down as soon as the last +interested devbox has been notified. + +Delivery contract: + +* The server replays every currently-pending eviction on connect, so a devbox that + registers after its eviction was scheduled is still notified. +* Notifications for devboxes not in the interest set are discarded. +* A devbox is removed from the interest set *before* its callback runs, so the + callback fires at most once even if the server repeats the notification. +""" + +from __future__ import annotations + +import time +import logging +import threading +from typing import TYPE_CHECKING, Dict, Tuple, Callable, Optional +from weakref import WeakKeyDictionary + +if TYPE_CHECKING: + from ..types import DevboxEvictionEventView + from .devbox import Devbox + from .._client import Runloop + from .._streaming import Stream + +EvictionCallback = Callable[["Devbox", int], None] +"""Invoked once when the devbox it was registered for has a pending eviction. + +Receives the :class:`~runloop_api_client.sdk.devbox.Devbox` and the eviction +deadline as a Unix timestamp in milliseconds. +""" + +_logger = logging.getLogger(__name__) + + +class EvictionMonitor: + """Fans account-wide eviction notifications out to per-devbox callbacks. + + One monitor is shared by every :class:`~runloop_api_client.sdk.devbox.Devbox` + built from the same generated client (see :func:`monitor_for`), so all + registered devboxes are served by a single SSE connection. + """ + + def __init__(self, client: "Runloop") -> None: + self._client = client + self._lock = threading.Lock() + self._callbacks: Dict[str, Tuple["Devbox", EvictionCallback]] = {} + self._thread: Optional[threading.Thread] = None + self._stream: Optional["Stream[DevboxEvictionEventView]"] = None + + def register(self, devbox: "Devbox", callback: EvictionCallback) -> None: + """Add ``devbox`` to the interest set, starting the stream if idle.""" + with self._lock: + self._callbacks[devbox.id] = (devbox, callback) + if self._thread is None or not self._thread.is_alive(): + self._thread = threading.Thread( + target=self._run, + name="runloop-eviction-monitor", + daemon=True, + ) + self._thread.start() + + def unregister(self, devbox_id: str) -> None: + """Drop ``devbox_id``; close the stream if it was the last interested devbox.""" + with self._lock: + self._callbacks.pop(devbox_id, None) + empty = not self._callbacks + if empty: + self._close_stream() + + def close(self) -> None: + """Clear all interest and tear down the stream.""" + with self._lock: + self._callbacks.clear() + self._close_stream() + + # Reconnect backoff bounds (seconds). The server force-closes the stream on purpose — on a + # leader change (FAILED_PRECONDITION) or a slow consumer (RESOURCE_EXHAUSTED) — and expects the + # client to reconnect and re-read the snapshot, which re-delivers anything missed. So a single + # stream ending is normal, not terminal: reconnect until no devbox is still interested. + _RECONNECT_BACKOFF_INITIAL_S = 0.5 + _RECONNECT_BACKOFF_MAX_S = 30.0 + + def _run(self) -> None: + backoff = self._RECONNECT_BACKOFF_INITIAL_S + try: + while True: + with self._lock: + if not self._callbacks: + return + try: + # Force the SSE Accept header: the endpoint only streams for + # text/event-stream; the generated client's default (application/json) gets an + # empty text/plain response, so the feed would silently deliver nothing. + stream = self._client.devboxes.watch_evictions(extra_headers={"Accept": "text/event-stream"}) + with self._lock: + self._stream = stream + _logger.debug("eviction monitor stream connected") + with stream: + for event in stream: + _logger.debug("eviction monitor received event for %s", event.devbox_id) + self._dispatch(event) + with self._lock: + if not self._callbacks: + return + # Clean end (server closed the stream): reset backoff and reconnect if still + # interested. The reconnect's snapshot re-delivers still-pending evictions. + backoff = self._RECONNECT_BACKOFF_INITIAL_S + _logger.debug("eviction monitor stream ended; reconnecting") + except Exception: + # An intentional teardown (close/unregister clears the interest set, then closes + # the stream) surfaces here as a read error — exit quietly in that case. + with self._lock: + interested = bool(self._callbacks) + if not interested: + return + # Routine: the server force-closes on leader change / slow consumer, and a + # long-lived stream can drop (e.g. an HTTP/2 disconnect). Reconnecting recovers + # it, so keep this at debug to avoid log spam. + _logger.debug("eviction monitor stream error; reconnecting", exc_info=True) + with self._lock: + if not self._callbacks: + return + time.sleep(backoff) + backoff = min(backoff * 2, self._RECONNECT_BACKOFF_MAX_S) + finally: + with self._lock: + self._stream = None + self._thread = None + + def _dispatch(self, event: "DevboxEvictionEventView") -> None: + with self._lock: + entry = self._callbacks.pop(event.devbox_id, None) + if entry is None: + return + devbox, callback = entry + try: + callback(devbox, event.eviction_deadline_ms) + except Exception: + _logger.exception("error in eviction callback for devbox %s", event.devbox_id) + + def _close_stream(self) -> None: + with self._lock: + stream = self._stream + self._stream = None + if stream is not None: + try: + stream.close() + except Exception: + _logger.debug("error closing eviction stream", exc_info=True) + + +_monitors: "WeakKeyDictionary[Runloop, EvictionMonitor]" = WeakKeyDictionary() +_monitors_lock = threading.Lock() + + +def monitor_for(client: "Runloop") -> EvictionMonitor: + """Return the shared :class:`EvictionMonitor` for ``client``, creating it once.""" + with _monitors_lock: + monitor = _monitors.get(client) + if monitor is None: + monitor = EvictionMonitor(client) + _monitors[client] = monitor + return monitor + + +def shutdown_monitor_for(client: "Runloop") -> None: + """Tear down the shared monitor for ``client`` if one exists.""" + with _monitors_lock: + monitor = _monitors.pop(client, None) + if monitor is not None: + monitor.close() diff --git a/src/runloop_api_client/sdk/sync.py b/src/runloop_api_client/sdk/sync.py index 7d3afcb24..957299a7f 100644 --- a/src/runloop_api_client/sdk/sync.py +++ b/src/runloop_api_client/sdk/sync.py @@ -45,6 +45,7 @@ from .._types import Timeout, NotGiven, not_given from .._client import DEFAULT_MAX_RETRIES, Runloop from ._helpers import detect_content_type +from .eviction import shutdown_monitor_for as shutdown_eviction_monitor from .scenario import Scenario from .snapshot import Snapshot from .benchmark import Benchmark @@ -1398,6 +1399,7 @@ def __init__( def close(self) -> None: """Close the underlying HTTP client and release resources.""" + shutdown_eviction_monitor(self.api) self.api.close() def __enter__(self) -> "RunloopSDK":