From 421610ff8245ca3de38473ce042dfdfba9c52398 Mon Sep 17 00:00:00 2001
From: Francesco Bonacci <195596869+f-trycua@users.noreply.github.com>
Date: Tue, 8 Sep 2026 00:30:29 -0500
Subject: [PATCH 1/3] fix(sandbox): bound typed Driver lifecycle and request
 deadlines

---
 .../cua_sandbox/interfaces/driver.py          |  98 ++++++++---
 .../cua_sandbox/transport/fleet.py            |  11 +-
 .../docs/typed-driver-development.md          |  16 ++
 .../cua-sandbox/tests/test_fleet_transport.py |  15 ++
 .../cua-sandbox/tests/test_typed_driver.py    | 157 +++++++++++++++++-
 .../tests/test_typed_driver_native.py         |   6 +-
 6 files changed, 275 insertions(+), 28 deletions(-)

diff --git a/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py b/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py
index c3cbd305ed..79d03ca97e 100644
--- a/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py
+++ b/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py
@@ -7,6 +7,7 @@
 import json
 import logging
 import re
+import time
 import uuid
 from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Any, AsyncIterator
@@ -23,15 +24,16 @@
 logger = logging.getLogger(__name__)
 
 
+def _consume_result(done: asyncio.Future) -> None:
+    if not done.cancelled():
+        done.exception()
+
+
 async def _bounded_cleanup(awaitable: Any) -> bool:
     """Bound best-effort cleanup, including callbacks that ignore cancellation."""
     task = asyncio.ensure_future(awaitable)
 
-    def consume_result(done: asyncio.Future) -> None:
-        if not done.cancelled():
-            done.exception()
-
-    task.add_done_callback(consume_result)
+    task.add_done_callback(_consume_result)
     done, _ = await asyncio.wait({task}, timeout=_CLEANUP_TIMEOUT)
     if task not in done:
         task.cancel()
@@ -75,6 +77,8 @@ def __init__(self, transport: Any):
         self._principal = uuid.uuid4().hex
         self._lock = asyncio.Lock()
         self._connections: dict[Any, Any] = {}
+        self._opening: dict[Any, asyncio.Task] = {}
+        self._closing: dict[Any, asyncio.Task] = {}
         self._closed = False
         self._loop: asyncio.AbstractEventLoop | None = None
 
@@ -114,14 +118,26 @@ async def connect(self, *, service: str = "driver") -> AsyncIterator[CuaDriver]:
                 )
             sdk = _sdk()
             channel = _channel(sdk, self._transport, service, self._principal)
-            try:
-                await channel.open()
-                driver = sdk.connect_remote_channel(channel)
-            except BaseException:
-                await channel.close()
-                raise
-            self._connections[channel] = driver
+
+            async def open_channel():
+                try:
+                    await channel.open()
+                finally:
+                    # A cancelled carrier may still return an allocated session.
+                    if channel.closed:
+                        await channel.close()
+
+            opening = asyncio.create_task(open_channel())
+            opening.add_done_callback(_consume_result)
+            self._opening[channel] = opening
+            self._connections[channel] = None
         try:
+            await asyncio.shield(opening)
+            async with self._lock:
+                if self._closed or channel.closed:
+                    raise DriverConnectionError("Sandbox Driver accessor is disconnected")
+                driver = sdk.connect_remote_channel(channel)
+                self._connections[channel] = driver
             yield driver
         finally:
             await self._close_connection(channel)
@@ -129,11 +145,25 @@ async def connect(self, *, service: str = "driver") -> AsyncIterator[CuaDriver]:
     async def _close_connection(self, channel: Any) -> None:
         self._check_loop()
         async with self._lock:
-            driver = self._connections.pop(channel, None)
-            if driver is not None:
+            task = self._closing.get(channel)
+            if task is None and channel in self._connections:
+                driver = self._connections.pop(channel)
+                opening = self._opening.pop(channel)
                 channel.closed = True
-                await _bounded_cleanup(channel.close())
-                await _bounded_cleanup(driver.shutdown())
+
+                async def cleanup():
+                    if not opening.done():
+                        await _bounded_cleanup(opening)
+                    await channel.close()
+                    if driver is not None:
+                        await _bounded_cleanup(driver.shutdown())
+
+                task = asyncio.create_task(cleanup())
+                self._closing[channel] = task
+                task.add_done_callback(_consume_result)
+                task.add_done_callback(lambda done: self._closing.pop(channel, None))
+        if task is not None:
+            await asyncio.shield(task)
 
     async def close(self) -> None:
         """Invalidate all connections before the owning transport is closed."""
@@ -142,12 +172,17 @@ async def close(self) -> None:
             self._closed = True
             for channel in self._connections:
                 channel.closed = True
-        for channel in list(self._connections):
+
+        async def close_channel(channel):
             try:
                 await self._close_connection(channel)
             except Exception:
                 logger.warning("Driver cleanup failed; remote session cleanup is unconfirmed")
 
+        await asyncio.gather(
+            *(close_channel(channel) for channel in set(self._connections) | set(self._closing))
+        )
+
 
 def _channel(sdk: Any, transport: FleetTransport, service: str, principal: str) -> Any:
     class Channel(sdk.ForeignDriverEnvelopeChannel):
@@ -164,7 +199,7 @@ def __init__(self):
         def fail(self, reason: str):
             return sdk.ForeignDriverChannelError.Failed(reason)
 
-        async def request(self, method: str, suffix: str = "", body: Any = None):
+        async def request(self, method: str, suffix: str = "", body: Any = None, *, timeout=None):
             path = "/v1/connections"
             headers = None
             if self.connection_id is not None:
@@ -172,7 +207,12 @@ async def request(self, method: str, suffix: str = "", body: Any = None):
                 headers = {"X-Cua-Driver-Generation": self.generation}
             try:
                 response = await transport.request_service(
-                    service, method=method, path=path, json_body=body, headers=headers
+                    service,
+                    method=method,
+                    path=path,
+                    json_body=body,
+                    headers=headers,
+                    **({"timeout": timeout} if timeout is not None else {}),
                 )
             except Exception as error:
                 if getattr(error, "status", None) in (401, 403):
@@ -250,6 +290,9 @@ async def exchange(self, request):
             if self.closed or request.request_id in self.cancelled:
                 raise self.fail("Driver connection is closed or request was cancelled")
             try:
+                deadline = request.deadline_unix_ms
+                if type(deadline) is not int or deadline < 0:
+                    raise ValueError
                 body = {
                     "envelope_version": request.envelope_version,
                     "request_id": request.request_id,
@@ -266,11 +309,19 @@ async def exchange(self, request):
                     raise ValueError
             except (ValueError, TypeError):
                 raise self.fail("Driver request is malformed or exceeds the size limit") from None
+            timeout = min(deadline - int(time.time() * 1000), 120000) / 1000
+            if timeout <= 0:
+                raise self.fail("Driver request deadline has expired")
+            exchange = asyncio.create_task(self.request("POST", "/exchange", body, timeout=timeout))
+            exchange.add_done_callback(_consume_result)
             try:
-                response = await self.request("POST", "/exchange", body)
-                return await self.decode_response(request, response)
+                done, _ = await asyncio.wait({exchange}, timeout=timeout)
+                if exchange not in done:
+                    raise TimeoutError("Driver request deadline has expired; completion is unknown")
+                return await self.decode_response(request, exchange.result())
             except BaseException:
-                self.closed = True
+                exchange.cancel()
+                await self.cancel(request.request_id)
                 await self.close()
                 raise
 
@@ -329,6 +380,9 @@ async def cancel(self, request_id):
 
         async def close(self):
             self.closed = True
+            # Do not memoize a no-op before an in-flight open reveals its ID.
+            if self.connection_id is None:
+                return
             if self._close_task is None:
 
                 async def cleanup():
diff --git a/libs/python/cua-sandbox/cua_sandbox/transport/fleet.py b/libs/python/cua-sandbox/cua_sandbox/transport/fleet.py
index a7004d895a..873c090f77 100644
--- a/libs/python/cua-sandbox/cua_sandbox/transport/fleet.py
+++ b/libs/python/cua-sandbox/cua_sandbox/transport/fleet.py
@@ -91,11 +91,17 @@ async def request_service(
         path: str,
         json_body: Any = None,
         headers: dict[str, str] | None = None,
+        timeout: float | None = None,
     ) -> httpx.Response:
         if name not in self._bound.services:
             raise ValueError(f"Fleet sandbox does not expose service {name!r}")
         return await self._request(
-            method, path, json_body=json_body, service_name=name, extra_headers=headers
+            method,
+            path,
+            json_body=json_body,
+            service_name=name,
+            extra_headers=headers,
+            timeout=timeout,
         )
 
     async def create_signed_service_url(
@@ -129,6 +135,7 @@ async def _request(
         json_body: Any = None,
         service_name: str | None = None,
         extra_headers: dict[str, str] | None = None,
+        timeout: float | None = None,
     ) -> httpx.Response:
         assert self._connected, "Transport not connected"
         body = None if json_body is None else json.dumps(json_body).encode()
@@ -146,7 +153,7 @@ async def _request(
                 url=f"https://service.invalid{path}",
                 headers=headers,
                 body=body,
-                timeout_secs=_whole_seconds(self._timeout),
+                timeout_secs=_whole_seconds(self._timeout if timeout is None else timeout),
             ),
         )
         request = httpx.Request(method, f"https://service.invalid{path}")
diff --git a/libs/python/cua-sandbox/docs/typed-driver-development.md b/libs/python/cua-sandbox/docs/typed-driver-development.md
index aa36fb432b..4b8625a474 100644
--- a/libs/python/cua-sandbox/docs/typed-driver-development.md
+++ b/libs/python/cua-sandbox/docs/typed-driver-development.md
@@ -62,12 +62,28 @@ waits; an unconfirmed cleanup logs a warning and does not block claim release.
 The carrier's independent session expiry remains the fallback for unreachable
 cleanup. A successful local close is not proof that a remote guest was deleted.
 
+Opening handshakes do not hold the accessor's lifecycle lock. Closing the
+accessor invalidates pending opens and bounds the wait for them, so a stalled
+handshake cannot indefinitely delay claim release. If a cancelled open later
+returns a connection ID, the accessor attempts to delete it without yielding a
+Driver. Deletion can remain unconfirmed if the owning transport has already
+disconnected or the handshake never returns its connection ID.
+
+Each Driver exchange uses the envelope's remaining deadline, capped at 120
+seconds, for its per-request Fleet timeout and local wait. This does not change
+the transport's default timeout for computer-server or other callers. Expired
+requests fail before dispatch. A locally timed-out or cancelled exchange
+invalidates the connection and attempts bounded cancellation and deletion; it
+does not replay the desktop action.
+
 ## Verification and remaining proof
 
 The focused tests cover named-service routing, canonical method dispatch,
 response validation, cancellation, and lifecycle ordering. CI requires the
 native bridge integration test through `CUA_SANDBOX_REQUIRE_NATIVE_DRIVER=1`;
 it fails instead of skipping when the matching native package is absent.
+Regression tests also cover pending-open cleanup, late-open invalidation,
+per-request deadlines, and preservation of ordinary transport timeouts.
 
 These synthetic tests do not prove a real guest desktop effect. Before a
 released tutorial or supported-image claim, qualify the exact candidate on a
diff --git a/libs/python/cua-sandbox/tests/test_fleet_transport.py b/libs/python/cua-sandbox/tests/test_fleet_transport.py
index f011123e2b..d5b1689b35 100644
--- a/libs/python/cua-sandbox/tests/test_fleet_transport.py
+++ b/libs/python/cua-sandbox/tests/test_fleet_transport.py
@@ -91,3 +91,18 @@ async def test_requests_are_bounded_by_the_transport_timeout():
 
     assert sdk.calls[0][3].timeout_secs == 30
     assert sdk.calls[1][3].timeout_secs == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("default_timeout", [30, 90])
+async def test_service_timeout_override_does_not_change_existing_callers(default_timeout):
+    sdk = FakeSDK([response(), response(), response(body=b'data: {"success":true}\n\n')])
+    transport = FleetTransport(sdk=sdk, bound=sandbox(), timeout=default_timeout)
+    await transport.connect()
+
+    await transport.request_service("api", method="POST", path="/exchange", timeout=119.25)
+    await transport.request_service("api", method="GET", path="/status")
+    await transport.send("shell.run", timeout=15)
+
+    assert [call[3].timeout_secs for call in sdk.calls] == [120, default_timeout, default_timeout]
+    assert transport._timeout == default_timeout
diff --git a/libs/python/cua-sandbox/tests/test_typed_driver.py b/libs/python/cua-sandbox/tests/test_typed_driver.py
index a357ffd0a0..7496ec0e0a 100644
--- a/libs/python/cua-sandbox/tests/test_typed_driver.py
+++ b/libs/python/cua-sandbox/tests/test_typed_driver.py
@@ -87,7 +87,9 @@ def __init__(self):
         )
         self.status = 200
 
-    async def request_service(self, name, *, method, path, json_body=None, headers=None):
+    async def request_service(
+        self, name, *, method, path, json_body=None, headers=None, timeout=None
+    ):
         assert self._connected
         self.events.append((name, method, path, json_body, headers))
         data = self.open_data if path == "/v1/connections" else self.response_data
@@ -235,7 +237,7 @@ async def test_exchange_errors_are_sanitized_and_never_replayed(sandbox, status,
         with pytest.raises(ChannelError, match=reason) as error:
             await driver.channel.exchange(envelope())
         assert "secret" not in str(error.value)
-        assert len(sandbox._transport.events) == 3
+        assert len(sandbox._transport.events) == (3 if status in (404, 409) else 4)
         assert sandbox._transport.events[-1][1] == "DELETE"
         assert driver.channel.closed
         before = len(sandbox._transport.events)
@@ -481,3 +483,154 @@ async def hanging_exchange(*args, **kwargs):
         assert driver.channel.closed
         assert driver.channel.cleanup_confirmed
         assert sandbox._transport.events[-1][1] == "DELETE"
+
+
+@pytest.mark.parametrize("close_claim", [False, True])
+async def test_pending_open_does_not_block_disconnect(sandbox, monkeypatch, close_claim):
+    monkeypatch.setattr("cua_sandbox.interfaces.driver._CLEANUP_TIMEOUT", 0.01)
+    started = asyncio.Event()
+    original = sandbox._transport.request_service
+
+    async def pending(*args, **kwargs):
+        if kwargs["path"] == "/v1/connections":
+            started.set()
+            await asyncio.Event().wait()
+        return await original(*args, **kwargs)
+
+    monkeypatch.setattr(sandbox._transport, "request_service", pending)
+    if close_claim:
+
+        async def release():
+            sandbox._transport.events.append("release")
+
+        sandbox._claim_handle = SimpleNamespace(release=release, name=None)
+    opening = asyncio.create_task(sandbox.driver.connect().__aenter__())
+    await started.wait()
+    await asyncio.wait_for(sandbox.close() if close_claim else sandbox.disconnect(), 0.5)
+    with pytest.raises(asyncio.CancelledError):
+        await opening
+    assert not sandbox.driver._connections
+    assert sandbox._transport.events[-1] == "disconnect"
+    if close_claim:
+        assert sandbox._transport.events[-2] == "release"
+
+
+@pytest.mark.parametrize("cancel_context", [False, True])
+async def test_late_open_is_cleaned_without_resurrecting(sandbox, monkeypatch, cancel_context):
+    monkeypatch.setattr("cua_sandbox.interfaces.driver._CLEANUP_TIMEOUT", 0.01)
+    started, cancelled, finish, deleted = (asyncio.Event() for _ in range(4))
+    original = sandbox._transport.request_service
+
+    async def resistant(*args, **kwargs):
+        if kwargs["path"] == "/v1/connections":
+            started.set()
+            try:
+                await finish.wait()
+            except asyncio.CancelledError:
+                cancelled.set()
+                await finish.wait()
+        response = await original(*args, **kwargs)
+        if kwargs["method"] == "DELETE":
+            deleted.set()
+        return response
+
+    monkeypatch.setattr(sandbox._transport, "request_service", resistant)
+    opening = asyncio.create_task(sandbox.driver.connect().__aenter__())
+    await started.wait()
+    if cancel_context:
+        opening.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(opening, 0.5)
+    else:
+        await asyncio.wait_for(sandbox.driver.close(), 0.5)
+    await cancelled.wait()
+    assert not sandbox.driver._connections
+    finish.set()
+    await asyncio.wait_for(deleted.wait(), 0.5)
+    if not cancel_context:
+        with pytest.raises(DriverConnectionError, match="disconnected"):
+            await opening
+    assert [e[1] for e in sandbox._transport.events] == ["POST", "DELETE"]
+
+
+@pytest.mark.parametrize("deadline", [1120000, 10**1000])
+async def test_driver_deadline_overrides_short_transport_timeout(sandbox, monkeypatch, deadline):
+    monkeypatch.setattr("cua_sandbox.interfaces.driver.time.time", lambda: 1000)
+    original = sandbox._transport.request_service
+    timeouts = []
+
+    async def record(*args, **kwargs):
+        timeouts.append(kwargs.get("timeout"))
+        return await original(*args, **kwargs)
+
+    monkeypatch.setattr(sandbox._transport, "request_service", record)
+    async with sandbox.driver.connect() as driver:
+        await driver.channel.exchange(envelope(deadline_unix_ms=deadline))
+    assert timeouts == [None, 120, None]
+    assert sandbox._transport._timeout == 30
+
+
+@pytest.mark.parametrize("deadline", [0, 999999, 1000000, -1])
+async def test_expired_or_invalid_deadline_never_dispatches(sandbox, monkeypatch, deadline):
+    monkeypatch.setattr("cua_sandbox.interfaces.driver.time.time", lambda: 1000)
+    async with sandbox.driver.connect() as driver:
+        with pytest.raises(ChannelError, match="expired|malformed"):
+            await driver.channel.exchange(envelope(deadline_unix_ms=deadline))
+        assert len(sandbox._transport.events) == 1
+
+
+async def test_local_deadline_cancels_then_deletes_without_replay(sandbox, monkeypatch):
+    monkeypatch.setattr("cua_sandbox.interfaces.driver.time.time", lambda: 1000)
+    original = sandbox._transport.request_service
+    dispatched = []
+
+    async def pending(*args, **kwargs):
+        dispatched.append((kwargs["method"], kwargs["path"]))
+        if kwargs["path"].endswith("/exchange"):
+            await asyncio.Event().wait()
+        return await original(*args, **kwargs)
+
+    monkeypatch.setattr(sandbox._transport, "request_service", pending)
+    async with sandbox.driver.connect() as driver:
+        with pytest.raises(TimeoutError):
+            await driver.channel.exchange(envelope(deadline_unix_ms=1000010))
+        assert driver.channel.closed
+        assert driver.channel.cleanup_confirmed
+    assert dispatched == [
+        ("POST", "/v1/connections"),
+        ("POST", "/v1/connections/connection-1/exchange"),
+        ("POST", "/v1/connections/connection-1/cancel"),
+        ("DELETE", "/v1/connections/connection-1"),
+    ]
+    assert not sandbox.driver._closing
+
+
+async def test_deadline_bounds_cancellation_resistant_exchange(sandbox, monkeypatch):
+    monkeypatch.setattr("cua_sandbox.interfaces.driver.time.time", lambda: 1000)
+    cancelled, finish, returned = (asyncio.Event() for _ in range(3))
+    original = sandbox._transport.request_service
+
+    async def resistant(*args, **kwargs):
+        if kwargs["path"].endswith("/exchange"):
+            try:
+                await finish.wait()
+            except asyncio.CancelledError:
+                cancelled.set()
+                await finish.wait()
+            result = await original(*args, **kwargs)
+            returned.set()
+            return result
+        return await original(*args, **kwargs)
+
+    monkeypatch.setattr(sandbox._transport, "request_service", resistant)
+    async with sandbox.driver.connect() as driver:
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(driver.channel.exchange(envelope(deadline_unix_ms=1000010)), 0.5)
+        await cancelled.wait()
+        assert driver.channel.closed
+        assert driver.channel.cleanup_confirmed
+        finish.set()
+        await asyncio.wait_for(returned.wait(), 0.5)
+        with pytest.raises(ChannelError, match="closed"):
+            await driver.channel.exchange(envelope(request_id="next"))
+    assert [event[1] for event in sandbox._transport.events] == ["POST", "POST", "DELETE", "POST"]
diff --git a/libs/python/cua-sandbox/tests/test_typed_driver_native.py b/libs/python/cua-sandbox/tests/test_typed_driver_native.py
index ac6e42cc80..129591ff5c 100644
--- a/libs/python/cua-sandbox/tests/test_typed_driver_native.py
+++ b/libs/python/cua-sandbox/tests/test_typed_driver_native.py
@@ -41,9 +41,11 @@ def __init__(self, result=None):
             }
         )
 
-    async def request_service(self, name, *, method, path, json_body=None, headers=None):
+    async def request_service(
+        self, name, *, method, path, json_body=None, headers=None, timeout=None
+    ):
         response = await super().request_service(
-            name, method=method, path=path, json_body=json_body, headers=headers
+            name, method=method, path=path, json_body=json_body, headers=headers, timeout=timeout
         )
         if path.endswith("/exchange"):
             return httpx.Response(

From 89ea6e5ed237c95e847d648af8f0a54e51e1e521 Mon Sep 17 00:00:00 2001
From: Francesco Bonacci <f@trycua.com>
Date: Tue, 8 Sep 2026 16:34:09 -0500
Subject: [PATCH 2/3] fix(cua-driver): allow explicit unrestricted envelope
 sessions

---
 libs/cua-driver/docs/private-envelope-http.md |  27 ++-
 .../cua-driver/src/driver_service_http.rs     | 166 +++++++++++++++++-
 .../rust/crates/cua-driver/src/sdk_adapter.rs |   7 +-
 3 files changed, 188 insertions(+), 12 deletions(-)

diff --git a/libs/cua-driver/docs/private-envelope-http.md b/libs/cua-driver/docs/private-envelope-http.md
index fb08059427..34d7ff1f44 100644
--- a/libs/cua-driver/docs/private-envelope-http.md
+++ b/libs/cua-driver/docs/private-envelope-http.md
@@ -30,12 +30,27 @@ value returned at creation. Missing generations fail with HTTP 400, absent
 connections with 404, and mismatches with 409. A client must not reopen a
 connection automatically after those failures.
 
-Creation binds a Standard session with a one-hour maximum lifetime and a
-five-minute idle lifetime. The immutable runtime ceiling still applies:
-incompatible runtimes refuse creation. This slice does not inherit unrestricted
-mode or accept permission modes, manifests, or arbitrary session options from
-the wire. Ordinary typed calls use `session=None`; operations requiring a
-session label use the returned `public_session`.
+Creation binds a Standard session by default, with a one-hour maximum lifetime
+and a five-minute idle lifetime. The immutable runtime ceiling still applies:
+incompatible runtimes refuse creation. This slice does not implicitly inherit
+unrestricted mode or accept permission modes, manifests, or arbitrary session
+options from the wire. Ordinary typed calls use `session=None`; operations
+requiring a session label use the returned `public_session`.
+
+For an explicitly authorized disposable or trusted environment, the launcher
+can set `CUA_DRIVER_ENVELOPE_PERMISSION_MODE=unrestricted` and launch the daemon
+with `--permission-mode unrestricted --dangerously-bypass-approvals`. Both the
+carrier opt-in and the existing runtime risk acknowledgement are required.
+The carrier reads the setting once at startup; clients cannot change it.
+The default remains `standard`, even on an unrestricted daemon. Other values,
+including `bounded`, fail startup. A carrier with a host capability manifest
+also fails startup: this first slice does not support manifest configuration.
+The SDK treats compatibility-call and trusted-session manifests separately;
+this carrier neither inherits the former nor exposes the latter. This is a
+carrier limitation, not a change to the SDK's per-session manifest contract.
+Managed and user policies remain binding.
+This option does not authorize public exposure, alter Fleet authorization,
+or change existing computer-server sessions.
 
 `capabilities` contains `minimum_envelope_version`, `maximum_envelope_version`,
 and `supports_cancellation`. This carrier supports envelope version 1 and
diff --git a/libs/cua-driver/rust/crates/cua-driver/src/driver_service_http.rs b/libs/cua-driver/rust/crates/cua-driver/src/driver_service_http.rs
index 886a18dc02..9b57a5b84c 100644
--- a/libs/cua-driver/rust/crates/cua-driver/src/driver_service_http.rs
+++ b/libs/cua-driver/rust/crates/cua-driver/src/driver_service_http.rs
@@ -6,8 +6,8 @@
 //! Connection IDs and generations are routing/lifecycle markers, not credentials.
 //! Each connection already has a host-bound root session; independent bound
 //! sessions are unsupported. No request supplies permission options or paths.
-//! This first slice requests Standard sessions only; incompatible runtime
-//! ceilings refuse creation. It does not inherit or widen the runtime mode.
+//! Standard is the default. Unrestricted sessions require a separate trusted
+//! launcher opt-in and an already acknowledged unrestricted runtime.
 
 use cua_driver_sdk::remote::DriverRequestEnvelope;
 use cua_driver_sdk::remote_receiver::DriverEnvelopeReceiver;
@@ -348,11 +348,53 @@ impl Drop for Server {
     }
 }
 
+fn select_session_mode(
+    requested: Option<&str>,
+    host: cua_driver_core::authorization::PermissionMode,
+    has_manifest: bool,
+) -> anyhow::Result<cua_driver_sdk::SessionPermissionMode> {
+    use cua_driver_core::authorization::PermissionMode;
+    use cua_driver_sdk::SessionPermissionMode;
+    // This carrier cannot propagate a host manifest into its bound session.
+    anyhow::ensure!(
+        !has_manifest,
+        "envelope sessions with a host capability manifest are unsupported"
+    );
+    match requested {
+        None | Some("standard") => Ok(SessionPermissionMode::Standard),
+        Some("unrestricted") => {
+            anyhow::ensure!(
+                host == PermissionMode::Unrestricted,
+                "unrestricted envelope sessions require an acknowledged unrestricted host"
+            );
+            Ok(SessionPermissionMode::Unrestricted)
+        }
+        Some(_) => {
+            anyhow::bail!("CUA_DRIVER_ENVELOPE_PERMISSION_MODE must be standard or unrestricted")
+        }
+    }
+}
+
+fn configured_session_mode() -> anyhow::Result<cua_driver_sdk::SessionPermissionMode> {
+    let requested = match std::env::var("CUA_DRIVER_ENVELOPE_PERMISSION_MODE") {
+        Ok(value) => Some(value),
+        Err(std::env::VarError::NotPresent) => None,
+        Err(error) => return Err(error.into()),
+    };
+    let host =
+        cua_driver_core::authorization::configured_permission_mode().map_err(anyhow::Error::msg)?;
+    let has_manifest = cua_driver_core::session_manifest::configured_capability_manifest()
+        .map_err(anyhow::Error::msg)?
+        .is_some();
+    select_session_mode(requested.as_deref(), host, has_manifest)
+}
+
 pub async fn start(sdk: Arc<crate::sdk_adapter::SdkAdapter>, port: u16) -> anyhow::Result<Server> {
+    let mode = configured_session_mode()?;
     let listener = TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, port)).await?;
     let service = Arc::new(Service {
         entries: Mutex::new(HashMap::new()),
-        factory: Arc::new(move || sdk.create_envelope_receiver()),
+        factory: Arc::new(move || sdk.create_envelope_receiver(mode)),
         exchanges: tokio::sync::Semaphore::new(MAX_EXCHANGES),
     });
     let task = tokio::spawn(async move {
@@ -393,6 +435,124 @@ mod tests {
     use cua_driver_sdk::{remote_receiver::DriverEnvelopeExecutor, DriverError};
     use std::sync::atomic::{AtomicUsize, Ordering};
 
+    #[test]
+    fn envelope_mode_never_implicitly_inherits_unrestricted() {
+        use cua_driver_core::authorization::PermissionMode as Host;
+        for host in [Host::Standard, Host::Bounded, Host::Unrestricted] {
+            for requested in [None, Some("standard")] {
+                assert_eq!(
+                    select_session_mode(requested, host, false).unwrap(),
+                    cua_driver_sdk::SessionPermissionMode::Standard
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn envelope_unrestricted_requires_matching_host_and_no_manifest() {
+        use cua_driver_core::authorization::PermissionMode as Host;
+        assert_eq!(
+            select_session_mode(Some("unrestricted"), Host::Unrestricted, false).unwrap(),
+            cua_driver_sdk::SessionPermissionMode::Unrestricted
+        );
+        for host in [Host::Standard, Host::Bounded] {
+            assert!(select_session_mode(Some("unrestricted"), host, false).is_err());
+        }
+        assert!(select_session_mode(Some("unrestricted"), Host::Unrestricted, true).is_err());
+        assert!(select_session_mode(None, Host::Standard, true).is_err());
+        assert!(select_session_mode(Some("standard"), Host::Standard, true).is_err());
+    }
+
+    #[test]
+    fn envelope_mode_rejects_unknown_and_bounded_values() {
+        use cua_driver_core::authorization::PermissionMode;
+        for value in ["", "bounded", "UNRESTRICTED", " unrestricted", "inherit"] {
+            assert!(select_session_mode(Some(value), PermissionMode::Unrestricted, false).is_err());
+        }
+    }
+
+    #[test]
+    fn envelope_startup_mode_requires_explicit_acknowledgement() {
+        const CHILD: &str = "CUA_TEST_ENVELOPE_MODE_CHILD";
+        if let Ok(expected) = std::env::var(CHILD) {
+            let selected = configured_session_mode();
+            match expected.as_str() {
+                "standard" => assert_eq!(
+                    selected.unwrap(),
+                    cua_driver_sdk::SessionPermissionMode::Standard
+                ),
+                "unrestricted" => assert_eq!(
+                    selected.unwrap(),
+                    cua_driver_sdk::SessionPermissionMode::Unrestricted
+                ),
+                "error" => assert!(selected.is_err()),
+                "manifest_error" => {
+                    assert!(
+                        cua_driver_core::session_manifest::configured_capability_manifest()
+                            .unwrap()
+                            .is_some()
+                    );
+                    assert!(selected.is_err());
+                }
+                _ => panic!("unknown synthetic test expectation"),
+            }
+            return;
+        }
+        // Startup mode is process-cached; each case must get a fresh process.
+        for (requested, host, acknowledged, manifest, expected) in [
+            (None, "standard", false, false, "standard"),
+            (None, "unrestricted", true, false, "standard"),
+            (Some("unrestricted"), "standard", false, false, "error"),
+            (Some("unrestricted"), "unrestricted", false, false, "error"),
+            (
+                Some("unrestricted"),
+                "unrestricted",
+                true,
+                false,
+                "unrestricted",
+            ),
+            (Some("bounded"), "unrestricted", true, false, "error"),
+            (None, "standard", false, true, "manifest_error"),
+            (
+                Some("unrestricted"),
+                "unrestricted",
+                true,
+                true,
+                "manifest_error",
+            ),
+        ] {
+            use std::io::Write;
+            let mut file = tempfile::NamedTempFile::new().unwrap();
+            file.write_all(b"version: 3\nallow:\n  tools: [get_config]\n")
+                .unwrap();
+            let mut command = std::process::Command::new(std::env::current_exe().unwrap());
+            command
+                .args([
+                    "--exact",
+                    "driver_service_http::tests::envelope_startup_mode_requires_explicit_acknowledgement",
+                    "--nocapture",
+                ])
+                .env(CHILD, expected)
+                .env("CUA_DRIVER_PERMISSION_MODE", host)
+                .env("CUA_DRIVER_DANGEROUSLY_BYPASS_APPROVALS", if acknowledged { "1" } else { "0" })
+                .env_remove("CUA_DRIVER_ENVELOPE_PERMISSION_MODE")
+                .env_remove("CUA_DRIVER_CAPABILITY_MANIFEST_FILE")
+                .env_remove("CUA_DRIVER_SESSION_POLICY_FILE");
+            if manifest {
+                command.env("CUA_DRIVER_CAPABILITY_MANIFEST_FILE", file.path());
+            }
+            if let Some(requested) = requested {
+                command.env("CUA_DRIVER_ENVELOPE_PERMISSION_MODE", requested);
+            }
+            let output = command.output().unwrap();
+            assert!(
+                output.status.success(),
+                "startup mode case failed: {}",
+                String::from_utf8_lossy(&output.stderr)
+            );
+        }
+    }
+
     struct Fake(Arc<AtomicUsize>);
     #[async_trait::async_trait]
     impl DriverEnvelopeExecutor for Fake {
diff --git a/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs b/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs
index 9d12833315..a26a6bfe95 100644
--- a/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs
+++ b/libs/cua-driver/rust/crates/cua-driver/src/sdk_adapter.rs
@@ -59,6 +59,7 @@ pub struct SdkAdapter {
 impl SdkAdapter {
     pub fn create_envelope_receiver(
         &self,
+        mode: cua_driver_sdk::SessionPermissionMode,
     ) -> Result<
         (
             Arc<cua_driver_sdk::remote_receiver::DriverEnvelopeReceiver>,
@@ -66,12 +67,12 @@ impl SdkAdapter {
         ),
         String,
     > {
-        // The private HTTP slice requests only Standard; the runtime's immutable
-        // ceiling rejects incompatible hosts rather than widening their policy.
+        // Only the trusted launcher selects this mode. The runtime's immutable
+        // ceiling still rejects incompatible sessions.
         let public_session = format!("http-{}", uuid::Uuid::new_v4());
         let options = TrustedSessionOptions {
             public_session: public_session.clone(),
-            mode: cua_driver_sdk::SessionPermissionMode::Standard,
+            mode,
             ttl_seconds: 3600,
             idle_ttl_seconds: 300,
             capability_manifest_path: None,

From e4d58d39e53b802251733187aa26efaa59863158 Mon Sep 17 00:00:00 2001
From: Francesco Bonacci <f@trycua.com>
Date: Tue, 8 Sep 2026 18:22:48 -0500
Subject: [PATCH 3/3] docs(sandbox): clarify launcher-selected Driver
 permissions

---
 libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py | 9 +++++----
 libs/python/cua-sandbox/docs/typed-driver-development.md | 7 +++++--
 libs/python/cua-sandbox/tests/test_typed_driver.py       | 2 +-
 3 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py b/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py
index 79d03ca97e..fd723fece5 100644
--- a/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py
+++ b/libs/python/cua-sandbox/cua_sandbox/interfaces/driver.py
@@ -64,9 +64,10 @@ def _sdk() -> Any:
 class Driver:
     """Accessor for optional typed Driver sessions; does not replace Sandbox transport.
 
-    Each connection is a host-bound Standard session. Use typed inputs with
-    ``session=None`` where optional. For required session fields, obtain the
-    bound name with ``sandbox.driver.session_name(driver)``. Creating or
+    Each connection is a host-bound session with a launcher-selected permission
+    mode (Standard by default). Remote clients cannot select its authority.
+    Use typed inputs with ``session=None`` where optional. For required session
+    fields, obtain the bound name with ``sandbox.driver.session_name(driver)``. Creating or
     rebinding trusted sessions is unsupported. Remote cleanup is best effort
     with a bounded timeout; failures warn and do not block claim release.
     """
@@ -283,7 +284,7 @@ async def negotiate(self):
 
         async def bind_session(self, options):
             raise self.fail(
-                "Fleet Driver connections use a host-bound Standard session; rebinding is unsupported"
+                "Fleet Driver connections use a host-bound session; rebinding is unsupported"
             )
 
         async def exchange(self, request):
diff --git a/libs/python/cua-sandbox/docs/typed-driver-development.md b/libs/python/cua-sandbox/docs/typed-driver-development.md
index 4b8625a474..48658b2db6 100644
--- a/libs/python/cua-sandbox/docs/typed-driver-development.md
+++ b/libs/python/cua-sandbox/docs/typed-driver-development.md
@@ -44,8 +44,11 @@ async def inspect_guest(pool):
 `driver` is the generated `cua_driver.CuaDriver`, not a parallel desktop API.
 Use `session=None` for optional session fields. For required session fields,
 `session_name(driver)` returns the active connection's host-bound label. The
-carrier creates a Standard session; remote permission grants, new trusted
-sessions, and session rebinding are not supported.
+carrier creates a Standard session by default. Only the trusted launcher can
+opt into Unrestricted mode, with an explicitly acknowledged Unrestricted daemon;
+remote clients cannot select session authority. Bounded mode and carrier manifest
+configuration are unsupported, as are remote permission grants, new trusted
+sessions, and session rebinding.
 
 ## Failure and cleanup behavior
 
diff --git a/libs/python/cua-sandbox/tests/test_typed_driver.py b/libs/python/cua-sandbox/tests/test_typed_driver.py
index 7496ec0e0a..4e5e750dcf 100644
--- a/libs/python/cua-sandbox/tests/test_typed_driver.py
+++ b/libs/python/cua-sandbox/tests/test_typed_driver.py
@@ -301,7 +301,7 @@ async def test_cancelled_request_never_dispatches_and_cancel_after_close_is_loca
 
 async def test_bind_session_is_unsupported_without_dispatch(sandbox):
     async with sandbox.driver.connect() as driver:
-        with pytest.raises(ChannelError, match="Standard.*unsupported"):
+        with pytest.raises(ChannelError, match="rebinding is unsupported"):
             await driver.channel.bind_session(object())
         assert len(sandbox._transport.events) == 1