fix(devbox): reconcile on_evict with generated stream and 2-arg callback

- Use the real generated event type DevboxEvictionEventView (the endpoint's view)
  in place of the assumed DevboxEvictionEvent.
- Invoke the callback as callback(devbox, eviction_deadline_ms): the Devbox object
  and the Unix millisecond deadline, instead of passing the raw event.

Requires the main-repo stainless config change that generates watch_evictions as
Stream[DevboxEvictionEventView] (runloopai/runloop#10404).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: claude

src/runloop_api_client/sdk/async_.py

@@ -45,8 +45,8 @@
 from .async_agent import AsyncAgent
 from .async_devbox import AsyncDevbox
 from .async_scorer import AsyncScorer
-from .async_eviction import shutdown_monitor_for as shutdown_eviction_monitor
 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

src/runloop_api_client/sdk/async_devbox.py

@@ -33,9 +33,9 @@
 from .._types import omit
 from .._client import AsyncRunloop
 from ._helpers import filter_params
-from .async_eviction import AsyncEvictionCallback, monitor_for as _eviction_monitor_for
 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
@@ -222,14 +222,15 @@ async def on_evict(self, callback: AsyncEvictionCallback) -> None:
 
         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
-        receives the eviction event (with its ``eviction_deadline_ms``) — use it to
-        run cleanup before the devbox is suspended. The callback may be sync or async.
+        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 the eviction event for this devbox.
+        :param callback: Callable invoked with this devbox and its eviction deadline (ms).
         :type callback: AsyncEvictionCallback
         """
-        await _eviction_monitor_for(self._client).register(self._id, callback)
+        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`."""

src/runloop_api_client/sdk/async_eviction.py

@@ -9,20 +9,17 @@
 import asyncio
 import inspect
 import logging
-from typing import TYPE_CHECKING, Dict, Union, Callable, Optional, Awaitable
+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
 
-    # Assumed Stainless-generated SSE event type for ``watch_evictions``; carries
-    # ``devbox_id`` and ``eviction_deadline_ms``. Reconcile the import once the
-    # generated code lands.
-    from ..types import DevboxEvictionEvent
-
-AsyncEvictionCallback = Callable[["DevboxEvictionEvent"], Union[None, Awaitable[None]]]
-"""Sync or async callable invoked once with the eviction event for its devbox."""
+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__)
 
@@ -33,14 +30,14 @@ class AsyncEvictionMonitor:
     def __init__(self, client: "AsyncRunloop") -> None:
         self._client = client
         self._lock = asyncio.Lock()
-        self._callbacks: Dict[str, AsyncEvictionCallback] = {}
+        self._callbacks: Dict[str, Tuple["AsyncDevbox", AsyncEvictionCallback]] = {}
         self._task: Optional["asyncio.Task[None]"] = None
-        self._stream: Optional["AsyncStream[DevboxEvictionEvent]"] = None
+        self._stream: Optional["AsyncStream[DevboxEvictionEventView]"] = None
 
-    async def register(self, devbox_id: str, callback: AsyncEvictionCallback) -> None:
-        """Add ``devbox_id`` to the interest set, starting the stream task if idle."""
+    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] = callback
+            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")
 
@@ -78,13 +75,14 @@ async def _run(self) -> None:
                 self._stream = None
                 self._task = None
 
-    async def _dispatch(self, event: "DevboxEvictionEvent") -> None:
+    async def _dispatch(self, event: "DevboxEvictionEventView") -> None:
         async with self._lock:
-            callback = self._callbacks.pop(event.devbox_id, None)
-        if callback is None:
+            entry = self._callbacks.pop(event.devbox_id, None)
+        if entry is None:
             return
+        devbox, callback = entry
         try:
-            result = callback(event)
+            result = callback(devbox, event.eviction_deadline_ms)
             if inspect.isawaitable(result):
                 await result
         except Exception:

src/runloop_api_client/sdk/devbox.py

@@ -33,8 +33,8 @@
 )
 from .._types import omit
 from .._client import Runloop
-from .eviction import EvictionCallback, monitor_for as _eviction_monitor_for
 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
@@ -225,14 +225,15 @@ def on_evict(self, callback: EvictionCallback) -> None:
 
         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
-        receives the eviction event (with its ``eviction_deadline_ms``) — use it to
-        run cleanup before the devbox is suspended.
+        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 the eviction event for this devbox.
+        :param callback: Callable invoked with this devbox and its eviction deadline (ms).
         :type callback: EvictionCallback
         """
-        _eviction_monitor_for(self._client).register(self._id, callback)
+        _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`."""

src/runloop_api_client/sdk/eviction.py

@@ -18,20 +18,21 @@
 
 import logging
 import threading
-from typing import TYPE_CHECKING, Dict, Callable, Optional
+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
 
-    # Assumed Stainless-generated SSE event type for ``watch_evictions``; carries
-    # ``devbox_id`` and ``eviction_deadline_ms``. Reconcile the import once the
-    # generated code lands.
-    from ..types import DevboxEvictionEvent
+EvictionCallback = Callable[["Devbox", int], None]
+"""Invoked once when the devbox it was registered for has a pending eviction.
 
-EvictionCallback = Callable[["DevboxEvictionEvent"], None]
-"""Invoked once with the eviction event for the devbox it was registered for."""
+Receives the :class:`~runloop_api_client.sdk.devbox.Devbox` and the eviction
+deadline as a Unix timestamp in milliseconds.
+"""
 
 _logger = logging.getLogger(__name__)
 
@@ -47,14 +48,14 @@ class EvictionMonitor:
     def __init__(self, client: "Runloop") -> None:
         self._client = client
         self._lock = threading.Lock()
-        self._callbacks: Dict[str, EvictionCallback] = {}
+        self._callbacks: Dict[str, Tuple["Devbox", EvictionCallback]] = {}
         self._thread: Optional[threading.Thread] = None
-        self._stream: Optional["Stream[DevboxEvictionEvent]"] = None
+        self._stream: Optional["Stream[DevboxEvictionEventView]"] = None
 
-    def register(self, devbox_id: str, callback: EvictionCallback) -> None:
-        """Add ``devbox_id`` to the interest set, starting the stream if idle."""
+    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] = callback
+            self._callbacks[devbox.id] = (devbox, callback)
             if self._thread is None or not self._thread.is_alive():
                 self._thread = threading.Thread(
                     target=self._run,
@@ -95,13 +96,14 @@ def _run(self) -> None:
                 self._stream = None
                 self._thread = None
 
-    def _dispatch(self, event: "DevboxEvictionEvent") -> None:
+    def _dispatch(self, event: "DevboxEvictionEventView") -> None:
         with self._lock:
-            callback = self._callbacks.pop(event.devbox_id, None)
-        if callback is None:
+            entry = self._callbacks.pop(event.devbox_id, None)
+        if entry is None:
             return
+        devbox, callback = entry
         try:
-            callback(event)
+            callback(devbox, event.eviction_deadline_ms)
         except Exception:
             _logger.exception("error in eviction callback for devbox %s", event.devbox_id)
 

src/runloop_api_client/sdk/sync.py

@@ -41,11 +41,11 @@
 )
 from .devbox import Devbox
 from .scorer import Scorer
-from .eviction import shutdown_monitor_for as shutdown_eviction_monitor
 from .secret import Secret
 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