fix(python): enforce lifecycle wait deadlines

architecture/gateway.md

@@ -66,6 +66,8 @@ it does not establish that a mutation is safe to repeat. SDKs retain the origina
 transport status, metadata, and unknown details alongside decoded fields.
 SDK deletion waits recognize missing-resource status through typed error wrappers
 without suppressing other failures.
+Python SDK lifecycle waits use monotonic deadlines and bound each poll RPC and
+sleep by the remaining wait budget, while retaining shorter per-call timeouts.
 
 Ordinary user-callable unary mutations explicitly opt into durable request
 admission when the client supplies a UUID. Typed adapters

docs/sdk/python.mdx

@@ -107,6 +107,11 @@ workspace for repeated exec, stop, start, and delete operations. List methods
 return lazy `Pager` instances; use `list_all()` only when you want to fetch the
 complete collection.
 
+`wait_ready()`, `wait_stopped()`, and `wait_deleted()` apply `timeout_seconds`
+to the whole wait, including lookup RPCs and polling delays. Each lookup also
+respects a shorter client `timeout`. Changes to the system clock do not extend
+or shorten the wait.
+
 ## Next Steps
 
 - Review [Manage Sandboxes](/how-it-works/sandboxes/overview) for labels, templates, services, and Python SDK examples.

python/openshell/sandbox.py

@@ -913,12 +913,15 @@ def sandbox_templates(self) -> SandboxTemplateClient:
         return SandboxTemplateClient(self._channel, timeout=self._timeout)
 
     def get(self, name: str, *, workspace: str) -> SandboxRef:
+        return self._get(name, workspace=workspace, timeout=self._timeout)
+
+    def _get(self, name: str, *, workspace: str, timeout: float) -> SandboxRef:
         response = self._stub.GetSandbox(
             openshell_pb2.GetSandboxRequest(
                 workspace_scope=_workspace_scope(workspace),
                 name=name,
             ),
-            timeout=self._timeout,
+            timeout=timeout,
         )
         return _sandbox_ref(response.sandbox)
 
@@ -1075,20 +1078,28 @@ def wait_deleted(
         timeout_seconds: float = 60.0,
         expected_sandbox_id: str | None = None,
     ) -> None:
-        deadline = time.time() + timeout_seconds
-        while time.time() < deadline:
+        deadline = time.monotonic() + timeout_seconds
+        while (remaining := deadline - time.monotonic()) > 0:
             try:
-                current = self.get(name, workspace=workspace)
+                current = self._get(
+                    name, workspace=workspace, timeout=min(self._timeout, remaining)
+                )
                 if (
                     expected_sandbox_id is not None
                     and current.id != expected_sandbox_id
                 ):
                     return
             except grpc.RpcError as exc:
-                if getattr(exc, "code", lambda: None)() == grpc.StatusCode.NOT_FOUND:
+                code = getattr(exc, "code", lambda: None)()
+                if code == grpc.StatusCode.NOT_FOUND:
                     return
+                if (
+                    code == grpc.StatusCode.DEADLINE_EXCEEDED
+                    and time.monotonic() >= deadline
+                ):
+                    break
                 raise
-            time.sleep(1)
+            time.sleep(min(1, max(0, deadline - time.monotonic())))
         raise SandboxError(f"sandbox {name} was not deleted within timeout")
 
     def wait_ready(
@@ -1122,9 +1133,20 @@ def _wait_for_phase(
         target_name: str,
         timeout_seconds: float,
     ) -> SandboxRef:
-        deadline = time.time() + timeout_seconds
-        while time.time() < deadline:
-            sandbox = self.get(name, workspace=workspace)
+        deadline = time.monotonic() + timeout_seconds
+        while (remaining := deadline - time.monotonic()) > 0:
+            try:
+                sandbox = self._get(
+                    name, workspace=workspace, timeout=min(self._timeout, remaining)
+                )
+            except grpc.RpcError as exc:
+                if (
+                    getattr(exc, "code", lambda: None)()
+                    == grpc.StatusCode.DEADLINE_EXCEEDED
+                    and time.monotonic() >= deadline
+                ):
+                    break
+                raise
             if sandbox.status.phase == target_phase:
                 return sandbox
             if (
@@ -1139,7 +1161,7 @@ def _wait_for_phase(
                 raise SandboxError(f"sandbox {name} stopped before becoming ready")
             if sandbox.status.phase == openshell_pb2.SANDBOX_PHASE_ERROR:
                 raise SandboxError(f"sandbox {name} entered error phase")
-            time.sleep(1)
+            time.sleep(min(1, max(0, deadline - time.monotonic())))
         raise SandboxError(f"sandbox {name} was not {target_name} within timeout")
 
     def exec_stream(

python/openshell/sandbox_wait_test.py

@@ -0,0 +1,164 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from concurrent import futures
+from types import SimpleNamespace
+
+import grpc
+import pytest
+
+import openshell.sandbox as sandbox_module
+from openshell._proto import openshell_pb2
+from openshell.sandbox import SandboxClient, SandboxError
+
+
+class Clock:
+    def __init__(self) -> None:
+        self.elapsed = 0.0
+        self.wall_offset = 0.0
+
+    def monotonic(self) -> float:
+        return self.elapsed
+
+    def time(self) -> float:
+        return self.elapsed + self.wall_offset
+
+    def sleep(self, seconds: float) -> None:
+        assert seconds >= 0
+        self.elapsed += seconds
+
+
+@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"])
+@pytest.mark.parametrize("timeout_seconds", [0, -1])
+def test_expired_wait_does_not_poll(monkeypatch, method, timeout_seconds):
+    def get(_request, timeout):
+        pytest.fail(f"unexpected lookup with timeout {timeout}")
+
+    with SandboxClient("localhost:1") as client:
+        monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get))
+        with pytest.raises(SandboxError, match="within timeout"):
+            getattr(client, method)(
+                "job", workspace="team", timeout_seconds=timeout_seconds
+            )
+
+
+@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"])
+@pytest.mark.parametrize(
+    "code", [grpc.StatusCode.DEADLINE_EXCEEDED, grpc.StatusCode.PERMISSION_DENIED]
+)
+def test_wait_preserves_lookup_errors_before_deadline(monkeypatch, method, code):
+    class LookupError(grpc.RpcError):
+        def code(self):
+            return code
+
+    error = LookupError()
+
+    def get(_request, timeout):
+        assert 0 < timeout <= 0.1
+        raise error
+
+    with SandboxClient("localhost:1", timeout=0.1) as client:
+        monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get))
+        with pytest.raises(LookupError) as caught:
+            getattr(client, method)("job", workspace="team", timeout_seconds=10)
+        assert caught.value is error
+
+
+@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"])
+@pytest.mark.parametrize("client_timeout", [0.1, 30.0])
+def test_wait_bounds_rpc_and_sleep(monkeypatch, method, client_timeout):
+    clock = Clock()
+    monkeypatch.setattr(sandbox_module, "time", clock)
+    timeouts = []
+
+    def get(request, timeout):
+        assert request.name == "job"
+        assert request.workspace_scope.workspace == "team"
+        timeouts.append(timeout)
+        clock.elapsed += 0.05
+        return openshell_pb2.SandboxResponse()
+
+    with SandboxClient("localhost:1", timeout=client_timeout) as client:
+        monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get))
+        with pytest.raises(SandboxError, match="within timeout"):
+            getattr(client, method)("job", workspace="team", timeout_seconds=1.25)
+
+    assert timeouts == pytest.approx(
+        [min(client_timeout, 1.25), min(client_timeout, 0.2)]
+    )
+    assert clock.elapsed == pytest.approx(1.25)
+
+
+@pytest.mark.parametrize(
+    ("method", "phase"),
+    [
+        ("wait_ready", openshell_pb2.SANDBOX_PHASE_READY),
+        ("wait_stopped", openshell_pb2.SANDBOX_PHASE_STOPPED),
+        ("wait_deleted", openshell_pb2.SANDBOX_PHASE_READY),
+    ],
+)
+def test_wait_ignores_wall_clock_changes(monkeypatch, method, phase):
+    clock = Clock()
+    monkeypatch.setattr(sandbox_module, "time", clock)
+    calls = 0
+
+    def get(_request, timeout):
+        nonlocal calls
+        assert timeout > 0
+        calls += 1
+        clock.wall_offset += 1000
+        response = openshell_pb2.SandboxResponse()
+        response.sandbox.metadata.id = "original" if calls == 1 else "replacement"
+        if calls == 2:
+            response.sandbox.status.phase = phase
+        return response
+
+    with SandboxClient("localhost:1") as client:
+        monkeypatch.setattr(client, "_stub", SimpleNamespace(GetSandbox=get))
+        options = (
+            {"expected_sandbox_id": "original"} if method == "wait_deleted" else {}
+        )
+        getattr(client, method)("job", workspace="team", timeout_seconds=2, **options)
+
+    assert calls == 2
+    assert clock.elapsed == 1
+
+
+@pytest.mark.parametrize("method", ["wait_ready", "wait_stopped", "wait_deleted"])
+def test_wait_cancels_stalled_lookup_at_deadline(method):
+    import threading
+    import time
+
+    def get(_request, context):
+        cancelled = threading.Event()
+        context.add_callback(cancelled.set)
+        cancelled.wait(5)
+        return openshell_pb2.SandboxResponse()
+
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
+    server.add_generic_rpc_handlers(
+        (
+            grpc.method_handlers_generic_handler(
+                "openshell.v1.OpenShell",
+                {
+                    "GetSandbox": grpc.unary_unary_rpc_method_handler(
+                        get,
+                        request_deserializer=openshell_pb2.GetSandboxRequest.FromString,
+                        response_serializer=openshell_pb2.SandboxResponse.SerializeToString,
+                    ),
+                },
+            ),
+        )
+    )
+    port = server.add_insecure_port("127.0.0.1:0")
+    server.start()
+    try:
+        with SandboxClient(f"127.0.0.1:{port}", timeout=2) as client:
+            started = time.monotonic()
+            with pytest.raises(SandboxError, match="within timeout"):
+                getattr(client, method)("job", workspace="team", timeout_seconds=0.1)
+            assert time.monotonic() - started < 1
+    finally:
+        server.stop(0).wait()