Add hybrid shared runtime routing

docs/phase-foundation-roadmap.md

@@ -111,7 +111,7 @@ quotas, worker isolation, distributed state, hybrid routing, and E2E proof.
 - [x] Add isolated shared worker runtime for allowlisted stateless upstreams.
 - [x] Add distributed deployment state with locking, conflict handling,
       rollback, recovery, and audit.
-- [ ] Add hybrid routing between local edge tools and shared workers.
+- [x] Add hybrid routing between local edge tools and shared workers.
 - [ ] Add shared-runtime E2E proof for tenant isolation, authz denial, quota
       denial, session affinity, audit, rollback, and degraded mode.
 

docs/protocol.md

@@ -228,3 +228,12 @@ upstream and original upstream tool name, passes the upstream call timeout from
 config, and maps upstream timeout, crash, unknown-tool, invalid-argument,
 disabled-prefix, profile-denial, and mutating-profile-denial failures to
 broker-owned error codes.
+
+`mcp_broker.hybrid_router.HybridToolRouter` is the Phase 3 route classifier for
+ordinary upstream `tools/call` requests. It preserves the same client-visible
+tool name and argument object. By default it calls the local edge upstream path.
+Only upstreams tagged `stateless` and `shared-worker`, configured in shared
+mode, free of local state/env/session metadata, and supplied with tenant,
+team, and quota metadata can route to the in-process fake shared worker proof.
+Quota denial on that shared path fails closed and does not retry through the
+local edge broker.

docs/shared-runtime-guardrails.md

@@ -183,6 +183,31 @@ shared-runtime state under the runtime state directory, appends a rollback
 journal before recovery replay is needed, and fails closed on lock or revision
 conflicts.
 
+## Hybrid Routing
+
+Contract statement: hybrid routing preserves the existing tools/call client shape.
+Client-visible tool names and argument objects remain unchanged when a route is
+classified for local edge or shared-worker execution.
+
+Contract statement: local-only tools continue to route to the edge broker.
+Local-only includes stateful, browser, file-access, local-secret, OAuth,
+unknown, mutating, configured-state, env-backed, request-meta-backed, and
+session-env-backed upstreams.
+
+Contract statement: allowlisted stateless tools can route to shared workers only after quota approval.
+The current allowlist is config-derived from upstream tags and safety metadata:
+the upstream must be tagged stateless and shared-worker, use shared mode, avoid
+local state, and pass the supplied quota snapshot. The default route remains
+local edge when shared-runtime metadata is absent.
+
+Contract statement: shared-worker quota denial does not fall back to local edge execution.
+Quota denial is returned as an audited denial for the shared-worker attempt so a
+shared route cannot bypass quota by retrying locally.
+
+The code contract lives in `mcp_broker.hybrid_router`. The daemon delegates
+ordinary upstream calls through the hybrid router while preserving broker
+catalog tools and the local edge fallback.
+
 ## Mandatory Non-Goals
 
 Phase 3 does not add hosted execution. It does not add remote tool calls, remote

src/mcp_broker/daemon.py

@@ -13,7 +13,7 @@
 
 from mcp_broker import __version__
 from mcp_broker.broker import BrokerCore, BrokerToolError
-from mcp_broker.catalog import BrokerCatalogFacade, profile_allows_upstream
+from mcp_broker.catalog import profile_allows_upstream
 from mcp_broker.config import BrokerConfig, UpstreamConfig
 from mcp_broker.daemon_helpers import (
     configured_upstream_health as _configured_upstream_health,
@@ -30,10 +30,12 @@
 from mcp_broker.daemon_provenance import source_provenance as _source_provenance_impl
 from mcp_broker.daemon_request_context import BrokerDaemonRequestContextMixin
 from mcp_broker.daemon_status import BrokerDaemonStatusMixin
+from mcp_broker.daemon_tool_calls import BrokerDaemonToolCallMixin
 from mcp_broker.daemon_upstreams import BrokerDaemonUpstreamMixin
 from mcp_broker.jsonrpc import JsonRpcRequest, JsonRpcResponse
 from mcp_broker.protocol import McpProtocolHandler
 from mcp_broker.runtime_reaper import RuntimePaths, write_socket_metadata
+from mcp_broker.shared_worker import SharedWorkerRuntime
 from mcp_broker.upstream_http import HttpUpstreamClient, HttpUpstreamError
 from mcp_broker.upstream_protocols import (
     HttpUpstreamClientProtocol,
@@ -75,6 +77,7 @@ class BrokerDaemon(
     BrokerDaemonStatusMixin,
     BrokerDaemonUpstreamMixin,
     BrokerDaemonRequestContextMixin,
+    BrokerDaemonToolCallMixin,
 ):
     runtime_root: Path
     socket_path: Path
@@ -108,6 +111,7 @@ def __post_init__(self) -> None:
         self._last_request_method: str | None = None
         self._last_request_status: str | None = None
         self._auth_repair_stats: dict[str, dict[str, int | str]] = {}
+        self._shared_worker_runtime = SharedWorkerRuntime(tools=[])
 
     @property
     def lock_path(self) -> Path:
@@ -328,53 +332,6 @@ def _handle_tools_list(self, request: JsonRpcRequest) -> JsonRpcResponse:
             return JsonRpcResponse.error(request.id, -32000, str(exc))
         return JsonRpcResponse.result(request.id, result)
 
-    def _handle_tools_call(self, request: JsonRpcRequest) -> JsonRpcResponse:
-        if self.broker_config is None:
-            return JsonRpcResponse.error(request.id, -32000, "broker config is not loaded")
-        params = request.params
-        if not isinstance(params, dict):
-            return JsonRpcResponse.error(request.id, -32602, "tools/call params must be an object")
-        name = params.get("name")
-        arguments = params.get("arguments", {})
-        if not isinstance(name, str) or not isinstance(arguments, dict):
-            return JsonRpcResponse.error(request.id, -32602, "tools/call name and arguments required")
-        try:
-            session_id = self._session_id_from_params(params)
-            session_context = self._session_context_from_params(params)
-            profile = self._effective_profile(params, session_context)
-        except ValueError as exc:
-            return JsonRpcResponse.error(request.id, -32602, str(exc))
-        call_upstream = self._call_upstream_for_session(session_id, session_context)
-        list_upstream = self._list_upstream_for_session(session_id, session_context)
-        canonical_name = profile.canonical_broker_tool_name(name) if profile is not None else name
-        if canonical_name.startswith("broker."):
-            try:
-                result = BrokerCatalogFacade(
-                    broker_config=self.broker_config,
-                    profile=profile,
-                    list_upstream=list_upstream,
-                    call_upstream=call_upstream,
-                    call_locks=self._upstream_call_locks,
-                    status_provider=self._upstream_health_for_status,
-                    client_cwd=session_context.get("client_cwd"),
-                ).call_tool(name, arguments)
-            except (BrokerToolError, ValueError) as exc:
-                return JsonRpcResponse.error(request.id, -32000, str(exc))
-            return JsonRpcResponse.result(request.id, result)
-        arguments = self._inject_cwd_project_arg(name, arguments, session_context)
-        core = BrokerCore(
-            settings=self.broker_config.broker,
-            upstreams=self.broker_config.upstreams,
-            profile=profile,
-            call_locks=self._upstream_call_locks,
-        )
-        try:
-            result = core.call_tool(name, arguments, call_upstream)
-        except (BrokerToolError, ValueError) as exc:
-            message = exc.message if isinstance(exc, BrokerToolError) else str(exc)
-            return JsonRpcResponse.error(request.id, -32000, message)
-        return JsonRpcResponse.result(request.id, result)
-
     def _create_stdio_upstream_process(
         self,
         upstream: UpstreamConfig,

src/mcp_broker/daemon_tool_calls.py

@@ -0,0 +1,109 @@
+"""Daemon JSON-RPC tools/call handling."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp_broker.broker import BrokerToolError
+from mcp_broker.catalog import BrokerCatalogFacade
+from mcp_broker.hybrid_router import HybridRoutingContext, HybridToolRouter
+from mcp_broker.jsonrpc import JsonRpcRequest, JsonRpcResponse
+
+
+class BrokerDaemonToolCallMixin:
+    def _handle_tools_call(self, request: JsonRpcRequest) -> JsonRpcResponse:
+        if self.broker_config is None:
+            return JsonRpcResponse.error(request.id, -32000, "broker config is not loaded")
+        params = request.params
+        if not isinstance(params, dict):
+            return JsonRpcResponse.error(request.id, -32602, "tools/call params must be an object")
+        name = params.get("name")
+        arguments = params.get("arguments", {})
+        if not isinstance(name, str) or not isinstance(arguments, dict):
+            return JsonRpcResponse.error(request.id, -32602, "tools/call name and arguments required")
+        try:
+            session_id = self._session_id_from_params(params)
+            session_context = self._session_context_from_params(params)
+            profile = self._effective_profile(params, session_context)
+        except ValueError as exc:
+            return JsonRpcResponse.error(request.id, -32602, str(exc))
+        call_upstream = self._call_upstream_for_session(session_id, session_context)
+        list_upstream = self._list_upstream_for_session(session_id, session_context)
+        canonical_name = profile.canonical_broker_tool_name(name) if profile is not None else name
+        if canonical_name.startswith("broker."):
+            return self._handle_broker_catalog_tool_call(
+                request_id=request.id,
+                name=name,
+                arguments=arguments,
+                profile=profile,
+                list_upstream=list_upstream,
+                call_upstream=call_upstream,
+                session_context=session_context,
+            )
+        return self._handle_upstream_tool_call(
+            request_id=request.id,
+            name=name,
+            arguments=arguments,
+            params=params,
+            profile=profile,
+            call_upstream=call_upstream,
+            session_context=session_context,
+        )
+
+    def _handle_broker_catalog_tool_call(
+        self,
+        *,
+        request_id: str | int | None,
+        name: str,
+        arguments: dict[str, Any],
+        profile: Any,
+        list_upstream: Any,
+        call_upstream: Any,
+        session_context: dict[str, str],
+    ) -> JsonRpcResponse:
+        try:
+            result = BrokerCatalogFacade(
+                broker_config=self.broker_config,
+                profile=profile,
+                list_upstream=list_upstream,
+                call_upstream=call_upstream,
+                call_locks=self._upstream_call_locks,
+                status_provider=self._upstream_health_for_status,
+                client_cwd=session_context.get("client_cwd"),
+            ).call_tool(name, arguments)
+        except (BrokerToolError, ValueError) as exc:
+            return JsonRpcResponse.error(request_id, -32000, str(exc))
+        return JsonRpcResponse.result(request_id, result)
+
+    def _handle_upstream_tool_call(
+        self,
+        *,
+        request_id: str | int | None,
+        name: str,
+        arguments: dict[str, Any],
+        params: dict[str, Any],
+        profile: Any,
+        call_upstream: Any,
+        session_context: dict[str, str],
+    ) -> JsonRpcResponse:
+        arguments = self._inject_cwd_project_arg(name, arguments, session_context)
+        shared_context = HybridRoutingContext.from_params(params)
+        try:
+            result = HybridToolRouter(
+                upstreams=self.broker_config.upstreams,
+                settings=self.broker_config.broker,
+                profile=profile,
+                call_locks=self._upstream_call_locks,
+                shared_worker=self._shared_worker_runtime,
+            ).call_tool(
+                advertised_name=name,
+                arguments=arguments,
+                edge_caller=call_upstream,
+                tenant_context=shared_context.tenant_context,
+                team_id=shared_context.team_id,
+                quota_snapshot=shared_context.quota_snapshot,
+            )
+        except (BrokerToolError, ValueError) as exc:
+            message = exc.message if isinstance(exc, BrokerToolError) else str(exc)
+            return JsonRpcResponse.error(request_id, -32000, message)
+        return JsonRpcResponse.result(request_id, result)