feat(connection): dispatch TypeRT remotes through shared API

packages/client/connection/src/api-request-trust.ts

@@ -16,12 +16,13 @@
 import type { IncomingHttpHeaders } from 'node:http'
 import { isLoopbackHostname } from './loopback-hostname.ts'
 
-/** The request facts the fence reads (structural subset of IncomingMessage). */
+/** The request facts the fence reads from either HTTP representation. */
 interface ApiTrustRequest {
-  headers: IncomingHttpHeaders
+  headers: IncomingHttpHeaders | Headers
 }
 
-function header(headers: IncomingHttpHeaders, name: string): string | undefined {
+function header(headers: IncomingHttpHeaders | Headers, name: string): string | undefined {
+  if (headers instanceof Headers) return headers.get(name) ?? undefined
   const value = headers[name]
   return typeof value === 'string' ? value : undefined
 }
@@ -88,7 +89,7 @@ function isTrustedAuthority(hostUrl: URL, trustedHosts: readonly string[]): bool
 
 /**
  * Decide whether one /api request may reach the RPC bridge.
- * @param request - node HTTP request facts (headers).
+ * @param request - Node HTTP or Fetch request facts (headers).
  * @param trustedHosts - non-loopback authorities this deployment serves: exact `host:port`, or port-less `host` matching any port.
  * @returns true when the Host is ours (loopback or trusted) and any attached browser markers are same-origin.
  */

packages/client/connection/src/http-bridge.ts

@@ -5,7 +5,13 @@
 
 import type { IncomingMessage, ServerResponse } from 'node:http'
 
-interface FetchHandler {
+/** Transport-independent request handler consumed by the Host HTTP bridge. */
+export interface FetchHandler {
+  /**
+   * Handle one standard Fetch request.
+   * @param request - request produced by the active transport bridge.
+   * @returns complete or streaming Fetch response.
+   */
   fetch(request: Request): Promise<Response>
 }
 

packages/client/connection/src/index.ts

@@ -12,6 +12,7 @@ import { rejectWebSocketUpgrade, WebSocketDownlinks } from './websocket-downlink
 
 export type {
   ConnectionRpcAuthority,
+  ConnectionRpcEndpointMatcher,
   ConnectionRpcHandler,
   ConnectionRpcHandlerOptions,
   HostConnectionHandle,
@@ -24,7 +25,7 @@ export { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'
 /** Stable Cordis plugin name. */
 export const name = 'client-connection'
 
-/** Services required before providing Connection; legacy `/api` attaches when apiProxy is present. */
+/** Services required before providing Connection; API Proxy is an optional `/api` fallback. */
 export const inject = ['httpServer']
 
 /** Plugin config: the deployment's non-loopback serving authorities. */
@@ -93,35 +94,44 @@ export function apply(ctx: Context, config?: ConnectionConfig): void {
   // Config boundary: a malformed entry fails the load loudly here rather than
   // silently authorizing its hostname prefix at request time.
   for (const entry of trustedHosts) assertTrustedAuthority(entry)
-  new HostConnectionService(ctx, trustedHosts)
+  const connection = new HostConnectionService(ctx, trustedHosts)
+  const fetchHandler = connection.createSharedFetchHandler(API_PATH, {
+    async fetch(request) {
+      const pathname = new URL(request.url).pathname
+      const method = pathname.startsWith(`${API_PATH}/`)
+        ? pathname.slice(API_PATH.length + 1)
+        : undefined
+      if (method !== undefined
+        && PRIVILEGED_METHODS.has(method)
+        && !isTrustedApiRequest(request, [])) {
+        return new Response('forbidden', { status: 403 })
+      }
+      if (request.method === 'GET' && (pathname === MUX_EVENTS_PATH || pathname === HOST_EVENTS_PATH)) {
+        return new Response('upgrade required', {
+          status: 426,
+          headers: { connection: 'Upgrade', upgrade: 'websocket' },
+        })
+      }
+      const apiProxy = ctx.get('apiProxy')
+      if (apiProxy === undefined) return new Response('not found', { status: 404 })
+      return toFetchHandler(apiProxy).fetch(request)
+    },
+  })
+  const route: WebRoute = {
+    kind: 'prefix',
+    path: API_PATH,
+    handler: async (req, res) => {
+      if (!isTrustedApiRequest(req, trustedHosts)) {
+        res.writeHead(403)
+        res.end('forbidden')
+        return
+      }
+      await bridge(req, res, fetchHandler)
+    },
+  }
+  ctx.effect(() => ctx.httpServer.register(route), 'client-connection: /api route')
   ctx.inject(['apiProxy'], (apiCtx) => {
-    const apiHandler = toFetchHandler(apiCtx.apiProxy)
     const downlinks = new WebSocketDownlinks(apiCtx.apiProxy)
-    const route: WebRoute = {
-      kind: 'prefix',
-      path: API_PATH,
-      handler: async (req, res) => {
-        const pathname = new URL(req.url ?? '/', 'http://dsh.internal').pathname
-        const method = pathname.startsWith(`${API_PATH}/`)
-          ? pathname.slice(API_PATH.length + 1)
-          : undefined
-        const allowed = method !== undefined && PRIVILEGED_METHODS.has(method)
-          ? isTrustedApiRequest(req, [])
-          : isTrustedApiRequest(req, trustedHosts)
-        if (!allowed) {
-          res.writeHead(403)
-          res.end('forbidden')
-          return
-        }
-        if (req.method === 'GET' && (pathname === MUX_EVENTS_PATH || pathname === HOST_EVENTS_PATH)) {
-          res.writeHead(426, { connection: 'Upgrade', upgrade: 'websocket' })
-          res.end('upgrade required')
-          return
-        }
-        await bridge(req, res, apiHandler)
-      },
-    }
-    apiCtx.effect(() => apiCtx.httpServer.register(route), 'client-connection: /api route')
     const registerDownlink = (
       path: string,
       handle: WebUpgradeRoute['handler'],

packages/client/connection/src/rpc-host.ts

@@ -11,9 +11,11 @@ import {
   type RpcId as RpcIdType,
   type ServerResponse as RpcServerResponse,
 } from '@deepseek-ai/dsh-host-apiproxy/api'
-import { bridge } from './http-bridge.ts'
+import { bridge, type FetchHandler } from './http-bridge.ts'
 import { isTrustedApiRequest } from './api-request-trust.ts'
+import { API_PATH } from './api-path.ts'
 import type {
+  ConnectionRpcEndpointMatcher,
   ConnectionRpcHandler,
   ConnectionRpcHandlerOptions,
   HostConnectionHandle,
@@ -24,8 +26,23 @@ const INVALID_REQUEST_RPC_ID = RpcId('invalid-request')
 const CHANNEL_PATTERN = /^\/[A-Za-z0-9._~-]+$/
 const ENDPOINT_SEGMENT_PATTERN = /^[A-Za-z0-9_$.-]+$/
 
+interface ConnectionRpcInterceptor {
+  readonly matches: ConnectionRpcEndpointMatcher
+  readonly fetchHandler: FetchHandler
+  readonly options: ConnectionRpcHandlerOptions
+}
+
+declare module 'cordis' {
+  interface Context {
+    /** Host Connection transport and RPC registrations. */
+    connection: HostConnectionHandle
+  }
+}
+
 /** Host Connection service whose channel registrations belong to the caller fiber. */
 export class HostConnectionService extends Service implements HostConnectionHandle {
+  private readonly interceptors = new Map<string, ConnectionRpcInterceptor>()
+
   /**
    * Provide the Host half over the active HTTP server.
    * @param ctx - owning Connection plugin context.
@@ -40,6 +57,33 @@ export class HostConnectionService extends Service implements HostConnectionHand
     const owner = this.ctx
     return {
       handle: (channel, handler, options) => this.register(owner, channel, handler, options),
+      intercept: (channel, matches, handler, options) =>
+        this.registerInterceptor(owner, channel, matches, handler, options),
+    }
+  }
+
+  /**
+   * Compose one shared-channel Fetch handler from its interceptor and fallback.
+   * @param channel - shared channel mounted by Connection.
+   * @param fallback - handler for endpoints not claimed by the interceptor.
+   * @returns Fetch handler that selects exactly one target for each request.
+   */
+  createSharedFetchHandler(
+    channel: '/api',
+    fallback: FetchHandler,
+  ): FetchHandler {
+    return {
+      fetch: (request) => {
+        const endpoint = endpointFromPath(channel, new URL(request.url).pathname)
+        const interceptor = this.interceptors.get(channel)
+        if (endpoint === undefined || interceptor === undefined || !interceptor.matches(endpoint)) {
+          return fallback.fetch(request)
+        }
+        if (interceptor.options.authority === 'loopback' && !isTrustedApiRequest(request, [])) {
+          return Promise.resolve(new Response('forbidden', { status: 403 }))
+        }
+        return interceptor.fetchHandler.fetch(request)
+      },
     }
   }
 
@@ -69,12 +113,38 @@ export class HostConnectionService extends Service implements HostConnectionHand
       `client-connection: ${channel} rpc channel`,
     )
   }
+
+  private registerInterceptor(
+    owner: Context,
+    channel: string,
+    matches: ConnectionRpcEndpointMatcher,
+    handler: ConnectionRpcHandler,
+    options: ConnectionRpcHandlerOptions,
+  ): () => Promise<void> {
+    if (channel !== API_PATH) {
+      throw new Error(`connection: invalid shared RPC channel ${JSON.stringify(channel)}`)
+    }
+    const interceptor: ConnectionRpcInterceptor = {
+      matches,
+      fetchHandler: rpcFetchHandler(channel, handler),
+      options,
+    }
+    return owner.effect(() => {
+      if (this.interceptors.has(channel)) {
+        throw new Error(`connection: shared RPC channel ${JSON.stringify(channel)} already has an interceptor`)
+      }
+      this.interceptors.set(channel, interceptor)
+      return () => {
+        this.interceptors.delete(channel)
+      }
+    }, `client-connection: ${channel} rpc interceptor`)
+  }
 }
 
 function rpcFetchHandler(
   channel: string,
   handler: ConnectionRpcHandler,
-): { fetch(request: Request): Promise<Response> } {
+): FetchHandler {
   return {
     async fetch(request: Request): Promise<Response> {
       const endpoint = endpointFromPath(channel, new URL(request.url).pathname)

packages/client/connection/src/rpc.ts

@@ -18,11 +18,14 @@ export type ConnectionRpcHandler = (
   signal: AbortSignal,
 ) => Promise<RpcResult<unknown>>
 
+/** Synchronous ownership test for one endpoint on a shared RPC channel. */
+export type ConnectionRpcEndpointMatcher = (endpoint: string) => boolean
+
 /** Host registry for logical RPC channels carried by the current transport. */
 export interface HostConnectionRpc {
   /**
    * Register one absolute channel prefix and its trust policy.
-   * @param channel - absolute logical channel such as `/api2`.
+   * @param channel - absolute logical channel such as `/rpc`.
    * @param handler - decoded endpoint handler returning the existing RPC result shape.
    * @param options - channel trust policy.
    * @returns asynchronous disposer removing the channel and its physical route.
@@ -32,6 +35,21 @@ export interface HostConnectionRpc {
     handler: ConnectionRpcHandler,
     options: ConnectionRpcHandlerOptions,
   ): () => Promise<void>
+
+  /**
+   * Intercept owned endpoints on the shared `/api` channel before its fallback.
+   * @param channel - reserved shared channel; currently `/api`.
+   * @param matches - synchronous endpoint ownership test.
+   * @param handler - decoded endpoint handler returning the existing RPC result shape.
+   * @param options - trust policy for every endpoint claimed by this interceptor.
+   * @returns asynchronous disposer removing the interceptor.
+   */
+  intercept(
+    channel: '/api',
+    matches: ConnectionRpcEndpointMatcher,
+    handler: ConnectionRpcHandler,
+    options: ConnectionRpcHandlerOptions,
+  ): () => Promise<void>
 }
 
 /** Host `ctx.connection` shape consumed by transport-independent adapters. */
@@ -44,7 +62,7 @@ export interface HostConnectionHandle {
 export interface ClientConnectionRpc {
   /**
    * Call one endpoint through an already registered logical channel.
-   * @param channel - absolute logical channel such as `/api2`.
+   * @param channel - absolute logical channel such as `/api`.
    * @param endpoint - channel-relative endpoint such as `goals/create`.
    * @param payload - channel-owned request payload.
    * @param signal - optional caller cancellation.