diff --git a/agentix/__init__.py b/agentix/__init__.py
index 418495b..476faa3 100644
--- a/agentix/__init__.py
+++ b/agentix/__init__.py
@@ -22,8 +22,12 @@
     register_provider,
 )
 from agentix.runtime.client import (
+    CallCancelled,
     CallTimeout,
+    Failed,
+    Ok,
     RemoteCallError,
+    Result,
     RuntimeClient,
     RuntimeUnreachable,
     WorkerExited,
@@ -39,12 +43,16 @@
 __all__ = [
     "AsyncClientNamespace",
     "BundleDeployer",
+    "CallCancelled",
     "CallTimeout",
     "DeployedBundle",
+    "Failed",
     "Namespace",
+    "Ok",
     "RemoteCallable",
     "RemoteCallError",
     "RemoteSioError",
+    "Result",
     "RuntimeClient",
     "RuntimeUnreachable",
     "Sandbox",
diff --git a/agentix/provider/base.py b/agentix/provider/base.py
index 4790353..3cff659 100644
--- a/agentix/provider/base.py
+++ b/agentix/provider/base.py
@@ -40,7 +40,7 @@ async def get(self, sandbox_id): ...
 from agentix.provider._plugin import Registry
 
 if TYPE_CHECKING:
-    from agentix.runtime.client import RuntimeClient
+    from agentix.runtime.client import Result, RuntimeClient
     from agentix.runtime.shared.models import HealthResponse
 
 P = ParamSpec("P")
@@ -215,6 +215,16 @@ async def remote(
         """Execute `fn(*args, **kwargs)` in this sandbox and return its result."""
         return await self._runtime_client().remote(fn, *args, **kwargs)
 
+    async def try_remote(
+        self,
+        fn: Callable[P, R] | Callable[P, Awaitable[R]],
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> Result[R]:
+        """Execute `fn` and return a `Result[R]` (`Ok | Failed`) instead of
+        raising on a terminal error — see `RuntimeClient.try_remote`."""
+        return await self._runtime_client().try_remote(fn, *args, **kwargs)
+
     async def health(self) -> HealthResponse:
         return await self._runtime_client().health()
 
@@ -264,12 +274,24 @@ async def session(
                 result = await sandbox.remote(agent.run, task=task)
         """
         sandbox = await self.create(config)
+        # Contract: `create()` only provisions the sandbox; it must NOT
+        # materialize the RuntimeClient (the lazy `_runtime_client()` reads
+        # `call_deadline` at first `remote()`). That is what lets us stamp the
+        # deadline here, post-create. A provider that eagerly connected inside
+        # `create()` would bake in `call_deadline=None` — such a provider must
+        # accept the deadline through `create()` instead.
         sandbox.call_deadline = call_deadline
         try:
             yield sandbox
         finally:
-            await sandbox.aclose()
-            await self.delete(sandbox.sandbox_id)
+            # `delete()` must run even if `aclose()` raises (e.g. an httpx
+            # pool error or CancelledError during shutdown); otherwise the
+            # container and its reserved port leak — the exact failure the
+            # never-leak contract targets.
+            try:
+                await sandbox.aclose()
+            finally:
+                await self.delete(sandbox.sandbox_id)
 
 
 @runtime_checkable
diff --git a/agentix/runtime/PROTOCOL.md b/agentix/runtime/PROTOCOL.md
index 784d4b7..e30fdc9 100644
--- a/agentix/runtime/PROTOCOL.md
+++ b/agentix/runtime/PROTOCOL.md
@@ -1,7 +1,7 @@
 # Agentix RPC Protocol
 
 The runtime wire contract for `RuntimeClient.remote(fn, *args, **kwargs)`.
-Tests in `tests/test_rpc_protocol.py` enforce these rules.
+Tests in `tests/runtime/test_protocol.py` enforce these rules.
 
 ## Callable Reference
 
@@ -36,7 +36,7 @@ await client.remote(run, seed=42)
 | --- | --- | --- |
 | `GET /health` | health probe | HTTP JSON |
 | `POST /call` | internal short-call fast path | HTTP msgpack |
-| Socket.IO `/rpc` | `c.remote()` RPC | msgpack-wrapped `call` / `call:result` / `call:error` / `cancel` |
+| Socket.IO `/rpc` | `c.remote()` RPC | msgpack-wrapped `call` / `call:result` / `call:error` / `cancel` / `resume` / `ack` |
 | Socket.IO `/trace`, `/log`, `/<plugin>` | side channels | plugin-defined events (msgpack payloads) |
 | worker private pipe | runtime ↔ worker | length-prefixed msgpack frames |
 
@@ -54,11 +54,22 @@ call          {call_id, callable, arguments}
 call:result   {call_id, value}                  # value is pickle.dumps(result)
 call:error    {call_id, error}
 cancel        {call_id}
+resume        {call_ids}                        # host → server on (re)connect
+ack           {call_id}                         # host → server, frees the retained result
 ```
 
 `call_id` correlates request ↔ response. Cancellation produces a
 `call:error` with `error.cancelled=True`.
 
+`resume` and `ack` carry the reliability contract. On every (re)connect the
+host emits `resume` with the `call_ids` it is still awaiting; the server
+resolves each one — a retained terminal state is replayed as its
+`call:result` / `call:error`, a still-running call is left alone, and an
+evicted or unknown `call_id` gets a definite `call:error`
+(`error.type="ResultUnavailable"`) rather than silence, so a reconnecting
+host never hangs. After consuming a result the host emits `ack`, which lets
+the server drop it from its bounded retain buffer (`pending_results`).
+
 Trace, log, and plugin traffic use their own namespaces on the same
 Socket.IO connection. Sandbox plugins emit through `agentix.sio`; the
 worker forwards `sio_emit` / `sio_open` frames to the server, which
@@ -95,6 +106,10 @@ away from user subprocesses).
 4. **Worker death closes calls.** If the worker subprocess exits, the
    runtime fails every in-flight call with `WorkerExited` so the
    client never hangs.
+5. **No silent loss.** A `resume` for a `call_id` the runtime no longer
+   holds (its result was evicted under cap, or the id is unknown) gets a
+   definite `call:error` (`error.type="ResultUnavailable"`) — never
+   silence.
 
 ## Error Model
 
@@ -111,9 +126,14 @@ away from user subprocesses).
 
 Client mapping:
 
-- `cancelled=True` → `asyncio.CancelledError`
+- `cancelled=True` → `agentix.CallCancelled` (a `RemoteCallError` subclass —
+  a terminal *server-side* state, distinct from local task cancellation)
+- `type="WorkerDied"` → `agentix.WorkerExited`
 - everything else → `agentix.RemoteCallError`
 
+`try_remote()` returns the same terminal states as a `Result[R]`
+(`Ok(value) | Failed(error)`) instead of raising.
+
 ## Lifecycle
 
 | Edge | Connect | Cleanup |
diff --git a/agentix/runtime/client/__init__.py b/agentix/runtime/client/__init__.py
index ba8ffcc..68a1a56 100644
--- a/agentix/runtime/client/__init__.py
+++ b/agentix/runtime/client/__init__.py
@@ -14,16 +14,22 @@
 """
 
 from agentix.runtime.client.client import (
+    CallCancelled,
     CallTimeout,
     RemoteCallError,
     RuntimeClient,
     RuntimeUnreachable,
     WorkerExited,
 )
+from agentix.runtime.client.result import Failed, Ok, Result
 
 __all__ = [
+    "CallCancelled",
     "CallTimeout",
+    "Failed",
+    "Ok",
     "RemoteCallError",
+    "Result",
     "RuntimeClient",
     "RuntimeUnreachable",
     "WorkerExited",
diff --git a/agentix/runtime/client/client.py b/agentix/runtime/client/client.py
index 51178e8..9b07caa 100644
--- a/agentix/runtime/client/client.py
+++ b/agentix/runtime/client/client.py
@@ -32,6 +32,7 @@
 import socketio
 from socketio.exceptions import ConnectionError as SioConnectionError
 
+from agentix.runtime.client.result import Failed, Ok, Result
 from agentix.runtime.shared import MAX_MESSAGE_BYTES
 from agentix.runtime.shared.callables import RemoteCallable, display_name_for
 from agentix.runtime.shared.codec import pack, unpack
@@ -90,9 +91,18 @@ def returncode(self) -> int | None:
         return self.error.returncode
 
 
+class CallCancelled(RemoteCallError):
+    """The runtime reported the call as cancelled — a terminal *server-side*
+    state, distinct from local `asyncio` task cancellation. Subclasses
+    `RemoteCallError` so it rides the normal terminal-state path: `remote()`
+    raises it and `try_remote()` surfaces it as `Failed`. (A bare
+    `asyncio.CancelledError` here would escape the `Ok | Failed` sum type and
+    read as local cancellation.)"""
+
+
 def _raise_remote_error(display_name: str, error: RemoteError):
     if error.cancelled:
-        raise asyncio.CancelledError(error.message)
+        raise CallCancelled(display_name=display_name, error=error)
     if error.type == "WorkerDied":
         raise WorkerExited(display_name=display_name, error=error)
     raise RemoteCallError(display_name=display_name, error=error)
@@ -190,9 +200,13 @@ def _register_core_namespaces(self) -> None:
     # ── lifecycle ────────────────────────────────────────────────
 
     async def close(self):
-        if self._sio is not None and self._sio.connected:
+        if self._sio is not None:
+            # `shutdown()` (not `disconnect()`): it also stops a client that is
+            # mid-reconnect — aborting and awaiting the background reconnect
+            # task. `disconnect()` is a library no-op on a dropped transport
+            # and would leave the abandoned client reconnecting forever.
             with contextlib.suppress(BaseException):
-                await self._sio.disconnect()
+                await self._sio.shutdown()
         await self._client.aclose()
 
     async def __aenter__(self):
@@ -252,15 +266,25 @@ async def _try_http_fast_path(
 
         wait = self._http_sync_budget_ms / 1000  # ms → seconds for `Prefer: wait=`
         wait_token = str(int(wait)) if wait == int(wait) else str(wait)
-        r = await self._client.post(
-            "/call",
-            content=pack(payload),
-            headers={
-                "content-type": "application/msgpack",
-                "prefer": f"respond-async, wait={wait_token}",
-            },
-        )
-        r.raise_for_status()
+        try:
+            r = await self._client.post(
+                "/call",
+                content=pack(payload),
+                headers={
+                    "content-type": "application/msgpack",
+                    "prefer": f"respond-async, wait={wait_token}",
+                },
+            )
+            r.raise_for_status()
+        except httpx.HTTPError:
+            # The fast path is opportunistic: any transport/status failure
+            # falls back to the Socket.IO channel, whose connect path owns the
+            # typed RuntimeUnreachable conversion. A raw httpx error must
+            # never escape remote() — it would also breach try_remote()'s
+            # `Result` contract. Execution-once holds across the fallback:
+            # the server gates on `call_id in calls or pending_results`.
+            logger.debug("HTTP fast path unavailable; falling back to SIO", exc_info=True)
+            return "fallback", None
 
         # 202: not done within the budget — the result follows on SIO.
         if r.status_code == 202:
@@ -327,7 +351,14 @@ async def remote(
                         terminated = True
                         return cast(R, _unpickle_value(data.get("value")))
                     if kind == "error":
-                        err = RemoteError.model_validate(data["error"])
+                        # Defensive: a malformed frame with no `error` payload
+                        # must still resolve to a typed terminal error, not a
+                        # bare KeyError escaping `remote()` / `try_remote()`.
+                        raw_err = data.get("error") or {
+                            "type": "MalformedError",
+                            "message": "runtime sent a call:error with no error payload",
+                        }
+                        err = RemoteError.model_validate(raw_err)
                         terminated = True
                         _raise_remote_error(display_name, err)
                     if kind == "fatal":
@@ -349,6 +380,22 @@ async def remote(
                         namespace=RPC_NAMESPACE,
                     )
 
+    async def try_remote(
+        self,
+        fn: Callable[P, R] | Callable[P, Awaitable[R]],
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> Result[R]:
+        """Like `remote()`, but returns a `Result[R]` (`Ok | Failed`) instead
+        of raising on a terminal error — for callers that branch on the
+        outcome with `match`. Misuse (a non-importable callable) and *local*
+        task cancellation still raise; a server-side cancel arrives as
+        `Failed(CallCancelled)`."""
+        try:
+            return Ok(await self.remote(fn, *args, **kwargs))
+        except (RemoteCallError, CallTimeout, RuntimeUnreachable) as exc:
+            return Failed(exc)
+
     # ── Socket.IO connection management ─────────────────────────
 
     async def _ensure_sio(self) -> socketio.AsyncClient:
@@ -357,6 +404,20 @@ async def _ensure_sio(self) -> socketio.AsyncClient:
         async with self._sio_lock:
             if self._sio is not None and self._sio.connected:
                 return self._sio
+            if self._sio is not None:
+                # A handle exists but is disconnected (a transport drop mid
+                # reconnect). Shut it down before building a fresh client —
+                # overwriting it without stopping it leaks its aiohttp session
+                # and background reconnect task, and would leave a second live
+                # `/rpc` socket once the abandoned client reconnects on its own.
+                # Must be `shutdown()`: `disconnect()` is a library no-op on a
+                # dropped transport and doesn't touch the reconnect loop.
+                # In-flight calls are not lost: their `self._pending` entries
+                # persist and the fresh client's `connect` handler re-emits
+                # `resume` to recover their results.
+                with contextlib.suppress(BaseException):
+                    await self._sio.shutdown()
+                self._sio = None
             # `max_msg_size` lifts the websocket's receive cap (engineio's
             # client rides aiohttp, default 4 MB) — large `c.remote`
             # payloads / plugin events otherwise kill the connection.
@@ -460,6 +521,7 @@ def _fail_pending(self, exc: BaseException) -> None:
 
 
 __all__ = [
+    "CallCancelled",
     "CallTimeout",
     "RemoteCallError",
     "RuntimeClient",
diff --git a/agentix/runtime/client/result.py b/agentix/runtime/client/result.py
new file mode 100644
index 0000000..b8dbdae
--- /dev/null
+++ b/agentix/runtime/client/result.py
@@ -0,0 +1,41 @@
+"""`Result[T]` — the typed outcome of a remote call.
+
+`remote()` raises on failure (idiomatic Python, clean happy path).
+`try_remote()` returns this `Ok | Failed` sum type instead, for callers
+that branch on the outcome at scale (a rollout harness) and want
+exhaustive matching:
+
+    match await sandbox.try_remote(solve, task=t):
+        case Ok(patch):                   use(patch)
+        case Failed(WorkerExited() as e): retry_with_more_memory(e.returncode)
+        case Failed(error):               record_failure(error)
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+@dataclass(frozen=True)
+class Ok(Generic[T]):
+    """A remote call that returned a value."""
+
+    value: T
+
+
+@dataclass(frozen=True)
+class Failed:
+    """A remote call that ended in a terminal error — carries the same
+    exception `remote()` would have raised (`RemoteCallError` /
+    `WorkerExited` / `CallTimeout` / `RuntimeUnreachable`)."""
+
+    error: Exception
+
+
+# `Ok[T] | Failed`, subscriptable as `Result[R]` (a generic union alias).
+Result = Ok[T] | Failed
+
+__all__ = ["Failed", "Ok", "Result"]
diff --git a/agentix/runtime/server/sio.py b/agentix/runtime/server/sio.py
index 75834e2..3623be5 100644
--- a/agentix/runtime/server/sio.py
+++ b/agentix/runtime/server/sio.py
@@ -38,7 +38,9 @@
 # Cap on the unacked-result cache. A host that completes calls and never acks
 # (a crashed or buggy client) would otherwise pin every result — each holding a
 # full pickled return value, up to MAX_MESSAGE_BYTES — in memory forever. Past
-# the cap, the oldest unacked entry is evicted.
+# the cap, the oldest unacked entry is evicted. Eviction is not silent loss:
+# a later `resume` for an evicted call_id gets a definite
+# `call:error type=ResultUnavailable` instead of hanging the host forever.
 _MAX_PENDING_RESULTS = 4096
 
 
@@ -79,6 +81,19 @@ def _cancelled_error(call_id: str) -> dict[str, Any]:
     }
 
 
+def _unavailable_error(call_id: str) -> dict[str, Any]:
+    return {
+        "call_id": call_id,
+        "error": RemoteError(
+            type="ResultUnavailable",
+            message=(
+                "result is no longer held by the runtime (evicted or unknown call_id); "
+                "retry as a new call"
+            ),
+        ).model_dump(),
+    }
+
+
 def _store_pending_result(
     cache: dict[str, tuple[str, dict[str, Any]]],
     call_id: str,
@@ -145,6 +160,11 @@ def make_sio(
     pending_results: dict[str, tuple[str, dict[str, Any]]] = {}
     evictions = 0  # count of cap evictions, for throttled warning
     opened_namespaces: set[str] = set()  # paths the worker has opened
+    # Keep a strong reference to every in-flight result-emit task. Bare
+    # `asyncio.create_task` in a done-callback is only weakly referenced by
+    # the loop — under GC pressure the task can be collected before it runs,
+    # and the host never receives the completed call's result.
+    emit_tasks: set[asyncio.Task] = set()
 
     async def _execute_call(payload: dict[str, Any], call_id: str) -> tuple[str, dict[str, Any]]:
         try:
@@ -182,10 +202,17 @@ async def _execute_call(payload: dict[str, Any], call_id: str) -> tuple[str, dic
         error = (resp.error or RemoteError(type="Unknown", message="")).model_dump()
         return "call:error", {"call_id": call_id, "error": error}
 
-    async def _emit_task_result(task: asyncio.Task, call_id: str) -> None:
+    def _cache_task_result(task: asyncio.Task, call_id: str) -> tuple[str, dict[str, Any]] | None:
+        """Synchronously store a finished task's frame into `pending_results`.
+        Returns the stored (event, frame), or None for a cancelled task.
+
+        Runs inside the task's done-callback — the same loop turn that pops
+        the call from `calls` — so `on_resume` can never observe a completing
+        call_id in *neither* map and misreport a live result as
+        ResultUnavailable."""
         nonlocal evictions
         if task.cancelled():
-            return
+            return None
         try:
             event, frame = task.result()
         except Exception as exc:
@@ -196,9 +223,9 @@ async def _emit_task_result(task: asyncio.Task, call_id: str) -> None:
                 "call_id": call_id,
                 "error": RemoteError(type=type(exc).__name__, message=str(exc)).model_dump(),
             }
-        # Store first, emit second. If the host is currently disconnected
-        # the emit is a no-op and the cached entry carries the result
-        # through to the next `resume`.
+        # Store first, emit second (the caller). If the host is currently
+        # disconnected the emit is a no-op and the cached entry carries the
+        # result through to the next `resume`.
         evicted = _store_pending_result(pending_results, call_id, (event, frame))
         if evicted is not None:
             evictions += 1
@@ -211,9 +238,23 @@ async def _emit_task_result(task: asyncio.Task, call_id: str) -> None:
                     _MAX_PENDING_RESULTS,
                     evictions,
                 )
+        return event, frame
+
+    async def _emit_frame(event: str, frame: dict[str, Any]) -> None:
         with contextlib.suppress(BaseException):
             await sio.emit(event, pack(frame), namespace=RPC_NAMESPACE)
 
+    def _schedule_emit(task: asyncio.Task, call_id: str) -> None:
+        """Done-callback: cache the result synchronously, then emit it from a
+        strongly-referenced task (see `emit_tasks`)."""
+        cached = _cache_task_result(task, call_id)
+        if cached is None:
+            return
+        event, frame = cached
+        emit_task = asyncio.create_task(_emit_frame(event, frame))
+        emit_tasks.add(emit_task)
+        emit_task.add_done_callback(emit_tasks.discard)
+
     def _track_call(call_id: str, task: asyncio.Task) -> None:
         calls[call_id] = task
         task.add_done_callback(lambda _t: calls.pop(call_id, None))
@@ -235,16 +276,24 @@ async def submit_http_call(payload: dict[str, Any], *, wait_s: float = 1.0) -> d
             return {"accepted": True, "call_id": call_id}
 
         task = _start_call(payload, call_id)
+        # Attach the cache+emit callback NOW, not in the timeout branch: the
+        # store must be enqueued in the same completion-callback batch as the
+        # `calls` pop, or a resume racing a completion at the wait budget's
+        # edge observes the id in neither map and misfires ResultUnavailable.
+        # The HTTP-answered path self-acks the cached copy below; the extra
+        # broadcast is one frame nobody is waiting on.
+        task.add_done_callback(lambda t, cid=call_id: _schedule_emit(t, cid))
 
         timeout_s = max(wait_s, 0.0)
         try:
             event, frame = await asyncio.wait_for(asyncio.shield(task), timeout=timeout_s)
         except TimeoutError:
-            task.add_done_callback(
-                lambda t, cid=call_id: asyncio.create_task(_emit_task_result(t, cid))
-            )
             return {"accepted": True, "call_id": call_id}
 
+        # Delivered synchronously in the HTTP response — the host will never
+        # ack this copy, so retire it now rather than pinning a cap slot.
+        pending_results.pop(call_id, None)
+
         if event == "call:result":
             return {"accepted": False, "ok": True, **frame}
         return {"accepted": False, "ok": False, **frame}
@@ -282,9 +331,7 @@ async def on_call(sid: str, data: Any) -> None:
             return
 
         task = _start_call(payload, call_id)
-        task.add_done_callback(
-            lambda t, cid=call_id: asyncio.create_task(_emit_task_result(t, cid))
-        )
+        task.add_done_callback(lambda t, cid=call_id: _schedule_emit(t, cid))
 
     async def on_cancel(sid: str, data: Any) -> None:
         payload = _u(data)
@@ -306,8 +353,15 @@ async def on_cancel(sid: str, data: Any) -> None:
         )
 
     async def on_resume(sid: str, data: Any) -> None:
-        """Replay cached results for the call_ids the host is still
-        waiting on. Called by the host right after (re)connect."""
+        """Resolve the call_ids the host is still waiting on. Called by the
+        host right after (re)connect. Each id reaches a definite terminal
+        state — never silence:
+
+        - a cached result → replayed;
+        - still running → left alone (its result arrives on completion);
+        - no record (evicted under cap, or unknown) → a `call:error` so the
+          host's `remote()` fails instead of hanging forever.
+        """
         payload = _u(data)
         ids = payload.get("call_ids")
         if not isinstance(ids, list):
@@ -316,10 +370,18 @@ async def on_resume(sid: str, data: Any) -> None:
             if not isinstance(cid, str):
                 continue
             cached = pending_results.get(cid)
-            if cached is None:
+            if cached is not None:
+                event, frame = cached
+                await sio.emit(event, pack(frame), to=sid, namespace=RPC_NAMESPACE)
+                continue
+            if cid in calls:
                 continue
-            event, frame = cached
-            await sio.emit(event, pack(frame), to=sid, namespace=RPC_NAMESPACE)
+            await sio.emit(
+                "call:error",
+                pack(_unavailable_error(cid)),
+                to=sid,
+                namespace=RPC_NAMESPACE,
+            )
 
     async def on_ack(sid: str, data: Any) -> None:
         """Host confirms it has consumed the result. Free the slot."""
diff --git a/agentix/runtime/server/worker/process.py b/agentix/runtime/server/worker/process.py
index b08e3b0..8f75b44 100644
--- a/agentix/runtime/server/worker/process.py
+++ b/agentix/runtime/server/worker/process.py
@@ -128,7 +128,11 @@ async def run(self) -> None:
                 task.cancel()
             if pending:
                 await asyncio.gather(*pending, return_exceptions=True)
-        await self._outbound_q.join()
+        # Bound the drain: a wedged outbound pipe (writer.drain() blocked on a
+        # full OS pipe) would hang join() forever — task_done() never fires for
+        # the stuck frame. Mirror the server-side bounded join.
+        with contextlib.suppress(TimeoutError):
+            await asyncio.wait_for(self._outbound_q.join(), timeout=2.0)
         if self._drainer is not None:
             self._drainer.cancel()
 
@@ -276,18 +280,20 @@ def _cancel(self, call_id: str) -> None:
         task = self._calls.get(call_id)
         if task is not None:
             task.cancel()
-            asyncio.create_task(
-                self._send(
-                    {
-                        "type": "error",
-                        "call_id": call_id,
-                        "error": RemoteError(
-                            type="Cancelled",
-                            message="remote call cancelled",
-                            cancelled=True,
-                        ).model_dump(),
-                    }
-                )
+            # Enqueue synchronously (the outbound queue is unbounded) instead of
+            # spawning an untracked `create_task`, which the loop only weakly
+            # references and could GC before it runs — dropping the Cancelled
+            # frame.
+            self._enqueue_frame(
+                {
+                    "type": "error",
+                    "call_id": call_id,
+                    "error": RemoteError(
+                        type="Cancelled",
+                        message="remote call cancelled",
+                        cancelled=True,
+                    ).model_dump(),
+                }
             )
 
 
diff --git a/docs/concepts/remote-calls.mdx b/docs/concepts/remote-calls.mdx
index 8511788..26666c5 100644
--- a/docs/concepts/remote-calls.mdx
+++ b/docs/concepts/remote-calls.mdx
@@ -133,8 +133,9 @@ async with provider.session(SandboxConfig(image=..., bundle=...)) as sandbox:
 ## Errors and Cancellation
 
 Agentix keeps errors in-band. A failed call emits `call:error` with a
-structured payload; the client raises `RemoteCallError` (or
-`asyncio.CancelledError` when `error.cancelled=True`). `RemoteCallError`'s
+structured payload; the client raises `RemoteCallError` (or its subclass
+`CallCancelled` when `error.cancelled=True` — a terminal server-side state,
+distinct from local task cancellation). `RemoteCallError`'s
 message includes the sandbox-side traceback when one is available, and
 `exc.error.traceback` holds it on the exception for programmatic access.
 
diff --git a/tests/provider/test_session_contract.py b/tests/provider/test_session_contract.py
new file mode 100644
index 0000000..70ca7ec
--- /dev/null
+++ b/tests/provider/test_session_contract.py
@@ -0,0 +1,38 @@
+"""`SandboxProvider.session()` teardown contract: the sandbox is deleted even
+when closing its runtime client fails — otherwise the container (and its
+reserved port) leaks on every aclose error."""
+
+from __future__ import annotations
+
+import pytest
+
+from agentix.provider.base import Sandbox, SandboxConfig, SandboxId, SandboxInfo, SandboxProvider
+
+
+class _ExplodingCloseSandbox(Sandbox):
+    async def aclose(self) -> None:  # type: ignore[override]
+        raise RuntimeError("close boom")
+
+
+class _Provider(SandboxProvider):
+    def __init__(self) -> None:
+        self.deleted: list[SandboxId] = []
+
+    async def create(self, config: SandboxConfig) -> Sandbox:
+        return _ExplodingCloseSandbox(
+            sandbox_id=SandboxId("s-1"), runtime_url="http://127.0.0.1:1", status="running"
+        )
+
+    async def get(self, sandbox_id: SandboxId) -> SandboxInfo:
+        return SandboxInfo(sandbox_id=sandbox_id, runtime_url="http://127.0.0.1:1", status="running")
+
+    async def delete(self, sandbox_id: SandboxId) -> None:
+        self.deleted.append(sandbox_id)
+
+
+async def test_session_deletes_sandbox_even_when_aclose_fails() -> None:
+    provider = _Provider()
+    with pytest.raises(RuntimeError, match="close boom"):
+        async with provider.session(SandboxConfig(image="x", bundle="y")):
+            pass
+    assert provider.deleted == [SandboxId("s-1")]
diff --git a/tests/runtime/test_protocol.py b/tests/runtime/test_protocol.py
index c099f9e..63b86a7 100644
--- a/tests/runtime/test_protocol.py
+++ b/tests/runtime/test_protocol.py
@@ -14,7 +14,7 @@
 import pytest
 import socketio
 
-from agentix import RemoteCallError, RuntimeClient
+from agentix import Failed, Ok, RemoteCallError, RuntimeClient
 from agentix.runtime.shared.codec import pack, unpack
 from agentix.runtime.shared.models import RemoteRequest
 from tests import _worker_target as target
@@ -361,3 +361,168 @@ async def _on_error(data):
     assert payload["call_id"] == "cancel-me"
     assert payload["error"]["type"] == "Cancelled"
     assert payload["error"]["cancelled"] is True
+
+
+async def test_try_remote_returns_ok(use_inprocess_worker, live_server):
+    use_inprocess_worker()
+    base_url = await live_server()
+    async with RuntimeClient(base_url) as c:
+        result = await c.try_remote(target.add, 2, 3)
+    assert isinstance(result, Ok)
+    assert result.value == 5
+
+
+async def test_try_remote_returns_failed(use_inprocess_worker, live_server):
+    use_inprocess_worker()
+    base_url = await live_server()
+    async with RuntimeClient(base_url) as c:
+        result = await c.try_remote(target.boom)
+    assert isinstance(result, Failed)
+    assert isinstance(result.error, RemoteCallError)
+
+
+async def test_resume_for_unknown_call_id_fails_definitively(use_inprocess_worker, live_server):
+    """A `resume` for a call_id the runtime no longer holds (evicted
+    under cap, or never seen) must return a definite `call:error` — the
+    contract forbids silence, which would hang the host's `remote()`.
+    """
+    use_inprocess_worker()
+    base_url = await live_server()
+
+    sio = socketio.AsyncClient()
+    errors: asyncio.Queue = asyncio.Queue()
+
+    async def _on_error(data):
+        await errors.put(unpack(data))
+
+    sio.on("call:error", _on_error, namespace=RPC_NAMESPACE)
+    await sio.connect(base_url, namespaces=[RPC_NAMESPACE])
+    try:
+        await sio.emit(
+            "resume",
+            pack({"call_ids": ["never-existed"]}),
+            namespace=RPC_NAMESPACE,
+        )
+        payload = await asyncio.wait_for(errors.get(), timeout=5)
+    finally:
+        await sio.disconnect()
+
+    assert payload["call_id"] == "never-existed"
+    assert payload["error"]["type"] == "ResultUnavailable"
+
+
+async def test_resume_for_running_call_is_left_alone(use_inprocess_worker, live_server):
+    """A `resume` naming a call_id that is still RUNNING must not answer
+    ResultUnavailable — the call's real result arrives on completion. Guards
+    the `cid in calls` branch and the store-before-pop transition (a
+    completing call must never be observable in neither map)."""
+    use_inprocess_worker()
+    base_url = await live_server()
+
+    sio = socketio.AsyncClient()
+    events: asyncio.Queue = asyncio.Queue()
+
+    async def _on_result(data):
+        await events.put(("call:result", unpack(data)))
+
+    async def _on_error(data):
+        await events.put(("call:error", unpack(data)))
+
+    sio.on("call:result", _on_result, namespace=RPC_NAMESPACE)
+    sio.on("call:error", _on_error, namespace=RPC_NAMESPACE)
+    await sio.connect(base_url, namespaces=[RPC_NAMESPACE])
+    try:
+        req = request_for(target.count_exec_and_sleep, args=[0.4], call_id="resume-race-1")
+        await sio.emit("call", pack(req.model_dump()), namespace=RPC_NAMESPACE)
+        await asyncio.sleep(0.1)  # the call is now running server-side
+        await sio.emit("resume", pack({"call_ids": ["resume-race-1"]}), namespace=RPC_NAMESPACE)
+        kind, payload = await asyncio.wait_for(events.get(), timeout=5)
+    finally:
+        await sio.disconnect()
+
+    assert payload["call_id"] == "resume-race-1"
+    assert kind == "call:result"  # never a false ResultUnavailable
+
+
+class _StaleSio:
+    """A disconnected-but-reconnecting socketio handle: `disconnect()` is a
+    library no-op in this state; only `shutdown()` aborts the reconnect loop."""
+
+    connected = False
+
+    def __init__(self) -> None:
+        self.shutdown_calls = 0
+
+    async def shutdown(self) -> None:
+        self.shutdown_calls += 1
+
+
+async def test_stale_sio_handle_is_shut_down_before_replacement(use_inprocess_worker, live_server):
+    """Replacing a stale handle must abort its background reconnect loop —
+    `disconnect()` is a no-op on a dropped transport, leaving the abandoned
+    client to reconnect on its own (a second live /rpc socket + a leaked
+    aiohttp session per transport flap)."""
+    use_inprocess_worker()
+    base_url = await live_server()
+    async with RuntimeClient(base_url) as c:
+        stale = _StaleSio()
+        c._sio = stale
+        assert await c.remote(target.add, 1, 2) == 3  # rebuilds via _ensure_sio
+        assert stale.shutdown_calls == 1
+
+
+async def test_close_shuts_down_mid_reconnect_handle(use_inprocess_worker, live_server):
+    use_inprocess_worker()
+    base_url = await live_server()
+    c = RuntimeClient(base_url)
+    stale = _StaleSio()
+    c._sio = stale
+    await c.close()
+    assert stale.shutdown_calls == 1
+
+
+async def test_http_fast_path_falls_back_on_transport_error():
+    """The fast path is opportunistic: a transport failure (dead sandbox,
+    gateway 5xx) must fall back to the SIO channel — never leak a raw httpx
+    error out of `remote()` / `try_remote()`'s `Result` contract."""
+    import types as _types
+
+    c = RuntimeClient("http://127.0.0.1:1")  # nothing listens here
+    try:
+        fake_sio = _types.SimpleNamespace(connected=True, sid="s1")
+        kind, _ = await c._try_http_fast_path(
+            sio=fake_sio, payload={"call_id": "x", "callable": "os::getcwd", "arguments": b""}
+        )
+        assert kind == "fallback"
+    finally:
+        await c.close()
+
+
+async def test_cancelled_error_maps_to_call_cancelled():
+    """PROTOCOL.md client mapping: `cancelled=True` → CallCancelled, a
+    RemoteCallError subclass (terminal server-side state) — never a bare
+    asyncio.CancelledError that reads as local task cancellation."""
+    from agentix import CallCancelled
+    from agentix.runtime.client.client import _raise_remote_error
+    from agentix.runtime.shared.models import RemoteError
+
+    err = RemoteError(type="Cancelled", message="remote call cancelled", cancelled=True)
+    with pytest.raises(CallCancelled):
+        _raise_remote_error("fn", err)
+    assert issubclass(CallCancelled, RemoteCallError)
+
+
+async def test_malformed_call_error_is_typed_not_keyerror(use_inprocess_worker, live_server):
+    """A call:error frame with no `error` payload must resolve to a typed
+    MalformedError — not a bare KeyError escaping remote()/try_remote()."""
+    use_inprocess_worker()
+    base_url = await live_server()
+    async with RuntimeClient(base_url) as c:
+        task = asyncio.create_task(c.remote(target.count_exec_and_sleep, 30.0))
+        while not c._pending:
+            await asyncio.sleep(0.01)
+        ((cid, q),) = c._pending.items()
+        await q.put(("error", {"call_id": cid}))  # no "error" field
+        with pytest.raises(RemoteCallError) as ei:
+            await task
+        assert ei.value.error.type == "MalformedError"
diff --git a/tests/test_public_exports.py b/tests/test_public_exports.py
index 22c85df..5c2cb1e 100644
--- a/tests/test_public_exports.py
+++ b/tests/test_public_exports.py
@@ -10,23 +10,34 @@
 
 def test_failure_vocabulary_importable_from_agentix() -> None:
     from agentix import (
+        CallCancelled,
         CallTimeout,
+        Failed,
+        Ok,
         RemoteCallError,
+        Result,
         RuntimeUnreachable,
         WorkerExited,
         configure_logging,
     )
 
-    # WorkerExited is a RemoteCallError subclass, so `except RemoteCallError`
-    # still catches a dead-worker call while `except WorkerExited` branches on it.
+    # WorkerExited / CallCancelled are RemoteCallError subclasses, so
+    # `except RemoteCallError` still catches them while the narrower names
+    # allow branching.
     assert issubclass(WorkerExited, RemoteCallError)
+    assert issubclass(CallCancelled, RemoteCallError)
+    assert Ok(1).value == 1 and Failed(ValueError()).error is not None
+    assert Result[int]  # generic union alias is subscriptable
     assert callable(configure_logging)
     for name in (CallTimeout, RuntimeUnreachable):
         assert issubclass(name, Exception)
 
 
 def test_failure_vocabulary_in_dunder_all() -> None:
-    for name in ("CallTimeout", "RuntimeUnreachable", "WorkerExited", "configure_logging"):
+    for name in (
+        "CallCancelled", "CallTimeout", "Failed", "Ok", "Result",
+        "RuntimeUnreachable", "WorkerExited", "configure_logging",
+    ):
         assert name in agentix.__all__