rpc: surface protocol implementation on CallContext

Add ``implementation`` to CallContext so framework-driven callbacks
(producer-mode stream states, on_cancel hooks) can dispatch helper
calls back through the public protocol surface without holding their
own reference to the impl instance.  Includes any meta-worker that
the top-level RpcServer is fronting.

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

Author: rustyconover

vgi_rpc/http/server/_app_stream.py

@@ -225,6 +225,7 @@ def _run_stream_init_sync(
                 method_name=method_name,
                 protocol_name=protocol_name,
                 kind=app._server.transport_kind,
+                implementation=app._server.implementation,
             )
 
         # The chain-correlation id for all HTTP turns of this stream.  Carried
@@ -492,6 +493,7 @@ def _run_stream_exchange_sync(
                     method_name=method_name,
                     protocol_name=protocol_name,
                     kind=app._server.transport_kind,
+                    implementation=app._server.implementation,
                 )
                 try:
                     state_obj.on_cancel(cancel_ctx)
@@ -641,6 +643,7 @@ def _run_http_exchange_turn(
             method_name=method_name,
             protocol_name=protocol_name,
             kind=app._server.transport_kind,
+            implementation=app._server.implementation,
         )
         state.process(ab_in, out, process_ctx)
         if not out.finished:
@@ -831,6 +834,7 @@ def _emit_to_current(msg: Message) -> None:
             method_name=method_name,
             protocol_name=protocol_name,
             kind=app._server.transport_kind,
+            implementation=app._server.implementation,
         )
         try:
             while True:

vgi_rpc/http/server/_app_unary.py

@@ -121,6 +121,7 @@ def _run_unary_sync(
                 method_name=method_name,
                 protocol_name=protocol_name,
                 kind=app._server.transport_kind,
+                implementation=app._server.implementation,
             )
 
         schema = info.result_schema

vgi_rpc/rpc/_common.py

@@ -172,6 +172,7 @@ class CallContext:
         "_server_id",
         "auth",
         "emit_client_log",
+        "implementation",
         "kind",
         "transport_metadata",
     )
@@ -186,12 +187,24 @@ def __init__(
         method_name: str = "",
         protocol_name: str = "",
         kind: TransportKind | None = None,
+        implementation: Any = None,
     ) -> None:
-        """Initialize with auth context, client-log callback, and optional server context fields."""
+        """Initialize with auth context, client-log callback, and optional server context fields.
+
+        Args:
+            implementation: The protocol implementation object the
+                RpcServer was constructed with. Surfaced on the context so
+                producer-mode stream states (and other framework-driven
+                callbacks that lack direct access to the implementation
+                instance) can dispatch helper calls back through the
+                public protocol surface — including any meta-worker
+                dispatching the top-level RpcServer is fronting.
+        """
         self.auth = auth
         self.emit_client_log = emit_client_log
         self.transport_metadata: Mapping[str, Any] = transport_metadata or {}
         self.kind: TransportKind | None = kind
+        self.implementation: Any = implementation
         self._server_id = server_id
         self._method_name = method_name
         self._protocol_name = protocol_name

vgi_rpc/rpc/_server.py

@@ -630,6 +630,7 @@ def _prepare_method_call(
                 method_name=info.name,
                 protocol_name=self.protocol_name,
                 kind=self._transport_kind,
+                implementation=self._impl,
             )
         return sink, auth, transport_metadata
 
@@ -818,6 +819,7 @@ def _serve_stream(
                                 method_name=info.name,
                                 protocol_name=protocol_name,
                                 kind=self._transport_kind,
+                                implementation=self._impl,
                             )
                             try:
                                 state.on_cancel(cancel_ctx)
@@ -862,6 +864,7 @@ def _serve_stream(
                             method_name=info.name,
                             protocol_name=protocol_name,
                             kind=self._transport_kind,
+                            implementation=self._impl,
                         )
                         state.process(ab_in, out, process_ctx)
                         if not out.finished: