Merge pull request #47 from Agentiix/codex/rpc-namespace

Use /rpc namespace for remote calls

Author: FatPigeorz

agentix/runtime/PROTOCOL.md

@@ -35,15 +35,19 @@ await client.remote(run, seed=42)
 | Path | Carries | Wire |
 | --- | --- | --- |
 | `GET /health` | health probe | HTTP JSON |
-| Socket.IO `/` | `c.remote()` RPC | msgpack-wrapped `call` / `call:result` / `call:error` / `cancel` |
+| `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 `/trace`, `/log`, `/<plugin>` | side channels | plugin-defined events (msgpack payloads) |
 | worker private pipe | runtime ↔ worker | length-prefixed msgpack frames |
 
-HTTP is only for `/health`. Socket.IO is the host-to-runtime edge.
-The worker pipe is the runtime-to-worker edge inside the sandbox.
-The current implementation uses one worker subprocess per runtime.
+HTTP covers health plus the internal `/call` fast path for short
+remote calls. Socket.IO `/rpc` remains the RPC event channel when a
+call is submitted over SIO or an accepted HTTP call completes
+asynchronously. The worker pipe is the runtime-to-worker edge inside
+the sandbox. The current implementation uses one worker subprocess per
+runtime.
 
-## Socket.IO Events (RPC on `/`)
+## Socket.IO Events (RPC on `/rpc`)
 
 ```text
 call          {call_id, callable, arguments}

agentix/runtime/client/client.py

@@ -15,7 +15,7 @@
         await c.remote(abridge.start_service, ...)
 
 Core auto-registers `/trace` and `/log` namespaces so trace + log
-records flow from the sandbox without setup. `/` carries RPC.
+records flow from the sandbox without setup. `/rpc` carries RPC.
 """
 
 from __future__ import annotations
@@ -40,6 +40,7 @@
 
 P = ParamSpec("P")
 R = TypeVar("R")
+RPC_NAMESPACE = "/rpc"
 
 
 class RemoteCallError(RuntimeError):
@@ -198,7 +199,7 @@ async def remote(
             # and completes via the normal SIO result channel.
             kind, value = await self._try_http_fast_path(sio=sio, payload=payload)
             if kind == "fallback":
-                await sio.emit("call", pack(payload))
+                await sio.emit("call", pack(payload), namespace=RPC_NAMESPACE)
             elif kind == "result":
                 terminated = True
                 return cast(R, value)
@@ -218,7 +219,11 @@ async def remote(
             self._pending.pop(call_id, None)
             if not terminated:
                 with contextlib.suppress(BaseException):
-                    await sio.emit("cancel", pack({"call_id": call_id}))
+                    await sio.emit(
+                        "cancel",
+                        pack({"call_id": call_id}),
+                        namespace=RPC_NAMESPACE,
+                    )
 
     # ── Socket.IO connection management ─────────────────────────
 
@@ -242,8 +247,8 @@ async def _on_call_result(data):
             async def _on_call_error(data):
                 await self._route_event("error", data)
 
-            sio.on("call:result", _on_call_result)
-            sio.on("call:error", _on_call_error)
+            sio.on("call:result", _on_call_result, namespace=RPC_NAMESPACE)
+            sio.on("call:error", _on_call_error, namespace=RPC_NAMESPACE)
 
             async def _on_connect(*_args):
                 # Fires on initial connect and on every reconnect. Tell
@@ -253,18 +258,22 @@ async def _on_connect(*_args):
                 if not pending_ids:
                     return
                 with contextlib.suppress(BaseException):
-                    await sio.emit("resume", pack({"call_ids": pending_ids}))
+                    await sio.emit(
+                        "resume",
+                        pack({"call_ids": pending_ids}),
+                        namespace=RPC_NAMESPACE,
+                    )
 
             async def _on_disconnect(*_args):
                 # Tasks survive on the server side; results are buffered
                 # until ack. We rely on socketio's auto-reconnect to come
                 # back, then `_on_connect` will re-emit `resume`.
                 logger.debug("sio disconnect; will resume after reconnect")
 
-            sio.on("connect", _on_connect)
-            sio.on("disconnect", _on_disconnect)
+            sio.on("connect", _on_connect, namespace=RPC_NAMESPACE)
+            sio.on("disconnect", _on_disconnect, namespace=RPC_NAMESPACE)
 
-            namespaces = ["/"]
+            namespaces = [RPC_NAMESPACE]
             for ns in self._namespaces:
                 sio.register_namespace(ns)
                 if ns.namespace not in namespaces:
@@ -297,7 +306,7 @@ async def _ack(self, call_id: str) -> None:
         if sio is None or not sio.connected:
             return
         with contextlib.suppress(BaseException):
-            await sio.emit("ack", pack({"call_id": call_id}))
+            await sio.emit("ack", pack({"call_id": call_id}), namespace=RPC_NAMESPACE)
 
 
 __all__ = ["RemoteCallError", "RuntimeClient"]

agentix/runtime/server/app.py

@@ -9,7 +9,7 @@
   msgpack request/response. Returns the result inline if it lands
   within the caller's `prefer_sync_ms` budget; otherwise returns
   `accepted` and the result follows on Socket.IO.
-- Socket.IO at `/socket.io/` — unary RPC on `/` (`call` / `call:result` /
+- Socket.IO at `/socket.io/` — unary RPC on `/rpc` (`call` / `call:result` /
   `call:error`, `cancel`, plus `resume`/`ack` for reconnect-safe
   delivery), and side-channel namespaces (`/trace`, `/log`, and
   plugin paths registered via `agentix.sio`).

agentix/runtime/server/sio.py

@@ -2,7 +2,7 @@
 
 Two responsibilities:
 
-1. The RPC protocol on the default `/` namespace — `call` / `cancel`
+1. The RPC protocol on the `/rpc` namespace — `call` / `cancel`
    / `call:result` / `call:error`.
 
 2. Dynamic namespace forwarding. When a worker-side `agentix.Namespace`
@@ -11,7 +11,7 @@
    the worker. Outbound `sio_emit` frames become real SIO emits on the
    corresponding namespace.
 
-Reserved namespace paths (claimed by agentix-core): `/`, `/trace`,
+Reserved namespace paths (claimed by agentix-core): `/rpc`, `/trace`,
 `/log`. Plugins use their own paths (typically `/<package-name>`).
 """
 
@@ -33,6 +33,7 @@
 from agentix.runtime.shared.models import RemoteError, RemoteRequest
 
 logger = logging.getLogger("agentix.runtime.sio")
+RPC_NAMESPACE = "/rpc"
 
 
 def _u(data: Any) -> dict:
@@ -149,7 +150,7 @@ async def _emit_task_result(task: asyncio.Task, call_id: str) -> None:
             # disconnected the emit is a no-op and the cached entry
             # carries the result through to the next `resume`.
             pending_results[call_id] = (event, frame)
-            await sio.emit(event, pack(frame))
+            await sio.emit(event, pack(frame), namespace=RPC_NAMESPACE)
 
     def _track_call(call_id: str, task: asyncio.Task) -> None:
         calls[call_id] = task
@@ -189,18 +190,16 @@ async def submit_http_call(payload: dict[str, Any], *, prefer_sync_ms: int = 100
     # Runtime internal hook used by the HTTP fast-path endpoint.
     setattr(sio, "submit_http_call", submit_http_call)
 
-    @sio.event
-    async def connect(sid: str, environ: dict, auth: Any = None) -> None:
+    async def on_connect(sid: str, environ: dict, auth: Any = None) -> None:
         logger.debug("sio connect %s", sid)
 
-    @sio.event
-    async def disconnect(sid: str) -> None:
+    async def on_disconnect(sid: str) -> None:
         # Tasks intentionally outlive the connection. Their results
         # land in `pending_results` and will be replayed on the next
         # `resume`. The host may also cancel explicitly via `cancel`.
         logger.debug("sio disconnect %s", sid)
 
-    # ── RPC on `/` ───────────────────────────────────────────────
+    # ── RPC on `/rpc` ────────────────────────────────────────────
 
     async def on_call(sid: str, data: Any) -> None:
         payload = _u(data)
@@ -211,6 +210,7 @@ async def on_call(sid: str, data: Any) -> None:
                 event,
                 pack(frame),
                 to=sid,
+                namespace=RPC_NAMESPACE,
             )
             return
 
@@ -240,6 +240,7 @@ async def on_cancel(sid: str, data: Any) -> None:
             "call:error",
             pack(_cancelled_error(call_id)),
             to=sid,
+            namespace=RPC_NAMESPACE,
         )
 
     async def on_resume(sid: str, data: Any) -> None:
@@ -256,7 +257,7 @@ async def on_resume(sid: str, data: Any) -> None:
             if cached is None:
                 continue
             event, frame = cached
-            await sio.emit(event, pack(frame), to=sid)
+            await sio.emit(event, pack(frame), to=sid, namespace=RPC_NAMESPACE)
 
     async def on_ack(sid: str, data: Any) -> None:
         """Host confirms it has consumed the result. Free the slot."""
@@ -283,7 +284,7 @@ async def _on_worker_sio_frame(frame: dict[str, Any]) -> None:
                 return
             await sio.emit(event, pack(frame.get("data")), namespace=namespace)
         elif kind == "sio_open":
-            if namespace in opened_namespaces or namespace == "/":
+            if namespace in opened_namespaces or namespace in {"/", RPC_NAMESPACE}:
                 return
             opened_namespaces.add(namespace)
             _register_namespace(namespace)
@@ -306,13 +307,15 @@ async def trigger_event(self, event: str, *args: Any) -> Any:
 
     worker.set_sio_handler(_on_worker_sio_frame)
 
-    # Register RPC handlers on `/` non-decorator-style — `@sio.on(name)`
+    # Register RPC handlers on `/rpc` non-decorator-style — `@sio.on(name)`
     # decorates by side effect and pyright can't tell that the wrapped
     # function is still usable.
-    sio.on("call", on_call)
-    sio.on("cancel", on_cancel)
-    sio.on("resume", on_resume)
-    sio.on("ack", on_ack)
+    sio.on("connect", on_connect, namespace=RPC_NAMESPACE)
+    sio.on("disconnect", on_disconnect, namespace=RPC_NAMESPACE)
+    sio.on("call", on_call, namespace=RPC_NAMESPACE)
+    sio.on("cancel", on_cancel, namespace=RPC_NAMESPACE)
+    sio.on("resume", on_resume, namespace=RPC_NAMESPACE)
+    sio.on("ack", on_ack, namespace=RPC_NAMESPACE)
 
     # Pre-register core namespaces so the host can connect to them
     # immediately — the worker subscribes lazily, but the SIO server

agentix/sio.py

@@ -7,7 +7,7 @@
 
 Three reserved namespace paths are owned by agentix-core:
 
-  - `/`       — RPC (call / cancel / call:result / call:error)
+  - `/rpc`    — RPC (call / cancel / call:result / call:error)
   - `/trace`  — Trace/Span lifecycle
   - `/log`    — stdlib `logging` records
 
@@ -48,7 +48,7 @@ async def fetch_remote(self, payload):
 Handler = Callable[[Any], Any]
 
 
-RESERVED_NAMESPACES = frozenset({"/", "/trace", "/log"})
+RESERVED_NAMESPACES = frozenset({"/rpc", "/trace", "/log"})
 
 
 class RemoteSioError(RuntimeError):
@@ -125,7 +125,7 @@ class Namespace:
     be registered explicitly via `self.on("fetch:result", handler)`.
     """
 
-    namespace: str = "/"  # subclass MUST override
+    namespace: str = ""  # subclass MUST override
 
     def __init__(self, namespace: str | None = None) -> None:
         if namespace is not None:

docs/concepts/remote-calls.mdx

@@ -57,8 +57,9 @@ becomes this wire payload:
 runtime uses it to correlate `call:result` / `call:error` responses and
 to support cancellation.
 
-Remote calls use Socket.IO events on the `/` namespace. HTTP is only
-used for `/health`.
+Remote calls use Socket.IO events on the `/rpc` namespace. `RuntimeClient`
+may use the internal HTTP `/call` fast path for short calls; accepted
+long-running calls still complete over `/rpc`.
 
 ## Example
 
@@ -83,7 +84,7 @@ ride their own namespaces, separate from `c.remote()`:
 
 | Namespace | Direction | Purpose |
 | --- | --- | --- |
-| `/` | host ↔ sandbox | `c.remote()` |
+| `/rpc` | host ↔ sandbox | `c.remote()` |
 | `/trace` | sandbox → host | span lifecycle (auto-registered) |
 | `/log` | sandbox → host | stdlib logging records (auto-registered) |
 | `/<plugin>` | both | plugin-defined events via `agentix.sio` |

docs/reference/architecture.mdx

@@ -62,7 +62,7 @@ blobs inside `call:result`.
 1. Host imports `app.run`.
 2. Host calls `client.remote(run, input="hello")`.
 3. Client builds `RemoteCallable("app::run")` and pickles `((), {"input": "hello"})`.
-4. Client emits `call` on Socket.IO `/`.
+4. Client emits `call` on Socket.IO `/rpc`.
 5. Runtime server forwards the request to the worker subprocess.
 6. Worker imports `app`, resolves `run`, unpickles args, and calls it.
 7. Worker pickles the return value.
@@ -74,7 +74,8 @@ blobs inside `call:result`.
 | Path | Carries | Wire |
 | --- | --- | --- |
 | `GET /health` | health probe | HTTP JSON |
-| Socket.IO `/` | `c.remote()` RPC | `call` / `call:result` / `call:error` / `cancel` |
+| `POST /call` | internal short-call fast path | HTTP msgpack |
+| Socket.IO `/rpc` | `c.remote()` RPC | `call` / `call:result` / `call:error` / `cancel` |
 | Socket.IO `/trace`, `/log`, `/<plugin>` | side channels | plugin-defined events (msgpack payloads) |
 | worker private pipe | runtime ↔ worker | length-prefixed msgpack frames |