Carry rich argument and result types and refuse ambiguous signatures

miles/utils/workers/rpc/common/metadata.py

@@ -44,7 +44,10 @@ def collect_rpc_method_specs(worker_cls: type) -> dict[str, RpcMethodSpec]:
             continue
         if not callable(static_attr):
             continue
-        specs[name] = _build_method_spec(worker_cls=worker_cls, name=name, fn=inspect.unwrap(static_attr))
+        specs[name] = _build_method_spec(worker_cls=worker_cls, name=name, attr=static_attr)
+
+    if len(specs) == 0:
+        raise TypeError(f"{worker_cls.__name__} exposes no public rpc methods")
 
     return specs
 
@@ -54,9 +57,26 @@ class _RpcConfig:
     concurrency_group: str
 
 
-def _build_method_spec(*, worker_cls: type, name: str, fn: Callable[..., Any]) -> RpcMethodSpec:
-    config: _RpcConfig = getattr(fn, _RPC_CONFIG_ATTR, _RpcConfig(concurrency_group=DEFAULT_CONCURRENCY_GROUP))
-    is_async = inspect.iscoroutinefunction(inspect.unwrap(fn))
+def _find_rpc_config(attr: Callable[..., Any]) -> _RpcConfig:
+    layer: Any = attr
+    while layer is not None:
+        config = getattr(layer, _RPC_CONFIG_ATTR, None)
+        if config is not None:
+            return config
+        layer = getattr(layer, "__wrapped__", None)
+    return _RpcConfig(concurrency_group=DEFAULT_CONCURRENCY_GROUP)
+
+
+def _build_method_spec(*, worker_cls: type, name: str, attr: Callable[..., Any]) -> RpcMethodSpec:
+    fn = inspect.unwrap(attr)
+    if not inspect.isroutine(fn):
+        raise TypeError(
+            f"{worker_cls.__name__}.{name} is a public callable attribute but not a method, "
+            f"so it cannot be exposed over rpc; make it private or move it off the worker class"
+        )
+
+    config = _find_rpc_config(attr)
+    is_async = inspect.iscoroutinefunction(fn)
     if is_async and config.concurrency_group != DEFAULT_CONCURRENCY_GROUP:
         raise TypeError(
             f"{worker_cls.__name__}.{name} is async; concurrency groups only serialize sync methods, "
@@ -66,11 +86,39 @@ def _build_method_spec(*, worker_cls: type, name: str, fn: Callable[..., Any]) -
     signature = inspect.signature(fn)
     hints = typing.get_type_hints(fn, include_extras=True)
 
+    parameters = list(signature.parameters.values())
+    if len(parameters) == 0:
+        raise TypeError(f"{worker_cls.__name__}.{name} must take a receiver parameter for rpc exposure")
+    if parameters[0].name != "self":
+        raise TypeError(
+            f"{worker_cls.__name__}.{name} must name its receiver parameter 'self' for rpc exposure, "
+            f"got {parameters[0].name!r}; otherwise it would be silently dropped from the wire"
+        )
+    if parameters[0].kind not in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD):
+        raise TypeError(
+            f"{worker_cls.__name__}.{name} must take its receiver parameter positionally for rpc exposure, "
+            f"got a {parameters[0].kind.description} parameter"
+        )
+
     query_fields: dict[str, Any] = {}
-    for param in list(signature.parameters.values())[1:]:
+    for param in parameters[1:]:
+        if param.kind in (
+            inspect.Parameter.VAR_POSITIONAL,
+            inspect.Parameter.VAR_KEYWORD,
+            inspect.Parameter.POSITIONAL_ONLY,
+        ):
+            raise TypeError(
+                f"{worker_cls.__name__}.{name} must not use *args/**kwargs or positional-only parameters "
+                f"for rpc exposure"
+            )
+        if param.annotation is inspect.Parameter.empty:
+            raise TypeError(f"{worker_cls.__name__}.{name} parameter '{param.name}' must be type-annotated")
         default = ... if param.default is inspect.Parameter.empty else param.default
         query_fields[param.name] = (hints[param.name], default)
 
+    if signature.return_annotation is inspect.Signature.empty:
+        raise TypeError(f"{worker_cls.__name__}.{name} must have a return type annotation")
+
     return RpcMethodSpec(
         name=name,
         concurrency_group=config.concurrency_group,

miles/utils/workers/rpc/common/serialization.py

@@ -3,8 +3,11 @@
 import dataclasses
 from typing import Any
 
+import typing_extensions
 from pydantic import BaseModel, ConfigDict, TypeAdapter, create_model
 
+_WIRE_CONFIG = ConfigDict(ser_json_bytes="base64", val_json_bytes="base64")
+
 
 @dataclasses.dataclass(frozen=True)
 class RpcSerializer:
@@ -13,8 +16,11 @@ class RpcSerializer:
 
     @classmethod
     def create(cls, *, query_model_name: str, query_fields: dict[str, Any], result_annotation: Any) -> RpcSerializer:
-        query_model = create_model(query_model_name, __config__=ConfigDict(extra="forbid"), **query_fields)
-        return cls(query_model=query_model, result_adapter=TypeAdapter(result_annotation))
+        query_model = create_model(
+            query_model_name, __config__=ConfigDict(extra="forbid", **_WIRE_CONFIG), **query_fields
+        )
+        result_config = None if _carries_own_config(result_annotation) else _WIRE_CONFIG
+        return cls(query_model=query_model, result_adapter=TypeAdapter(result_annotation, config=result_config))
 
     def encode_query(self, kwargs: dict[str, Any]) -> dict[str, Any]:
         return self.query_model(**kwargs).model_dump(mode="json")
@@ -27,3 +33,13 @@ def encode_result(self, result: Any) -> Any:
 
     def decode_result(self, payload: Any) -> Any:
         return self.result_adapter.validate_python(payload)
+
+
+def _carries_own_config(annotation: Any) -> bool:
+    if not isinstance(annotation, type):
+        return False
+    return (
+        issubclass(annotation, BaseModel)
+        or dataclasses.is_dataclass(annotation)
+        or typing_extensions.is_typeddict(annotation)
+    )

tests/fast/utils/workers/e2e/test_client_local_validation.py

@@ -0,0 +1,97 @@
+import uuid
+
+import httpx
+import pytest
+from pydantic import ValidationError
+from tests.fast.utils.workers.e2e.e2e_worker import E2eWorker
+
+from miles.utils.workers.rpc.client.handle import RpcWorkerHandle
+
+
+class TestNoRequestIsSent:
+    async def test_unknown_method_is_an_attribute_error(self, dead_proxy, make_handle):
+        """A method the worker does not define fails before any request goes out."""
+        handle = make_handle(dead_proxy)
+        with pytest.raises(AttributeError, match="no rpc method"):
+            _ = handle.no_such_method
+        assert dead_proxy.requests == []
+
+    async def test_missing_required_argument(self, dead_proxy, make_handle):
+        """A missing argument is caught locally, not by the server."""
+        handle = make_handle(dead_proxy)
+        with pytest.raises(ValidationError):
+            await handle.demo_sync(a=1)
+        assert dead_proxy.requests == []
+
+    async def test_unknown_argument(self, dead_proxy, make_handle):
+        """An argument the method does not declare is caught locally."""
+        handle = make_handle(dead_proxy)
+        with pytest.raises(ValidationError):
+            await handle.demo_sync(a=1, b=2, c=3)
+        assert dead_proxy.requests == []
+
+    async def test_wrong_argument_type(self, dead_proxy, make_handle):
+        """An uncoercible argument is caught locally."""
+        handle = make_handle(dead_proxy)
+        with pytest.raises(ValidationError):
+            await handle.demo_sync(a="not-a-number", b=2)
+        assert dead_proxy.requests == []
+
+    async def test_positional_arguments_are_rejected(self, dead_proxy, make_handle):
+        """Calls are keyword-only, so positional arguments fail locally."""
+        handle = make_handle(dead_proxy)
+        with pytest.raises(TypeError):
+            await handle.demo_sync(1, 2)
+        assert dead_proxy.requests == []
+
+
+class TestHandleConstruction:
+    def test_reserved_method_name_is_rejected(self):
+        """A worker whose method shadows a handle attribute is refused at construction."""
+
+        class Shadowing:
+            async def wait_ready(self, timeout: float) -> None:
+                pass
+
+        with pytest.raises(TypeError, match="shadow"):
+            RpcWorkerHandle(Shadowing, server_url="http://127.0.0.1:9")
+
+    def test_worker_without_public_methods_is_rejected(self):
+        """A worker with nothing to expose is refused."""
+
+        class Empty:
+            def _demo_hidden(self) -> int:
+                return 1
+
+        with pytest.raises(TypeError):
+            RpcWorkerHandle(Empty, server_url="http://127.0.0.1:9")
+
+    def test_worker_with_unannotated_method_is_rejected(self):
+        """A method missing annotations is refused on the client too, matching the server."""
+
+        class Unannotated:
+            def demo_unannotated(self, x):
+                return x
+
+        with pytest.raises(TypeError):
+            RpcWorkerHandle(Unannotated, server_url="http://127.0.0.1:9")
+
+    async def test_trailing_slash_in_server_url(self, server, make_handle):
+        """A server url with a trailing slash still produces valid request paths."""
+        handle = make_handle(f"{server.url}/")
+        assert await handle.demo_sync(a=1, b=1) == 2
+
+    async def test_client_and_server_agree_on_the_method_set(self, server):
+        """Every method the client exposes is routable on the server, and private ones are not."""
+        handle = RpcWorkerHandle(E2eWorker, server_url=server.url)
+        assert "demo_sync" in handle._specs and "_bump" not in handle._specs
+
+        async with httpx.AsyncClient(base_url=server.url, timeout=30.0, trust_env=False) as client:
+            for name in handle._specs:
+                response = await client.post(
+                    f"/v1/{name}", json={"call_id": uuid.uuid4().hex, "query": {"__unknown__": 1}}
+                )
+                assert response.status_code == 400, name
+
+            private = await client.post("/v1/_bump", json={"call_id": uuid.uuid4().hex, "query": {}})
+            assert private.status_code == 404