fix(python): add opt-in sandbox delete timeout

docs/sdk/python.mdx

@@ -148,6 +148,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.
 
+Delete calls use the client timeout, 30 seconds by default. Kubernetes gateways
+can hold a delete open until the pod terminates, so pass a larger
+`timeout_seconds` to `delete`, or `delete_timeout_seconds` (for example, 120) to
+`Sandbox`.
+
 ## Call Generated RPCs
 
 The generated gRPC stubs include concrete request and response types for synchronous `grpc.Channel` and asynchronous `grpc.aio.Channel` connections. Install the dependency type stubs in your development environment for editor signatures and type checking:

e2e/python/conftest.py

@@ -56,7 +56,10 @@ def cluster_name() -> str | None:
 
 @pytest.fixture(scope="session")
 def sandbox_client(cluster_name: str | None) -> Iterator[SandboxClient]:
-    with SandboxClient.from_active_cluster(cluster=cluster_name) as client:
+    # Direct deletes can wait out the Kubernetes pod grace period.
+    with SandboxClient.from_active_cluster(
+        cluster=cluster_name, timeout=120.0
+    ) as client:
         yield client
 
 
@@ -95,6 +98,9 @@ def _create(*, spec: object | None = None, delete_on_exit: bool = True) -> Sandb
             delete_on_exit=delete_on_exit,
             # Allow time to pull an explicitly supplied workload fixture.
             ready_timeout_seconds=300.0,
+            # Kubernetes holds DeleteSandbox open until the pod terminates
+            # (30s grace plus API headroom), which can exceed the 30s default.
+            delete_timeout_seconds=120.0,
         )
 
     return _create

python/openshell/sandbox.py

@@ -680,9 +680,16 @@ def exec_python(
             timeout_seconds=timeout_seconds,
         )
 
-    def delete(self, *, allow_missing: bool = False) -> DeletionResult:
+    def delete(
+        self, *, allow_missing: bool = False, timeout_seconds: float | None = None
+    ) -> DeletionResult:
+        # Forward `timeout_seconds` only on opt-in so older clients still work.
+        extra = {} if timeout_seconds is None else {"timeout_seconds": timeout_seconds}
         return self._client.delete(
-            self.sandbox.name, workspace=self._workspace, allow_missing=allow_missing
+            self.sandbox.name,
+            workspace=self._workspace,
+            allow_missing=allow_missing,
+            **extra,
         )
 
     def stop(self) -> SandboxRef:
@@ -1138,15 +1145,25 @@ def list_ids_for_all_workspaces(
         ]
 
     def delete(
-        self, name: str, *, workspace: str, allow_missing: bool = False
+        self,
+        name: str,
+        *,
+        workspace: str,
+        allow_missing: bool = False,
+        timeout_seconds: float | None = None,
     ) -> DeletionResult:
+        """Delete a sandbox.
+
+        `timeout_seconds` is the gRPC deadline for this call (default: the
+        client timeout).
+        """
         response = self._stub.DeleteSandbox(
             openshell_pb2.DeleteSandboxRequest(
                 workspace_scope=_workspace_scope(workspace),
                 name=name,
                 allow_missing=allow_missing,
             ),
-            timeout=self._timeout,
+            timeout=self._timeout if timeout_seconds is None else timeout_seconds,
         )
         return DeletionResult(
             DeletionOutcome(response.outcome), response.sandbox_id or None
@@ -1751,13 +1768,18 @@ def __init__(
         workload_template: str | None = None,
         timeout: float = 30.0,
         ready_timeout_seconds: float = 120.0,
+        delete_timeout_seconds: float | None = None,
         auto_refresh: bool = True,
         write_back: bool = True,
         insecure: bool = False,
         client_credentials: ClientCredentialsAuth | None = None,
     ) -> None:
         """Bind a Sandbox context to the active gateway.
 
+        `delete_timeout_seconds` bounds the delete RPC on exit and defaults to
+        `timeout`. Raise it for drivers that hold the delete open while the
+        workload terminates.
+
         OIDC kwargs (`auto_refresh`, `write_back`, `insecure`, and
         `client_credentials`) forward
         directly to `SandboxClient.from_active_cluster` and have the
@@ -1777,6 +1799,7 @@ def __init__(
         self._workload_template = workload_template
         self._timeout = timeout
         self._ready_timeout_seconds = ready_timeout_seconds
+        self._delete_timeout_seconds = delete_timeout_seconds
         self._auto_refresh = auto_refresh
         self._write_back = write_back
         self._insecure = insecure
@@ -1872,7 +1895,9 @@ def __exit__(self, *args: object) -> None:
                 and self._session is not None
                 and self._client is not None
             ):
-                result = self._session.delete(allow_missing=True)
+                result = self._session.delete(
+                    allow_missing=True, timeout_seconds=self._delete_timeout_seconds
+                )
                 if result.outcome == DeletionOutcome.ACCEPTED:
                     self._client.wait_deleted(
                         self._session.sandbox.name,

python/openshell/sandbox_cleanup_test.py

@@ -3,6 +3,7 @@
 
 """Cleanup status handling through the curated client's real gRPC interceptor."""
 
+import time
 from concurrent import futures
 from contextlib import nullcontext
 from types import SimpleNamespace
@@ -20,6 +21,7 @@ def cleanup_client():
     state = SimpleNamespace(
         exists=False,
         code=grpc.StatusCode.NOT_FOUND,
+        delete_delay=0.0,
         calls=[],
         closed=[],
     )
@@ -46,6 +48,7 @@ def delete(request, context):
         assert request.name == "cleanup-test"
         assert request.workspace_scope.workspace == "default"
         assert request.allow_missing
+        time.sleep(state.delete_delay)
         if state.code == grpc.StatusCode.NOT_FOUND:
             return openshell_pb2.DeleteSandboxResponse(
                 outcome=openshell_pb2.DELETION_OUTCOME_ALREADY_ABSENT
@@ -158,6 +161,44 @@ def test_context_cleanup_handles_absence_and_intercepted_errors(
     assert managed._session is None
 
 
+@pytest.mark.parametrize(
+    ("client_timeout", "delete_timeout_seconds", "expect_deadline"),
+    [(0.2, None, True), (0.2, 5, False), (5, 0.2, True)],
+)
+def test_context_cleanup_delete_timeout_is_opt_in(
+    cleanup_client,
+    monkeypatch,
+    client_timeout,
+    delete_timeout_seconds,
+    expect_deadline,
+):
+    client, state = cleanup_client
+    state.exists = True
+    state.delete_delay = 0.5
+    monkeypatch.setattr(
+        SandboxClient,
+        "from_active_cluster",
+        classmethod(lambda _cls, **_kwargs: client),
+    )
+
+    managed = Sandbox(
+        workspace="default",
+        sandbox="cleanup-test",
+        delete_timeout_seconds=delete_timeout_seconds,
+    )
+    expected = pytest.raises(GatewayError) if expect_deadline else nullcontext()
+    with expected as caught, managed:
+        # Only the exit delete runs under the short deadline.
+        client._timeout = client_timeout
+        state.exists = False
+    if caught is not None:
+        assert caught.value.code() == grpc.StatusCode.DEADLINE_EXCEEDED
+    assert state.calls == ["GetSandbox", "GetSandbox", "DeleteSandbox"]
+    assert state.closed == [True]
+    assert managed._client is None
+    assert managed._session is None
+
+
 class _EntryFailureClient:
     """Fails the named client call and records close()."""
 

python/openshell/sandbox_test.py

@@ -3276,3 +3276,25 @@ def test_sandbox_session_delete_passes_workspace() -> None:
 
     assert stub.delete_request is not None
     assert _request_workspace(stub.delete_request) == "staging"
+
+
+def test_sandbox_session_delete_omits_timeout_for_legacy_clients() -> None:
+    from openshell.sandbox import SandboxSession
+
+    calls: list[tuple[str, str, bool]] = []
+
+    class LegacyClient:
+        def delete(self, name, *, workspace, allow_missing=False):
+            calls.append((name, workspace, allow_missing))
+
+    ref = SandboxRef(
+        id="sandbox-1",
+        name="job-1",
+        workspace="staging",
+        status=SandboxStatusRef(phase=2, current_policy_version=0),
+    )
+    session = SandboxSession(cast("Any", LegacyClient()), ref)
+
+    session.delete(allow_missing=True)
+
+    assert calls == [("job-1", "staging", True)]