feat(rpc): expose TransportKind and on_serve_start lifecycle hook

Workers (RPC implementations) had no way to learn which transport they
were bound to.  CallContext.transport_metadata only carried HTTP fields
and was ambiguous for non-HTTP transports.

Add a coarse TransportKind enum (PIPE / HTTP / UNIX) plus a duck-typed
on_serve_start(self, kind) lifecycle hook fired once per process before
the first request.  HTTP fires lazily via a Falcon middleware so
pre-fork servers (gunicorn, uwsgi) correctly run startup work in each
child.  Capabilities (currently {"shm"}) ride alongside the kind so
zero-copy paths can be detected without bloating the enum.  CallContext
gains a kind field for per-call branching.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rustyconover

CLAUDE.md

@@ -107,7 +107,9 @@ For HTTP transport, the wire protocol maps to separate endpoints: `POST /vgi/{me
 
 **Stream state**: Streaming methods return a `Stream[S]` where `S` is a `StreamState` subclass. The state's `process(input, out, ctx)` method is called once per iteration. Producer streams (default `input_schema=_EMPTY_SCHEMA`) ignore the input and call `out.finish()` to end. Exchange streams set `input_schema` to a real schema and process client data.
 
-**CallContext injection**: Server method implementations can accept an optional `ctx: CallContext` parameter. `CallContext` provides `auth` (`AuthContext`), `client_log()` (client-directed logging), `emit_client_log` (raw `ClientLog` callback), `transport_metadata` (e.g. `remote_addr` from HTTP), and a `logger` property returning a `LoggerAdapter` with request context pre-bound. The parameter is injected by the framework — it does **not** appear in the Protocol definition.
+**CallContext injection**: Server method implementations can accept an optional `ctx: CallContext` parameter. `CallContext` provides `auth` (`AuthContext`), `client_log()` (client-directed logging), `emit_client_log` (raw `ClientLog` callback), `transport_metadata` (e.g. `remote_addr` from HTTP), `kind` (the active `TransportKind`), and a `logger` property returning a `LoggerAdapter` with request context pre-bound. The parameter is injected by the framework — it does **not** appear in the Protocol definition.
+
+**Transport awareness**: Workers can read `RpcServer.transport_kind` (a `TransportKind` enum: `PIPE`, `HTTP`, `UNIX`) and `RpcServer.transport_capabilities` (currently `{"shm"}` when bound to a `ShmPipeTransport`). For one-shot startup work an implementation may define an `on_serve_start(self, kind: TransportKind) -> None` method (the `ServeStartHook` Protocol); the framework calls it once per process before the first dispatch. For pipe/unix the hook fires inside `RpcServer.serve(transport)`; for HTTP it fires lazily on the first request handled in the current process so pre-fork servers (gunicorn, uwsgi) correctly run it in each child. Hook exceptions propagate (and are logged via `logging.getLogger("vgi_rpc.rpc").exception` first); rebinding to a different kind re-fires the hook rather than raising. Per-call code can also branch on `ctx.kind`.
 
 **Authentication**: `AuthContext` (frozen dataclass) carries `domain`, `authenticated`, `principal`, and `claims`. For HTTP transport, `make_wsgi_app(authenticate=...)` installs `_AuthMiddleware` that calls the callback on each request and populates `CallContext.auth`. Pipe transport gets anonymous auth by default. Methods can call `ctx.auth.require_authenticated()` to gate access.
 

docs/api/auth.md

@@ -65,6 +65,20 @@ Over pipe/subprocess transport, `ctx.auth` is always `AuthContext.anonymous()`.
 
 `ctx.transport_metadata` provides transport-level information like `remote_addr` and `user_agent` (HTTP only). It's a read-only mapping populated by the transport layer.
 
+`ctx.kind` exposes the active `TransportKind` (`PIPE`, `HTTP`, or `UNIX`) so methods can branch per-call:
+
+```python
+from vgi_rpc import CallContext, TransportKind
+
+
+def fetch(self, key: str, ctx: CallContext) -> str:
+    if ctx.kind is TransportKind.HTTP:
+        return self._cached_lookup(key)
+    return self._direct_lookup(key)
+```
+
+For one-shot startup work driven by transport kind, see [`on_serve_start`](transports.md#transport-awareness) on the transports page.
+
 ## API Reference
 
 ### AuthContext

docs/api/transports.md

@@ -88,6 +88,46 @@ server_transport = ShmPipeTransport(server_pipe, shm)
 
 Falls back to normal pipe IPC for batches that exceed the segment size.
 
+## Transport awareness
+
+Workers can ask which transport they are bound to — useful for tailoring startup work, enabling transport-specific metrics, or branching per-call behaviour. Three things are exposed:
+
+- `RpcServer.transport_kind`: a `TransportKind` enum (`PIPE`, `HTTP`, `UNIX`) or `None` before serving begins.
+- `RpcServer.transport_capabilities`: a `frozenset[str]` of capability flags. Currently `{"shm"}` when bound to a `ShmPipeTransport`; empty otherwise.
+- `CallContext.kind`: per-call view of the same `TransportKind`, so methods that already accept `ctx` can branch without reaching for the server.
+
+For one-shot startup work, an implementation may define an `on_serve_start(self, kind)` method. The framework calls it once per process before the first request is dispatched:
+
+```python
+from vgi_rpc import CallContext, TransportKind
+
+
+class MyServiceImpl:
+    def on_serve_start(self, kind: TransportKind) -> None:
+        """Called once per process before the first request."""
+        if kind is TransportKind.HTTP:
+            self._cache = build_http_cache()
+        else:
+            self._cache = None
+
+    def fetch(self, key: str, ctx: CallContext) -> str:
+        if ctx.kind is TransportKind.HTTP and self._cache is not None:
+            return self._cache.get(key)
+        return load_from_disk(key)
+```
+
+The hook is duck-typed (no base class needed); a `ServeStartHook` Protocol is exported for users who want to type-hint their implementation. Hook exceptions propagate (and are logged via `logging.getLogger("vgi_rpc.rpc").exception` first), so a misconfigured worker dies loudly rather than serving in a broken state.
+
+For pipe / unix transports the hook fires inside `RpcServer.serve(transport)`. For HTTP it fires lazily on the first request handled in the current process — this is fork-safe under pre-fork WSGI servers (gunicorn, uwsgi), so each child worker runs its own startup logic. Subprocess workers report `PIPE` because they speak Arrow IPC over the parent's stdin/stdout.
+
+`SHM` availability is exposed via `transport_capabilities`, not the enum, so coarse transport-kind checks stay simple while workers that need zero-copy paths can still detect shared memory:
+
+```python
+def on_serve_start(self, kind: TransportKind) -> None:
+    if "shm" in self.server.transport_capabilities:
+        self._enable_zero_copy()
+```
+
 ## API Reference
 
 ### PipeTransport
@@ -106,6 +146,14 @@ Falls back to normal pipe IPC for batches that exceed the segment size.
 
 ::: vgi_rpc.rpc.StderrMode
 
+### TransportKind
+
+::: vgi_rpc.rpc.TransportKind
+
+### ServeStartHook
+
+::: vgi_rpc.rpc.ServeStartHook
+
 ### Utility Functions
 
 ::: vgi_rpc.rpc.make_pipe_pair

tests/serve_fixture_kind_pipe.py

@@ -0,0 +1,23 @@
+# © Copyright 2025-2026, Query.Farm LLC - https://query.farm
+# SPDX-License-Identifier: Apache-2.0
+
+"""Subprocess worker for ``test_transport_kind`` end-to-end tests.
+
+Serves a tiny ``_KindService`` over stdin/stdout pipes via ``run_server``
+so the parent test can confirm the worker's ``on_serve_start`` sees
+``TransportKind.PIPE`` even when launched as a child process.
+"""
+
+from __future__ import annotations
+
+from tests.test_transport_kind import _KindService, _RecordingImpl
+from vgi_rpc.rpc import run_server
+
+
+def main() -> None:
+    """Serve the kind-reporting fixture over stdin/stdout."""
+    run_server(_KindService, _RecordingImpl())
+
+
+if __name__ == "__main__":
+    main()