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/.stats.yml b/.stats.yml index 4e9c9d433..10c388023 100644 --- a/.stats.yml +++ b/.stats.yml @@ -1,4 +1,4 @@ -configured_endpoints: 119 -openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/runloop-ai/runloop-0b7cd0c2fc193b18189cd7f44cf45ece7726b5d485fb72577f7d235266432ea0.yml -openapi_spec_hash: 78c340dbfb9d3d58b24ef318fc2a657b -config_hash: 218b8d25038e627faab98532392ee9a0 +configured_endpoints: 120 +openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/runloop-ai/runloop-6fb96a2de54d260cad2b76366cada4eda15335275dbab7f6c108c1bb29acf309.yml +openapi_spec_hash: 6044f4a4da13881c5f82fcc394c1f43b +config_hash: 92c04f0b12249cb2d2bcf9e503a9cbf9 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 b7db70b09..6becb60c5 100644 --- a/api.md +++ b/api.md @@ -201,6 +201,7 @@ Types: ```python from runloop_api_client.types import ( DevboxAsyncExecutionDetailView, + DevboxEvictionEventView, DevboxExecutionDetailView, DevboxKillExecutionRequest, DevboxListView, @@ -244,6 +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.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 f100014f8..19578d24c 100644 --- a/src/runloop_api_client/resources/devboxes/devboxes.py +++ b/src/runloop_api_client/resources/devboxes/devboxes.py @@ -69,6 +69,7 @@ LONG_POLL_CLIENT_BUFFER_SECONDS, EXEC_LONG_POLL_SERVER_MAX_SECONDS, ) +from ..._streaming import Stream, AsyncStream from ...pagination import ( SyncDevboxesCursorIDPage, AsyncDevboxesCursorIDPage, @@ -94,6 +95,7 @@ from ...types.shared_params.mount import Mount from ...types.devbox_snapshot_view import DevboxSnapshotView from ...types.shared.launch_parameters import LaunchParameters as SharedLaunchParameters +from ...types.devbox_eviction_event_view import DevboxEvictionEventView from ...types.devbox_resource_usage_view import DevboxResourceUsageView from ...types.devbox_execution_detail_view import DevboxExecutionDetailView from ...types.devbox_create_ssh_key_response import DevboxCreateSSHKeyResponse @@ -1742,6 +1744,33 @@ def wait_for_command( cast_to=DevboxAsyncExecutionDetailView, ) + def watch_evictions( + self, + *, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> Stream[DevboxEvictionEventView]: + """ + Subscribe, via server-sent events, to pending infrastructure evictions for every + Devbox in the account. On connect the stream emits one event per Devbox that + currently has a pending eviction, then one event as each further eviction is + scheduled. Best-effort and advisory: a Devbox stays running until its deadline, + and delivery is not guaranteed. + """ + return self._get( + "/v1/devboxes/evictions/watch", + options=make_request_options( + extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout + ), + cast_to=DevboxEvictionEventView, + stream=True, + stream_cls=Stream[DevboxEvictionEventView], + ) + def write_file_contents( self, id: str, @@ -3417,6 +3446,33 @@ async def wait_for_command( cast_to=DevboxAsyncExecutionDetailView, ) + async def watch_evictions( + self, + *, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> AsyncStream[DevboxEvictionEventView]: + """ + Subscribe, via server-sent events, to pending infrastructure evictions for every + Devbox in the account. On connect the stream emits one event per Devbox that + currently has a pending eviction, then one event as each further eviction is + scheduled. Best-effort and advisory: a Devbox stays running until its deadline, + and delivery is not guaranteed. + """ + return await self._get( + "/v1/devboxes/evictions/watch", + options=make_request_options( + extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout + ), + cast_to=DevboxEvictionEventView, + stream=True, + stream_cls=AsyncStream[DevboxEvictionEventView], + ) + async def write_file_contents( self, id: str, @@ -3555,6 +3611,9 @@ def __init__(self, devboxes: DevboxesResource) -> None: self.wait_for_command = to_raw_response_wrapper( devboxes.wait_for_command, ) + self.watch_evictions = to_raw_response_wrapper( + devboxes.watch_evictions, + ) self.write_file_contents = to_raw_response_wrapper( devboxes.write_file_contents, ) @@ -3651,6 +3710,9 @@ def __init__(self, devboxes: AsyncDevboxesResource) -> None: self.wait_for_command = async_to_raw_response_wrapper( devboxes.wait_for_command, ) + self.watch_evictions = async_to_raw_response_wrapper( + devboxes.watch_evictions, + ) self.write_file_contents = async_to_raw_response_wrapper( devboxes.write_file_contents, ) @@ -3747,6 +3809,9 @@ def __init__(self, devboxes: DevboxesResource) -> None: self.wait_for_command = to_streamed_response_wrapper( devboxes.wait_for_command, ) + self.watch_evictions = to_streamed_response_wrapper( + devboxes.watch_evictions, + ) self.write_file_contents = to_streamed_response_wrapper( devboxes.write_file_contents, ) @@ -3843,6 +3908,9 @@ def __init__(self, devboxes: AsyncDevboxesResource) -> None: self.wait_for_command = async_to_streamed_response_wrapper( devboxes.wait_for_command, ) + self.watch_evictions = async_to_streamed_response_wrapper( + devboxes.watch_evictions, + ) self.write_file_contents = async_to_streamed_response_wrapper( devboxes.write_file_contents, ) diff --git a/src/runloop_api_client/resources/network_policies.py b/src/runloop_api_client/resources/network_policies.py index b02062311..135eabea6 100644 --- a/src/runloop_api_client/resources/network_policies.py +++ b/src/runloop_api_client/resources/network_policies.py @@ -191,7 +191,8 @@ def update( ) -> NetworkPolicyView: """Update an existing NetworkPolicy. - All fields are optional. + All fields are optional - null fields preserve + existing values, provided fields replace entirely. Args: allow_agent_gateway: If true, allows devbox egress to the agent gateway. @@ -525,7 +526,8 @@ async def update( ) -> NetworkPolicyView: """Update an existing NetworkPolicy. - All fields are optional. + All fields are optional - null fields preserve + existing values, provided fields replace entirely. Args: allow_agent_gateway: If true, allows devbox egress to the agent gateway. 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": diff --git a/src/runloop_api_client/types/__init__.py b/src/runloop_api_client/types/__init__.py index 4abd94f92..0bda186a3 100644 --- a/src/runloop_api_client/types/__init__.py +++ b/src/runloop_api_client/types/__init__.py @@ -110,6 +110,7 @@ from .scenario_start_run_params import ScenarioStartRunParams as ScenarioStartRunParams from .benchmark_start_run_params import BenchmarkStartRunParams as BenchmarkStartRunParams from .blueprint_build_parameters import BlueprintBuildParameters as BlueprintBuildParameters +from .devbox_eviction_event_view import DevboxEvictionEventView as DevboxEvictionEventView from .devbox_execute_sync_params import DevboxExecuteSyncParams as DevboxExecuteSyncParams from .devbox_resource_usage_view import DevboxResourceUsageView as DevboxResourceUsageView from .gateway_config_list_params import GatewayConfigListParams as GatewayConfigListParams diff --git a/src/runloop_api_client/types/devbox_eviction_event_view.py b/src/runloop_api_client/types/devbox_eviction_event_view.py new file mode 100644 index 000000000..11d646c27 --- /dev/null +++ b/src/runloop_api_client/types/devbox_eviction_event_view.py @@ -0,0 +1,16 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from .._models import BaseModel + +__all__ = ["DevboxEvictionEventView"] + + +class DevboxEvictionEventView(BaseModel): + devbox_id: str + """The ID of the Devbox with a pending eviction.""" + + eviction_deadline_ms: int + """Unix timestamp (milliseconds) after which the Devbox will be suspended. + + Advisory and best-effort. + """ diff --git a/tests/api_resources/test_devboxes.py b/tests/api_resources/test_devboxes.py index 6c3e89c8f..0d9495f80 100644 --- a/tests/api_resources/test_devboxes.py +++ b/tests/api_resources/test_devboxes.py @@ -1190,6 +1190,30 @@ def test_path_params_wait_for_command(self, client: Runloop) -> None: statuses=["queued"], ) + @parametrize + def test_method_watch_evictions(self, client: Runloop) -> None: + devbox_stream = client.devboxes.watch_evictions() + devbox_stream.response.close() + + @parametrize + def test_raw_response_watch_evictions(self, client: Runloop) -> None: + response = client.devboxes.with_raw_response.watch_evictions() + + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + stream = response.parse() + stream.close() + + @parametrize + def test_streaming_response_watch_evictions(self, client: Runloop) -> None: + with client.devboxes.with_streaming_response.watch_evictions() as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + stream = response.parse() + stream.close() + + assert cast(Any, response.is_closed) is True + @parametrize def test_method_write_file_contents(self, client: Runloop) -> None: devbox = client.devboxes.write_file_contents( @@ -2859,6 +2883,30 @@ async def test_path_params_wait_for_command(self, async_client: AsyncRunloop) -> statuses=["queued"], ) + @parametrize + async def test_method_watch_evictions(self, async_client: AsyncRunloop) -> None: + devbox_stream = await async_client.devboxes.watch_evictions() + await devbox_stream.response.aclose() + + @parametrize + async def test_raw_response_watch_evictions(self, async_client: AsyncRunloop) -> None: + response = await async_client.devboxes.with_raw_response.watch_evictions() + + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + stream = await response.parse() + await stream.close() + + @parametrize + async def test_streaming_response_watch_evictions(self, async_client: AsyncRunloop) -> None: + async with async_client.devboxes.with_streaming_response.watch_evictions() as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + stream = await response.parse() + await stream.close() + + assert cast(Any, response.is_closed) is True + @parametrize async def test_method_write_file_contents(self, async_client: AsyncRunloop) -> None: devbox = await async_client.devboxes.write_file_contents(