runtime: core reliability — definitive resume, teardown + task-retention fixes (PR #122 stage D)

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",

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

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 |

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",

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",

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"]