diff --git a/docs/sdk-python.md b/docs/sdk-python.md
index 7ab7dae9..3b8f1f5f 100644
--- a/docs/sdk-python.md
+++ b/docs/sdk-python.md
@@ -170,6 +170,16 @@ automatically with the same transparent wake. A claim with a connection live
 when the sweep checks it (a relay stream, a buffered exec, a preview dial, an
 egress request) is not swept; the idle clock restarts when that connection ends.
 
+Data-plane calls share a handle's relay connection: after a call the SDK
+keeps the connection for 30 seconds (`Client(..., keep_alive=...)` tunes the
+window; 0 dials per call) and the next call on that handle sends its request
+on it, so a busy handle pays the dial, upgrade and TLS handshake once. A kept
+connection counts as live for `idle_hibernate_seconds` until it closes, so
+keep the window below that setting; `close` and `hibernate` drop it at once.
+Streams (`watch`, `open_pty`, `dial_port`, an LSP session) take a connection
+of their own, and a guest whose silkd predates the back-to-back protocol
+gets one connection per call as before.
+
 If that deployment also enables `archive_after_seconds`, archiving replaces
 the original claim deadline with the archive-retention deadline (or no
 deadline when archives are kept forever). Waking an archive starts a fresh
diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py
index 0febb4d1..85ad3be7 100644
--- a/sdk/python/cocoonsandbox/client.py
+++ b/sdk/python/cocoonsandbox/client.py
@@ -28,8 +28,15 @@ class Client:
     """Talks to one sandboxd node (and, transparently, its cluster)."""
 
     def __init__(
-        self, addr: str, api_token: str = "", timeout: float = 120.0, *, ssl_context: ssl.SSLContext | None = None
+        self,
+        addr: str,
+        api_token: str = "",
+        timeout: float = 120.0,
+        *,
+        ssl_context: ssl.SSLContext | None = None,
+        keep_alive: float = 30.0,
     ) -> None:
+        """keep_alive keeps a handle's idle relay connection, which holds the sandbox's idle clock; 0 dials per call."""
         endpoint = _endpoint_url(addr.split(",")[0].strip())
         self.addr = endpoint.geturl().removeprefix("http://")
         self._scheme = endpoint.scheme
@@ -39,6 +46,7 @@ def __init__(
         self._opener: urllib.request.OpenerDirector | None = None
         self.api_token = api_token
         self.timeout = timeout
+        self.keep_alive = keep_alive
 
     def new(
         self,
diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py
index 5c87ffab..d1f21596 100644
--- a/sdk/python/cocoonsandbox/conn.py
+++ b/sdk/python/cocoonsandbox/conn.py
@@ -3,8 +3,10 @@
 from __future__ import annotations
 
 import contextlib
+import select
 import socket
 import ssl
+import threading
 import time
 import urllib.parse
 from collections.abc import Iterator
@@ -13,6 +15,8 @@
 from .errors import APIError, ProtocolError, SilkdError
 from .frames import MAX_FRAME, decode_response, encode_request
 
+KEEP_ALIVE_CONNS = 8
+
 _CloseableT = TypeVar("_CloseableT", bound="_Closeable")
 
 
@@ -69,6 +73,14 @@ def recv_until(self, *terminal: str) -> Iterator[dict[str, Any]]:
             if frame["type"] in terminal:
                 return
 
+    def quiet(self) -> bool:
+        """Reports whether the peer has neither hung up nor spoken since the last frame."""
+        try:
+            readable, _, _ = select.select([self._sock], [], [], 0)
+        except (OSError, ValueError):
+            return False
+        return not readable
+
     def close(self) -> None:
         self.abort()
         try:
@@ -77,6 +89,60 @@ def close(self) -> None:
             self._sock.close()
 
 
+class ConnPool:
+    """Parks a handle's idle relay connections between calls; one sweeper timer closes them as they expire."""
+
+    def __init__(self, idle: float) -> None:
+        self._idle = idle
+        self._lock = threading.Lock()
+        self._parked: list[tuple[float, Conn]] = []
+        self._sweep: threading.Timer | None = None
+
+    def take(self) -> Conn | None:
+        while True:
+            with self._lock:
+                if not self._parked:
+                    return None
+                expires, conn = self._parked.pop()
+            if time.monotonic() < expires and conn.quiet():
+                return conn
+            conn.close()
+
+    def park(self, conn: Conn) -> None:
+        with self._lock:
+            if self._idle > 0 and len(self._parked) < KEEP_ALIVE_CONNS:
+                self._parked.append((time.monotonic() + self._idle, conn))
+                if self._sweep is None:
+                    self._arm(self._idle)
+                return
+        conn.close()
+
+    def drain(self) -> None:
+        with self._lock:
+            parked, self._parked = self._parked, []
+            if self._sweep is not None:
+                self._sweep.cancel()
+                self._sweep = None
+        for _, conn in parked:
+            conn.close()
+
+    def _arm(self, delay: float) -> None:
+        self._sweep = threading.Timer(delay, self._evict)
+        self._sweep.daemon = True
+        self._sweep.start()
+
+    def _evict(self) -> None:
+        now = time.monotonic()
+        with self._lock:
+            expired = [conn for expires, conn in self._parked if expires <= now]
+            self._parked = [entry for entry in self._parked if entry[0] > now]
+            self._sweep = None
+            if self._parked:
+                self._arm(self._parked[0][0] - now)
+        for conn in expired:
+            conn.close()
+
+
 def dial_agent(
     addr: str,
     sandbox_id: str,
diff --git a/sdk/python/cocoonsandbox/frames.py b/sdk/python/cocoonsandbox/frames.py
index a1cea6bf..2bfcefce 100644
--- a/sdk/python/cocoonsandbox/frames.py
+++ b/sdk/python/cocoonsandbox/frames.py
@@ -8,6 +8,8 @@
 from typing import Any
 
 PROTO_VERSION = 1
+# the info proto from which silkd serves RPCs back to back on one connection
+KEEP_ALIVE_PROTO = 2
 MAX_FRAME = 8 * 1024 * 1024
 FS_CHUNK = 256 * 1024
 # bulk streams chunk larger than silkd's FS_CHUNK: fewer frames per byte, still under MAX_FRAME after base64.
diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py
index e8ab84cb..0d6e988c 100644
--- a/sdk/python/cocoonsandbox/sandbox.py
+++ b/sdk/python/cocoonsandbox/sandbox.py
@@ -10,9 +10,9 @@
 from typing import TYPE_CHECKING, Any, cast
 
 from .checkpoint import Checkpoint
-from .conn import Conn, _Closeable
-from .errors import APIError, ExitError, ProtocolError, SandboxError
-from .frames import BULK_CHUNK, FS_CHUNK
+from .conn import Conn, ConnPool, _Closeable
+from .errors import APIError, ExitError, ProtocolError, SandboxError, SilkdError
+from .frames import BULK_CHUNK, FS_CHUNK, KEEP_ALIVE_PROTO
 from .template import Template
 
 if TYPE_CHECKING:
@@ -41,6 +41,8 @@ def __init__(
         self.from_checkpoint = from_checkpoint
         self.template_digest = template_digest
         self.volumes = [dict(volume) for volume in volumes or []]
+        self._pool = ConnPool(client.keep_alive)
+        self._proto = 0
 
     def __enter__(self) -> Sandbox:
         return self
@@ -97,13 +99,14 @@ def run(
         deadline = None if timeout is None else time.monotonic() + timeout
         expired = threading.Event()
         try:
-            conn = self._dial(deadline)
+            conn = self._connect(deadline)
         except (ProtocolError, TimeoutError):
             if deadline is not None and time.monotonic() >= deadline:
                 raise TimeoutError(f"command did not finish within {timeout}s") from None
             raise
-        with conn:
+        with self._lease(conn):
             watchdog = _arm_watchdog(conn, deadline, expired)
+            pump = threading.Thread(target=_feed_stdin, args=(conn, stdin), daemon=True) if stdin else None
             try:
                 conn.send(
                     "exec",
@@ -114,17 +117,25 @@ def run(
                     detach=False,
                     session=session or None,
                 )
-                pump = threading.Thread(target=_feed_stdin, args=(conn, stdin), daemon=True)
-                pump.start()
+                if pump is not None:
+                    pump.start()
+                else:
+                    with contextlib.suppress(OSError):
+                        conn.send("stdin_close")
                 code = _pump_stdio(conn, on_stdout, on_stderr)
             except (ProtocolError, OSError):
                 if expired.is_set():
                     raise TimeoutError(f"command did not finish within {timeout}s") from None
                 raise
+            except SilkdError:
+                if pump is not None:
+                    pump.join()
+                raise
             finally:
                 if watchdog is not None:
                     watchdog.cancel()
-        pump.join()  # the closed conn fails a stalled send, so this cannot hang
+            if pump is not None:
+                pump.join()  # the pump's frames must not land on the next RPC
         if code is None:
             raise ProtocolError("exec stream ended without an exit frame")
         return code
@@ -151,7 +162,7 @@ def logs(
         on_stderr: Callable[[bytes], object] | None = None,
     ) -> int | None:
         """Replays buffered output and returns the exit code, or None if the process still runs."""
-        return self._drain_proc("logs", pid, on_stdout, on_stderr)
+        return self._drain_proc("logs", pid, on_stdout, on_stderr, trailing_done=True)
 
     def attach(
         self,
@@ -164,19 +175,19 @@ def attach(
 
     def write_file(self, path: str, data: bytes, mode: int | None = None) -> None:
         """Writes data to path atomically (temp + rename on the guest)."""
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send("fs_write", path=path, mode=mode)
             _send_chunks(conn, data)
             conn.send("data_end")
             _expect(conn, "done")
 
     def read_file(self, path: str) -> bytes:
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send("fs_read", path=path)
             return _drain_data(conn)
 
     def list_dir(self, path: str) -> list[dict[str, Any]]:
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send("fs_list", path=path)
             entries: list[dict[str, Any]] = []
             for frame in conn.recv_until("done"):
@@ -198,7 +209,7 @@ def rename(self, src: str, dst: str) -> None:
 
     def push(self, dest: str, tar_stream: bytes) -> None:
         """Extracts a tar stream into dest; a truncated stream leaves dest untouched."""
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send("fs_push", dest=dest)
             _send_chunks(conn, tar_stream, chunk=BULK_CHUNK)
             conn.send("data_end")
@@ -206,7 +217,7 @@ def push(self, dest: str, tar_stream: bytes) -> None:
 
     def pull(self, path: str) -> bytes:
         """Returns path (file or tree) as a tar archive."""
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send("fs_pull", path=path)
             return _drain_data(conn)
 
@@ -215,14 +226,14 @@ def find(self, path: str, pattern: str, glob: str = "") -> list[dict[str, Any]]:
 
     def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict[str, Any]]:
         """Yields matches as they stream."""
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send("fs_find", path=path, pattern=pattern, glob=glob or None)
             for f in conn.recv_until("done"):
                 if f["type"] == "match":
                     yield f
 
     def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict[str, Any]]:
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send("fs_replace", files=files, pattern=pattern, replacement=replacement)
             return [f for f in conn.recv_until("done") if f["type"] == "replaced"]
 
@@ -284,6 +295,7 @@ def fork(self, count: int, ttl_seconds: int = 0) -> list[Sandbox]:
 
     def hibernate(self) -> None:
         """Snapshots and stops the VM; the next guest call restores its state."""
+        self._pool.drain()
         self._client._request(
             self.owner, "POST", f"/v1/sandboxes/{self.id}/hibernate", None, "hibernate", bearer=self.token
         )
@@ -347,6 +359,7 @@ def dial_port(self, port: int) -> PortConn:
 
     def close(self) -> None:
         """Releases the sandbox; its VM is destroyed."""
+        self._pool.drain()
         try:
             self._client._request(
                 self.owner, "POST", f"/v1/sandboxes/{self.id}/release", None, "release", bearer=self.token
@@ -358,8 +371,49 @@ def close(self) -> None:
     def _dial(self, deadline: float | None = None) -> Conn:
         return self._client._dial(self.owner, self.id, self.token, deadline)
 
+    def _connect(self, deadline: float | None = None) -> Conn:
+        """Takes a parked connection or dials one, asking the daemon's proto on a handle's first kept dial."""
+        conn = self._pool.take()
+        if conn is not None:
+            return conn
+        conn = self._dial(deadline)
+        if self._proto or self._client.keep_alive <= 0:
+            return conn
+        try:
+            conn.send("info")
+            proto = int(_expect(conn, "info").get("proto") or 1)
+        except Exception:
+            conn.close()
+            raise
+        self._proto = proto
+        if proto >= KEEP_ALIVE_PROTO:
+            return conn
+        conn.close()
+        return self._dial(deadline)
+
+    @contextlib.contextmanager
+    def _lease(self, conn: Conn | None = None) -> Iterator[Conn]:
+        """Runs one RPC on conn, dialed when absent; a terminal frame parks it and anything else drops it."""
+        if conn is None:
+            conn = self._connect()
+        try:
+            yield conn
+        except SilkdError:
+            self._park(conn)
+            raise
+        except BaseException:
+            conn.close()
+            raise
+        self._park(conn)
+
+    def _park(self, conn: Conn) -> None:
+        if self._proto >= KEEP_ALIVE_PROTO:
+            self._pool.park(conn)
+        else:
+            conn.close()
+
     def _open_stream(self, op: str, expect: str = "ready", **fields: object) -> tuple[Conn, dict[str, Any]]:
-        conn = self._dial()
+        conn = self._connect()
         try:
             conn.send(op, **fields)
             frame = _expect(conn, expect)
@@ -409,7 +463,7 @@ def pump_out() -> None:
                 local.close()
 
     def _call(self, op: str, expect: str, **fields: object) -> dict[str, Any]:
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send(op, **fields)
             return _expect(conn, expect)
 
@@ -422,10 +476,14 @@ def _drain_proc(
         pid: int,
         on_stdout: Callable[[bytes], object] | None,
         on_stderr: Callable[[bytes], object] | None,
+        trailing_done: bool = False,
     ) -> int | None:
-        with self._dial() as conn:
+        with self._lease() as conn:
             conn.send(op, pid=pid)
-            return _pump_stdio(conn, on_stdout, on_stderr)
+            code = _pump_stdio(conn, on_stdout, on_stderr)
+            if trailing_done and code is not None:
+                _expect(conn, "done")
+            return code
 
 
 class Session(_Closeable):
diff --git a/sdk/python/tests/conftest.py b/sdk/python/tests/conftest.py
index 8155508a..6101223b 100644
--- a/sdk/python/tests/conftest.py
+++ b/sdk/python/tests/conftest.py
@@ -1,13 +1,38 @@
-"""Fixtures shared by the cluster-behavior suites: in-process fake nodes and
-addresses that refuse a connection."""
+"""Fixtures and helpers shared by the suites: in-process fake nodes, addresses
+that refuse a connection, and the relay-side upgrade handshake."""
 
 import socket
 import threading
+import time
 from http.server import HTTPServer
+from typing import BinaryIO, Callable
 
 import pytest
 from test_client import FakeNode
 
+from cocoonsandbox import Client, Sandbox
+
+
+def sandbox_at(addr: str, **client_kwargs) -> Sandbox:
+    return Sandbox(client=Client(addr, **client_kwargs), id="sb_1", token="tok", owner=addr)
+
+
+def accept_upgrade(conn: socket.socket) -> BinaryIO:
+    reader = conn.makefile("rb")
+    while reader.readline() not in (b"\r\n", b""):
+        pass
+    conn.sendall(b"HTTP/1.1 101 Switching Protocols\r\n\r\n")
+    return reader
+
+
+def wait_until(cond: Callable[[], bool], message: str) -> None:
+    deadline = time.monotonic() + 3
+    while time.monotonic() < deadline:
+        if cond():
+            return
+        time.sleep(0.01)
+    raise AssertionError(message)
+
 
 @pytest.fixture
 def spawn_node():
diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py
new file mode 100644
index 00000000..8710ae4f
--- /dev/null
+++ b/sdk/python/tests/test_keepalive.py
@@ -0,0 +1,140 @@
+"""One relay connection serves a handle's calls back to back; the fallbacks dial afresh."""
+
+import base64
+import contextlib
+import json
+import socket
+import threading
+
+from conftest import accept_upgrade, sandbox_at, wait_until
+
+from cocoonsandbox.frames import KEEP_ALIVE_PROTO
+
+INPUT_OPS = ("stdin", "stdin_close", "data", "data_end")
+
+
+def test_calls_share_one_connection():
+    agent = FakeAgent()
+    sb = sandbox_at(agent.addr)
+    try:
+        for _ in range(3):
+            assert sb.stat("/")["kind"] == "dir"
+        assert sb.exec("echo", "42") == "42\n"
+    finally:
+        sb._pool.drain()
+        agent.stop()
+    assert agent.upgrades == 1
+
+
+def test_old_daemon_dials_per_call():
+    agent = FakeAgent(proto=1)
+    sb = sandbox_at(agent.addr)
+    try:
+        for _ in range(3):
+            assert sb.exec("echo", "42") == "42\n"
+    finally:
+        agent.stop()
+    assert agent.upgrades == 4, "the proto probe plus one dial per call"
+
+
+def test_keep_alive_off_dials_per_call():
+    agent = FakeAgent()
+    sb = sandbox_at(agent.addr, keep_alive=0)
+    try:
+        for _ in range(3):
+            sb.stat("/")
+    finally:
+        agent.stop()
+    assert agent.upgrades == 3
+
+
+def test_idle_connection_closes():
+    agent = FakeAgent()
+    sb = sandbox_at(agent.addr, keep_alive=0.05)
+    try:
+        sb.stat("/")
+        wait_until(lambda: agent.closed == 1, "idle connection still open after the keep-alive window")
+    finally:
+        sb._pool.drain()
+        agent.stop()
+
+
+def test_peer_hang_up_is_noticed_before_reuse():
+    agent = FakeAgent(hang_up_after=2)
+    sb = sandbox_at(agent.addr)
+    try:
+        sb.stat("/")
+        wait_until(lambda: agent.hangups == 1, "the fake never hung up")
+        assert sb.stat("/")["kind"] == "dir"
+    finally:
+        sb._pool.drain()
+        agent.stop()
+    assert agent.upgrades == 2
+
+
+def test_close_drains_the_parked_connection(monkeypatch):
+    agent = FakeAgent()
+    sb = sandbox_at(agent.addr)
+    monkeypatch.setattr(sb._client, "_request", lambda *args, **kwargs: {})
+    try:
+        sb.stat("/")
+        sb.close()
+        wait_until(lambda: agent.closed == 1, "parked connection survived close")
+    finally:
+        agent.stop()
+
+
+class FakeAgent:
+    """A relay-side silkd stand-in; proto 1 half-closes after one RPC and drains like the relay does."""
+
+    def __init__(self, proto: int = KEEP_ALIVE_PROTO, hang_up_after: int = 0) -> None:
+        self.proto = proto
+        self.hang_up_after = hang_up_after
+        self.upgrades = 0
+        self.hangups = 0
+        self.closed = 0
+        self._server = socket.create_server(("127.0.0.1", 0))
+        self.addr = f"127.0.0.1:{self._server.getsockname()[1]}"
+        threading.Thread(target=self._accept, daemon=True).start()
+
+    def stop(self) -> None:
+        self._server.close()
+
+    def _accept(self) -> None:
+        with contextlib.suppress(OSError):
+            while True:
+                conn, _ = self._server.accept()
+                self.upgrades += 1
+                threading.Thread(target=self._serve, args=(conn,), daemon=True).start()
+
+    def _serve(self, conn: socket.socket) -> None:
+        with conn, contextlib.suppress(OSError):
+            reader = accept_upgrade(conn)
+            replies = 0
+            while True:
+                line = reader.readline()
+                if not line:
+                    break
+                op = json.loads(line)["op"]
+                if op in INPUT_OPS:
+                    continue
+                for frame in self._answer(op):
+                    conn.sendall(json.dumps(frame).encode() + b"\n")
+                replies += 1
+                if self.proto < KEEP_ALIVE_PROTO or replies == self.hang_up_after:
+                    conn.shutdown(socket.SHUT_WR)
+                    self.hangups += 1
+                    while reader.readline():
+                        pass
+                    break
+        self.closed += 1
+
+    def _answer(self, op: str) -> list:
+        if op == "info":
+            return [{"type": "info", "version": "fake", "proto": self.proto, "uptime_secs": 0, "procs": 0}]
+        if op == "fs_stat":
+            return [{"type": "stat", "info": {"kind": "dir", "size": 0, "mode": 0o755, "mtime_epoch_secs": 0}}]
+        if op == "exec":
+            out = base64.b64encode(b"42\n").decode()
+            return [{"type": "started", "pid": 1}, {"type": "stdout", "data": out}, {"type": "exit", "code": 0}]
+        return [{"type": "error", "kind": "unimplemented", "message": op}]
diff --git a/sdk/python/tests/test_proc.py b/sdk/python/tests/test_proc.py
index d03e22dd..440d99b9 100644
--- a/sdk/python/tests/test_proc.py
+++ b/sdk/python/tests/test_proc.py
@@ -3,7 +3,8 @@
 
 import threading
 
-from cocoonsandbox import Client, Sandbox
+from conftest import sandbox_at
+
 from cocoonsandbox.frames import FS_CHUNK
 
 
@@ -43,7 +44,7 @@ def test_attach_returns_exit_code(monkeypatch):
 
 
 def test_run_pumps_stdin_while_reading_output(monkeypatch):
-    sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1")
+    sb = sandbox_at("127.0.0.1:1", keep_alive=0)
     blocking = BlockingStdinConn([{"type": "exit", "code": 0}], buffer_frames=1)
     monkeypatch.setattr(sb, "_dial", lambda deadline=None: blocking)
 
@@ -103,7 +104,7 @@ def recv(self):
 
 
 def fake_sandbox(monkeypatch, frames):
-    sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1")
+    sb = sandbox_at("127.0.0.1:1", keep_alive=0)
     conn = FakeConn(frames)
     monkeypatch.setattr(sb, "_dial", lambda deadline=None: conn)
     return sb, conn
diff --git a/sdk/python/tests/test_stream.py b/sdk/python/tests/test_stream.py
index 15a3858f..fb9e966c 100644
--- a/sdk/python/tests/test_stream.py
+++ b/sdk/python/tests/test_stream.py
@@ -7,8 +7,9 @@
 import time
 
 import pytest
+from conftest import accept_upgrade, sandbox_at
 
-from cocoonsandbox import Client, Sandbox
+from cocoonsandbox import Sandbox
 
 TIMEOUT = 0.2
 
@@ -30,13 +31,17 @@ def send(self, op: str, **fields) -> None:
     def abort(self) -> None:
         self.aborted.set()
 
+    def close(self) -> None:
+        pass
+
+
+def legacy_sandbox(addr: str) -> Sandbox:
+    return sandbox_at(addr, timeout=TIMEOUT, keep_alive=0)
+
 
 def serve_port_forward(server: socket.socket, quiet: float, ops: list[str]) -> None:
     conn, _ = server.accept()
-    reader = conn.makefile("rb")
-    while reader.readline() not in (b"\r\n", b""):
-        pass
-    conn.sendall(b"HTTP/1.1 101 Switching Protocols\r\n\r\n")
+    reader = accept_upgrade(conn)
     ops.append(json.loads(reader.readline())["op"])
     conn.sendall(b'{"type":"ready"}\n')
     time.sleep(quiet)
@@ -46,10 +51,7 @@ def serve_port_forward(server: socket.socket, quiet: float, ops: list[str]) -> N
 
 def serve_started_then_hang(server: socket.socket, quiet: float) -> None:
     conn, _ = server.accept()
-    reader = conn.makefile("rb")
-    while reader.readline() not in (b"\r\n", b""):
-        pass
-    conn.sendall(b"HTTP/1.1 101 Switching Protocols\r\n\r\n")
+    reader = accept_upgrade(conn)
     reader.readline()
     conn.sendall(b'{"type":"started","pid":7}\n')
     time.sleep(quiet)
@@ -68,7 +70,7 @@ def test_port_stream_outlives_the_client_timeout():
     addr = f"127.0.0.1:{server.getsockname()[1]}"
     ops: list[str] = []
     threading.Thread(target=serve_port_forward, args=(server, 3 * TIMEOUT, ops), daemon=True).start()
-    sb = Sandbox(client=Client(addr, timeout=TIMEOUT), id="sb_1", token="tok", owner=addr)
+    sb = legacy_sandbox(addr)
     try:
         with sb.dial_port(5000) as port:
             assert port.recv() == b"late"
@@ -82,7 +84,7 @@ def test_run_timeout_cuts_a_silent_command():
     server = socket.create_server(("127.0.0.1", 0))
     addr = f"127.0.0.1:{server.getsockname()[1]}"
     threading.Thread(target=serve_started_then_hang, args=(server, 5 * TIMEOUT), daemon=True).start()
-    sb = Sandbox(client=Client(addr, timeout=TIMEOUT), id="sb_1", token="tok", owner=addr)
+    sb = legacy_sandbox(addr)
     started = time.monotonic()
     try:
         with pytest.raises(TimeoutError):
@@ -93,7 +95,7 @@ def test_run_timeout_cuts_a_silent_command():
 
 
 def test_run_timeout_cuts_a_blocked_exec_send(monkeypatch):
-    sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1")
+    sb = legacy_sandbox("127.0.0.1:1")
     conn = BlockedSendConn()
     monkeypatch.setattr(sb, "_dial", lambda deadline=None: conn)
     with pytest.raises(TimeoutError):
@@ -102,7 +104,7 @@ def test_run_timeout_cuts_a_blocked_exec_send(monkeypatch):
 
 
 def test_run_rejects_a_non_positive_timeout():
-    sb = Sandbox(client=Client("127.0.0.1:1", timeout=TIMEOUT), id="sb_1", token="tok", owner="127.0.0.1:1")
+    sb = legacy_sandbox("127.0.0.1:1")
     with pytest.raises(ValueError):
         sb.run(["true"], timeout=0)
 
@@ -111,7 +113,7 @@ def test_dial_is_still_bounded_by_the_client_timeout():
     server = socket.create_server(("127.0.0.1", 0))
     addr = f"127.0.0.1:{server.getsockname()[1]}"
     threading.Thread(target=serve_silence, args=(server,), daemon=True).start()
-    sb = Sandbox(client=Client(addr, timeout=TIMEOUT), id="sb_1", token="tok", owner=addr)
+    sb = legacy_sandbox(addr)
     started = time.monotonic()
     try:
         with pytest.raises(OSError):
diff --git a/sdk/python/tests/test_wire_binding.py b/sdk/python/tests/test_wire_binding.py
index 4327ec2a..f6cecac4 100644
--- a/sdk/python/tests/test_wire_binding.py
+++ b/sdk/python/tests/test_wire_binding.py
@@ -16,8 +16,9 @@
 import threading
 
 import pytest
+from conftest import sandbox_at
 
-from cocoonsandbox import Client, Lsp, Pty, Sandbox, Session
+from cocoonsandbox import Lsp, Pty, Session
 from cocoonsandbox.conn import Conn
 from cocoonsandbox.frames import PROTO_VERSION
 
@@ -64,7 +65,7 @@
     ("req_exec_detach", [{"type": "started", "pid": 7}], lambda sb, f: sb.spawn(*f["argv"])),
     ("req_ps", [{"type": "procs", "procs": []}], lambda sb, f: sb.ps()),
     ("req_kill", [{"type": "done"}], lambda sb, f: sb.kill(f["pid"], signal=f["signal"])),
-    ("req_logs", [{"type": "exit", "code": 0}], lambda sb, f: sb.logs(f["pid"])),
+    ("req_logs", [{"type": "exit", "code": 0}, {"type": "done"}], lambda sb, f: sb.logs(f["pid"])),
     ("req_attach", [{"type": "exit", "code": 0}], lambda sb, f: sb.attach(f["pid"])),
     ("req_fs_watch", [{"type": "ready"}], lambda sb, f: sb.watch(f["path"], recursive=f["recursive"]).close()),
     ("req_git_branch", [{"type": "done"}], lambda sb, f: sb.git_create_branch(f["path"], f["name"])),
@@ -119,7 +120,7 @@ def test_enum_value_sets_match_corpus():
 
 def test_git_branch_actions_come_from_the_corpus(monkeypatch):
     enums = json.loads((FIXTURES / "enums.json").read_text())
-    sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1")
+    sb = sandbox_at("127.0.0.1:1", keep_alive=0)
     sent = []
     monkeypatch.setattr(sb, "_dial", lambda deadline=None: BranchActionConn(sent))
 
@@ -155,6 +156,9 @@ def recv(self):
     def recv_until(self, *terminal):
         yield self.recv()
 
+    def close(self):
+        pass
+
 
 def fake_sandbox(monkeypatch, replies):
     """A Sandbox whose _dial yields a real Conn over a socketpair; a guest
@@ -181,6 +185,6 @@ def guest():
     thread = threading.Thread(target=guest, daemon=True)
     thread.start()
 
-    sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1")
+    sb = sandbox_at("127.0.0.1:1", keep_alive=0)
     monkeypatch.setattr(sb, "_dial", lambda deadline=None: Conn(client_sock, client_sock.makefile("rb")))
     return sb, sent, thread