Merge remote-tracking branch 'origin/master' into worktree/web-multimodal-image-input

# Conflicts: # .agents/notes/implemented/architecture/2026-07-05-reconstructable-requests.i18n.yaml # .agents/notes/implemented/architecture/2026-07-05-reconstructable-requests.md # .agents/notes/implemented/architecture/2026-07-05-reconstructable-requests.zh.md # .agents/notes/implemented/architecture/2026-07-25-web-input-machine-and-slash-pipeline.i18n.yaml # .agents/notes/implemented/architecture/2026-07-25-web-input-machine-and-slash-pipeline.md # .agents/notes/implemented/architecture/2026-07-25-web-input-machine-and-slash-pipeline.zh.md # THIRD_PARTY_NOTICES.md # apps/cli/composition.md # apps/cli/config/base.cordis.yml # apps/cli/package.json # apps/cli/src/app-cli-entry.ts # apps/cli/src/bin.ts # apps/cli/tests/args.spec.ts # apps/web/tests/built-boot.snapshot.ts # apps/web/tests/navigation-panes.e2e.ts # docs/architecture.i18n.yaml # docs/architecture.md # docs/architecture.zh.md # docs/config-catalog.md # docs/cordis-catalog/services.md # docs/core-data-structures/core.i18n.yaml # docs/core-data-structures/llm-streaming.i18n.yaml # docs/event-producer-consumer.md # docs/module-graph.md # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl # packages/README.i18n.yaml # packages/bundle/README.i18n.yaml # packages/client/connection/README.i18n.yaml # packages/client/connection/README.md # packages/client/connection/README.zh.md # packages/client/connection/src/client/fixture.ts # packages/client/connection/src/http-bridge.ts # packages/client/connection/src/index.ts # packages/client/connection/tests/fixture.spec.ts # packages/client/connection/tests/node-half.spec.ts # packages/client/runtime/README.i18n.yaml # packages/client/runtime/README.md # packages/client/runtime/README.zh.md # packages/client/runtime/src/client/contract/session.ts # packages/client/runtime/src/client/sessions/session.ts # packages/client/ui-conversation/README.i18n.yaml # packages/client/ui-conversation/README.md # packages/client/ui-conversation/README.zh.md # packages/client/ui-conversation/src/client/apply.ts # packages/client/ui-conversation/src/client/chat/AssistantMarkdown.tsx # packages/client/ui-conversation/src/client/chat/ChatView.tsx # packages/client/ui-conversation/src/client/chat/MessageItem.module.css # packages/client/ui-conversation/src/client/chat/MessageItem.tsx # packages/client/ui-conversation/src/client/contract/slots.ts # packages/client/ui-conversation/src/client/index.ts # packages/client/ui-conversation/src/client/input/contract.ts # packages/client/ui-conversation/src/client/input/facade.ts # packages/client/ui-conversation/src/client/input/hub.ts # packages/client/ui-conversation/src/client/locales.ts # packages/client/ui-conversation/src/client/service.ts # packages/client/ui-conversation/src/client/skeleton/ConversationSession.tsx # packages/client/ui-conversation/src/client/skeleton/InputBar.tsx # packages/client/ui-conversation/tests/apply-inject.spec.tsx # packages/client/ui-conversation/tests/input-bar.spec.tsx # packages/client/ui-conversation/tests/input-matrix.spec.tsx # packages/client/ui-conversation/tests/input-scenarios.spec.tsx # packages/client/ui-conversation/tests/service-orchestration.spec.ts # packages/client/ui-conversation/tests/skeleton.spec.tsx # packages/client/ui-trajectory/tests/views.spec.tsx # packages/compact/compact-basic/README.i18n.yaml # packages/cordis/tool-cordis/src/api-catalog.ts # packages/host/apiproxy/README.i18n.yaml # packages/host/apiproxy/README.md # packages/host/apiproxy/README.zh.md # packages/host/apiproxy/src/api-proxy.ts # packages/host/apiproxy/src/api/rpc.ts # packages/host/apiproxy/src/api/sessions.ts # packages/host/apiproxy/src/index.ts # packages/host/apiproxy/tests/api-proxy-models.spec.ts # packages/host/apiproxy/tests/rpc-schemas.spec.ts # packages/llm/llm-pi-ai/README.i18n.yaml # packages/llm/llm-pi-ai/README.md # packages/llm/llm-pi-ai/README.zh.md # packages/llm/llm-pi-ai/src/adapter.ts # packages/llm/llm/README.i18n.yaml # packages/ui/tui/README.md # packages/ui/tui/README.zh.md # packages/ui/tui/src/components/content.ts # packages/ui/tui/src/components/transcript.ts # packages/ui/tui/tests/tui.spec.ts # pnpm-lock.yaml
2026-08-08 01:56:29 +08:00
parent 78d956a65d 698d65de2d
commit 25d959aa08
3973 changed files with 144864 additions and 60887 deletions
--- a/packages/client/connection/src/api-path.ts
+++ b/packages/client/connection/src/api-path.ts
@@ -1,8 +1,14 @@
 /**
 * The /api URL prefix — single source for both halves of the web transport.
- * The node half registers this prefix on the web server; browser-side path
- * literals currently live in the apiproxy client layer (out of scope here).
+ * The node half registers this prefix on the web server; both halves share the
+ * event paths below for the browser WebSocket downlinks.
 */

 /** Route prefix owning every api request (`/api` and `/api/<anything>`). */
 export const API_PATH = '/api'
+
+/** Browser mux-frame WebSocket pathname. */
+export const MUX_EVENTS_PATH = `${API_PATH}/events.mux`
+
+/** Browser host-frame WebSocket pathname. */
+export const HOST_EVENTS_PATH = `${API_PATH}/events.host`
--- a/packages/client/connection/src/api-request-trust.ts
+++ b/packages/client/connection/src/api-request-trust.ts
@@ -4,7 +4,7 @@
 * the attacker's domain while the socket reaches this server) and cross-site
 * requests fired from a malicious page. The Host fence binds every request,
 * browser-looking or not: over plain HTTP a browser attaches neither Origin
- * nor Fetch-Metadata to reads (EventSource, images, navigations — those
+ * nor Fetch-Metadata to reads (images and navigations — those
 * headers go only to trustworthy destinations), so an unmarked request may
 * still be a rebound browser read and Host is the one header rebinding cannot
 * forge. Non-browser and remote clients pass the same fence via loopback, the
@@ -14,25 +14,19 @@
 */

 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
 }

-function isLoopbackHostname(hostname: string): boolean {
-  if (hostname === 'localhost' || hostname === '[::1]') return true
-  const parts = hostname.split('.')
-  return parts.length === 4
-    && parts[0] === '127'
-    && parts.every(part => /^\d{1,3}$/.test(part) && Number(part) <= 255)
-}
-
 /** Normalized URL of a Host-header authority (hostname lowercased, default port stripped, IPv6 bracketed), or undefined when unparsable. */
 function parseAuthority(authority: string): URL | undefined {
  try {
@@ -95,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.
 */
@@ -104,7 +98,7 @@ export function isTrustedApiRequest(request: ApiTrustRequest, trustedHosts: read
  // fills Host from the URL it believes it is talking to, so a rebound page
  // carries the attacker's domain here even though the socket lands on this
  // server. There is no marker shortcut — a browser read over plain HTTP
-  // (EventSource, images, navigations) arrives with neither Origin nor
+  // (images and navigations) arrives with neither Origin nor
  // Fetch-Metadata, indistinguishable from curl, and its response is readable
  // by the rebound page.
  const host = header(request.headers, 'host')
--- a/packages/client/connection/src/client/api.ts
+++ b/packages/client/connection/src/client/api.ts
@@ -12,10 +12,11 @@ export type {
  ResponseValue, WorkspaceApi, WorkspaceId, WorkspaceView,
  CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
  ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
-  InboxItemId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
+  ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
  GoalsApi, GoalRef,
  SettingsApi, SettingsNamespaceView, SettingsPathOpView, SettingsSecretView,
-  CredentialsApi, CredentialView, ConfigurableProviderView, LlmApi,
+  CredentialsApi, CredentialView, ConfigurableProviderView, DiscoveredModelView, LlmApi,
+  SubagentsApi, SubagentAddress, SubagentCatalog, SubagentListEntry, SubagentPromptReceipt,
 } from '@deepseek-ai/dsh-host-apiproxy/api'
 export type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'
 export type {
@@ -33,6 +34,7 @@ export {
 export { AbstractApiClient } from '@deepseek-ai/dsh-host-apiproxy/client'
 export type { IApiClient } from '@deepseek-ai/dsh-host-apiproxy/client'
 export type { SessionId, SessionEvent } from '@deepseek-ai/dsh-session/types'
+export type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 export type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm/types'

 /** Successful value returned by the connection-generation host handshake. */
--- a/packages/client/connection/src/client/connection.ts
+++ b/packages/client/connection/src/client/connection.ts
@@ -126,7 +126,7 @@ export class ConnectionController {

      try {
        // Strict readiness handshake (audit C2): describe proves unary reachability, onOpen
-        // proves each SSE transport is established (response headers in, before any frame) —
+        // proves each physical stream is established before any frame —
        // only then may onConnected fire, so the resync it triggers cannot outrun the
        // subscribed baseline. The timeout guards against a carrier that never fires onOpen
        // (see ConnectionConfig.streamOpenTimeoutMs).
--- a/packages/client/connection/src/client/fixture.ts
+++ b/packages/client/connection/src/client/fixture.ts
--- a/packages/client/connection/src/client/index.ts
+++ b/packages/client/connection/src/client/index.ts
@@ -8,6 +8,9 @@ import type { IApiClient } from './api.ts'
 import { ConnectionController, type ConnectionConfig, type ConnectionSinks, type ConnectionState } from './connection.ts'
 import { FixtureApiClient } from './fixture.ts'
 import { WebApiClient } from './web-api-client.ts'
+import { createWebConnectionRpc } from './rpc.ts'
+import { isLoopbackHostname } from '../loopback-hostname.ts'
+import type { ClientConnectionRpc } from '../rpc.ts'

 // ---- Contract re-exports (browser-safe apiproxy channels + core types) ----
 export type {
@@ -17,13 +20,14 @@ export type {
  ToolCallView, ToolResultView, WorkspaceApi, WorkspaceId, WorkspaceView,
  CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
  ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
-  InboxItemId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
+  MessageId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
+  SubagentsApi, SubagentAddress, SubagentCatalog, SubagentListEntry, SubagentPromptReceipt,
  RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
  ClientRequest, ServerResponse, ServerRequest, ClientResponse, RpcMessage, RpcReceipt,
  HostDescription, IApiClient, SessionId, SessionEvent, ContentBlock, StreamChunk,
  GoalsApi, GoalRef,
  SettingsApi, SettingsNamespaceView, SettingsPathOpView, SettingsSecretView,
-  CredentialsApi, CredentialView, ConfigurableProviderView, LlmApi,
+  CredentialsApi, CredentialView, ConfigurableProviderView, DiscoveredModelView, LlmApi,
 } from './api.ts'
 export {
  RpcId,
@@ -34,6 +38,7 @@ export {
 // Connection loop types are public through ConnectionHandle.start; the
 // controller remains package-internal.
 export type { ConnectionConfig, ConnectionSinks, ConnectionState }
+export type { ClientConnectionRpc } from '../rpc.ts'


 /** Required services (none — this is the wire root). */
@@ -47,6 +52,10 @@ export const inject: string[] = []
 export interface ConnectionHandle {
  /** Shared api client (fixture or real, decided at boot from the page URL). */
  readonly api: IApiClient
+  /** Whether the current page authority is loopback; non-browser contexts default to true. */
+  readonly isLoopback: boolean
+  /** Generic logical RPC channels over the same Connection transport. */
+  readonly rpc: ClientConnectionRpc
  /**
   * Start the connect/pump/reconnect loop with the consumer's frame sinks.
   * One consumer owns the streams (the runtime object layer); a second call
@@ -63,11 +72,16 @@ export interface ConnectionHandle {
 * @param ctx - client cordis context.
 */
 export function apply(ctx: Context): void {
-  const fixture = typeof location !== 'undefined' && new URLSearchParams(location.search).has('fixture')
-  const api: IApiClient = fixture ? new FixtureApiClient() : new WebApiClient()
+  const pageLocation = typeof location === 'undefined' ? undefined : location
+  const fixture = pageLocation !== undefined && new URLSearchParams(pageLocation.search).has('fixture')
+  const fixtureClient = fixture ? new FixtureApiClient() : undefined
+  const api: IApiClient = fixtureClient ?? new WebApiClient()
+  const rpc = fixtureClient?.rpc ?? createWebConnectionRpc()
  let started = false
  const handle: ConnectionHandle = {
    api,
+    isLoopback: pageLocation === undefined || isLoopbackHostname(pageLocation.hostname),
+    rpc,
    start(sinks, config) {
      if (started) throw new Error('connection: the stream loop is already owned by another consumer')
      started = true
--- a/packages/client/connection/src/client/random-uuid.ts
+++ b/packages/client/connection/src/client/random-uuid.ts
@@ -0,0 +1,14 @@
+/** Browser-safe UUID generation for client-side wire correlation. */
+
+/**
+ * Generate an RFC 4122 version 4 UUID without requiring a secure context.
+ * @returns a UUID backed by `crypto.getRandomValues()`, which browsers expose on insecure origins.
+ */
+export function randomUuid(): string {
+  const bytes = globalThis.crypto.getRandomValues(new Uint8Array(16))
+  const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
+  view.setUint8(6, (view.getUint8(6) & 0x0f) | 0x40)
+  view.setUint8(8, (view.getUint8(8) & 0x3f) | 0x80)
+  const hex = Array.from(bytes, byte => byte.toString(16).padStart(2, '0')).join('')
+  return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`
+}
--- a/packages/client/connection/src/client/rpc.ts
+++ b/packages/client/connection/src/client/rpc.ts
@@ -0,0 +1,63 @@
+/** Browser caller for generic Connection unary RPC channels. */
+
+import {
+  RpcId,
+  serverResponseSchema,
+  type ClientRequest,
+} from '@deepseek-ai/dsh-host-apiproxy/api'
+import type { ClientConnectionRpc } from '../rpc.ts'
+import { randomUuid } from './random-uuid.ts'
+
+const INTERNAL_BASE = 'http://dsh.internal'
+const CHANNEL_PATTERN = /^\/[A-Za-z0-9._~-]+$/
+const ENDPOINT_SEGMENT_PATTERN = /^[A-Za-z0-9_$.-]+$/
+
+/**
+ * Create the browser-backed generic RPC caller.
+ * @returns caller that owns request correlation and response-envelope validation.
+ */
+export function createWebConnectionRpc(): ClientConnectionRpc {
+  return {
+    async call(channel, endpoint, payload, signal) {
+      assertTarget(channel, endpoint)
+      const rpcId = RpcId(randomUuid())
+      const message: ClientRequest = {
+        type: 'client-request',
+        rpcId,
+        method: endpoint,
+        payload,
+      }
+      const response = await globalThis.fetch(
+        new URL(`${channel}/${endpoint}`, resolveBase()),
+        {
+          method: 'POST',
+          headers: { 'content-type': 'application/json' },
+          body: JSON.stringify(message),
+          ...signal === undefined ? {} : { signal },
+        },
+      )
+      if (!response.ok) {
+        throw new Error(`transport failure for ${channel}/${endpoint}: HTTP ${response.status}`)
+      }
+      const full = serverResponseSchema.parse(await response.json())
+      if (full.rpcId !== rpcId) {
+        throw new Error(`rpcId mismatch for ${endpoint}: sent ${rpcId}, got ${full.rpcId}`)
+      }
+      return full.result
+    },
+  }
+}
+
+function resolveBase(): string {
+  const location = (globalThis as { location?: { origin?: string } }).location
+  return location?.origin !== undefined && location.origin !== 'null' ? location.origin : INTERNAL_BASE
+}
+
+function assertTarget(channel: string, endpoint: string): void {
+  const segments = endpoint.split('/')
+  if (!CHANNEL_PATTERN.test(channel)
+    || segments.some(segment =>
+      segment === '' || segment === '.' || segment === '..' || !ENDPOINT_SEGMENT_PATTERN.test(segment))) {
+    throw new Error(`connection: invalid RPC target ${JSON.stringify(`${channel}/${endpoint}`)}`)
+  }
+}
--- a/packages/client/connection/src/client/web-api-client.ts
+++ b/packages/client/connection/src/client/web-api-client.ts
@@ -1,12 +1,91 @@
-// WebApiClient: the browser platform subclass — transport = global fetch over same-origin
-// /api/* (base resolution handled by AbstractApiClient). Envelope observation comes from the
-// base batching aspect; subscribers attach via subscribeEnvelopes (see boot).
+/** Browser API carrier: HTTP upstream plus one WebSocket per downstream event stream. */

+import type { ApiProxy, HostFrame, MuxFrame, RpcRequest, ServerRequest } from './api.ts'
 import { AbstractApiClient } from './api.ts'
+import { hostFrameSchema, muxFrameSchema } from '@deepseek-ai/dsh-host-apiproxy/api/events.schema'
+import { serverRequestSchema } from '@deepseek-ai/dsh-host-apiproxy/api/rpc.schema'
+import { HOST_EVENTS_PATH, MUX_EVENTS_PATH } from '../api-path.ts'

-/** Browser platform subclass: transport = global fetch over same-origin /api/*. */
+type SocketItem<F> = { kind: 'frame'; envelope: RpcRequest<F> } | { kind: 'end' }
+type Parser<F> = { parse(value: unknown): F }
+
+/** Browser platform subclass: unary/respond use fetch; mux/host use downlink-only WebSockets. */
 export class WebApiClient extends AbstractApiClient {
  protected doFetch(input: URL, init?: RequestInit): Promise<Response> {
    return globalThis.fetch(input, init)
  }
+
+  protected override openMux(
+    _payload: Parameters<ApiProxy['events']['mux']>[0]['payload'],
+    signal: AbortSignal,
+    onOpen?: () => void,
+  ): AsyncIterable<RpcRequest<MuxFrame>> {
+    return this.readWebSocket(MUX_EVENTS_PATH, signal, muxFrameSchema, onOpen)
+  }
+
+  protected override openHost(
+    _payload: Parameters<ApiProxy['events']['host']>[0]['payload'],
+    signal: AbortSignal,
+    onOpen?: () => void,
+  ): AsyncIterable<RpcRequest<HostFrame>> {
+    return this.readWebSocket(HOST_EVENTS_PATH, signal, hostFrameSchema, onOpen)
+  }
+
+  private async *readWebSocket<F extends MuxFrame | HostFrame>(
+    path: string,
+    signal: AbortSignal,
+    frameSchema: Parser<F>,
+    onOpen?: () => void,
+  ): AsyncGenerator<RpcRequest<F>> {
+    const url = new URL(path, this.resolveBase())
+    url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'
+    const socket = new WebSocket(url)
+    const inbox: SocketItem<F>[] = []
+    let wake: (() => void) | undefined
+    const enqueue = (item: SocketItem<F>): void => {
+      inbox.push(item)
+      wake?.()
+      wake = undefined
+    }
+    const handleOpen = (): void => { onOpen?.() }
+    const handleMessage = (event: MessageEvent): void => {
+      let full: ServerRequest
+      let frame: F
+      try {
+        if (typeof event.data !== 'string') throw new Error('binary WebSocket frame')
+        full = serverRequestSchema.parse(JSON.parse(event.data))
+        frame = frameSchema.parse(full.payload)
+      } catch (error) {
+        console.error(`[client-connection] dropping malformed WebSocket frame on ${path}:`, error)
+        return
+      }
+      this.onEnvelope(full)
+      enqueue({ kind: 'frame', envelope: { rpcId: full.rpcId, payload: frame } })
+    }
+    const handleClose = (): void => { enqueue({ kind: 'end' }) }
+    const handleAbort = (): void => {
+      if (socket.readyState === WebSocket.CONNECTING || socket.readyState === WebSocket.OPEN) socket.close()
+    }
+    socket.addEventListener('open', handleOpen)
+    socket.addEventListener('message', handleMessage)
+    socket.addEventListener('close', handleClose, { once: true })
+    signal.addEventListener('abort', handleAbort, { once: true })
+    if (signal.aborted) handleAbort()
+    try {
+      while (true) {
+        while (inbox.length > 0) {
+          const item = inbox.shift() as SocketItem<F>
+          if (item.kind === 'end') return
+          yield item.envelope
+        }
+        await new Promise<void>((resolve) => { wake = resolve })
+      }
+    } finally {
+      signal.removeEventListener('abort', handleAbort)
+      socket.removeEventListener('open', handleOpen)
+      socket.removeEventListener('message', handleMessage)
+      socket.removeEventListener('close', handleClose)
+      handleAbort()
+    }
+  }
 }
--- a/packages/client/connection/src/http-bridge.ts
+++ b/packages/client/connection/src/http-bridge.ts
@@ -5,6 +5,16 @@

 import type { IncomingMessage, ServerResponse } from 'node:http'

+/** 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>
+}
+
 /**
 * Bridge one node:http request to the fetch-shaped handler (client close
 * aborts; SSE bodies stream out chunk by chunk).
@@ -16,8 +26,8 @@ import type { IncomingMessage, ServerResponse } from 'node:http'
 export async function bridge(
  req: IncomingMessage,
  res: ServerResponse,
-  apiHandler: { fetch: typeof fetch },
-  maxRequestBodyBytes: number,
+  apiHandler: FetchHandler,
+  maxRequestBodyBytes = 32 * 1024 * 1024,
 ): Promise<void> {
  const abort = new AbortController()
  // Client-disconnect detection MUST hang off the response, not the request:
@@ -30,10 +40,6 @@ export async function bridge(
  })
  const declaredLength = req.headers['content-length']
  if (declaredLength !== undefined && Number(declaredLength) > maxRequestBodyBytes) {
-    // Same discipline as the chunked-overrun path below: destroy, never
-    // drain. resume() would keep the socket open while the client trickles
-    // its declared length — an already-rejected request holding a server
-    // socket for as long as it likes.
    res.writeHead(413, { connection: 'close' })
    res.end()
    req.destroy()
--- a/packages/client/connection/src/index.ts
+++ b/packages/client/connection/src/index.ts
@@ -3,24 +3,50 @@ import type { Context } from 'cordis'
 import z from 'schemastery'
 import type {} from '@deepseek-ai/dsh-attachment'
 // Activates the httpServer Context merge used below.
-import type { WebRoute } from '@deepseek-ai/dsh-host-webserver'
+import type { WebRoute, WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver'
 import { toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
-import { API_PATH } from './api-path.ts'
+import { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'
 import { bridge } from './http-bridge.ts'
 import { assertTrustedAuthority, isTrustedApiRequest } from './api-request-trust.ts'
+import { HostConnectionService } from './rpc-host.ts'
+import { rejectWebSocketUpgrade, WebSocketDownlinks } from './websocket-downlink.ts'

-export { API_PATH } from './api-path.ts'
+export type {
+  ConnectionRpcAuthority,
+  ConnectionRpcEndpointMatcher,
+  ConnectionRpcHandler,
+  ConnectionRpcHandlerOptions,
+  HostConnectionHandle,
+  HostConnectionRpc,
+} from './rpc.ts'
+export { HostConnectionService } from './rpc-host.ts'
+
+export { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'

 /** Stable Cordis plugin name. */
 export const name = 'client-connection'

-/** Headroom for RPC JSON fields around the aggregate base64 image payload. */
+/** Headroom for RPC JSON fields around aggregate base64 image payloads. */
 const REQUEST_ENVELOPE_HEADROOM_BYTES = 1024 * 1024
-/** Independent default carrier cap; deployments may raise it for larger valid non-image RPCs. */
+
+function assertImageBodyCapacity(ctx: Context, maxRequestBodyBytes: number): void {
+  const attachments = ctx.get('attachments')
+  if (attachments === undefined) return
+  const requiredImageBodyBytes = Math.ceil(
+    attachments.imageLimits.maxMessageImageBytes * 4 / 3,
+  ) + REQUEST_ENVELOPE_HEADROOM_BYTES
+  if (maxRequestBodyBytes < requiredImageBodyBytes) {
+    throw new Error(
+      `client-connection maxRequestBodyBytes (${String(maxRequestBodyBytes)}) must be at least `
+      + `${String(requiredImageBodyBytes)} for the configured aggregate image limit`,
+    )
+  }
+}
+/** Default carrier cap for all HTTP RPC bodies. */
 const DEFAULT_MAX_REQUEST_BODY_BYTES = 32 * 1024 * 1024

-/** Services required before mounting the route. */
-export const inject = ['httpServer', 'apiProxy', 'attachments']
+/** 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. */
 export interface ConnectionConfig {
@@ -33,11 +59,7 @@ export interface ConnectionConfig {
   * that is not a bare, canonical authority fails the plugin load.
   */
  trustedHosts?: string[]
-  /**
-   * Maximum buffered JSON body for every `/api` request. This carrier policy
-   * is independent of image limits but must be large enough for the configured
-   * aggregate image bytes after base64 and envelope expansion.
-   */
+  /** Maximum buffered JSON body for every `/api` request. */
  maxRequestBodyBytes?: number
 }

@@ -56,21 +78,28 @@ export const Config: z<ConnectionConfig> = z.object({
 * reconnaissance no anonymous caller should have. `trustedHosts` is a
 * DNS-rebinding fence, explicitly not authentication, so the whole
 * configuration plane stays loopback-same-origin until a real authentication
- * layer exists. The model catalog (`llm.providers`, `llm.models`) is
- * deliberately NOT here: it carries provider ids, display names, and model
- * lists — no endpoints, keys, or key state — and a LAN client's model picker
- * legitimately needs it.
+ * layer exists. `llm.discoverModels` belongs to that plane on both counts: it
+ * carries a draft credential, and it makes the HOST issue a GET to a URL the
+ * caller chose and reports back the status or the parsed body — an anonymous
+ * LAN caller would have a probe for whatever the host can reach and the
+ * browser cannot.
+ *
+ * The model catalog (`llm.providers`, `llm.models`) is deliberately NOT here:
+ * it carries provider ids, display names, and model lists — no endpoints,
+ * keys, or key state — and a LAN client's model picker legitimately needs it.
 */
 const PRIVILEGED_METHODS = new Set([
  'host.pickDirectory',
  'host.openPath',
  'settings.describe',
+  'settings.openDocument',
  'settings.update',
  'settings.replace',
  'settings.mutate',
  'credentials.describe',
  'credentials.set',
  'credentials.unset',
+  'llm.discoverModels',
 ])

 /**
@@ -85,38 +114,67 @@ const PRIVILEGED_METHODS = new Set([
 export function apply(ctx: Context, config?: ConnectionConfig): void {
  // The Loader resolves schema defaults; hand-built test contexts may pass none.
  const trustedHosts = config?.trustedHosts ?? []
+  const maxRequestBodyBytes = config?.maxRequestBodyBytes ?? DEFAULT_MAX_REQUEST_BODY_BYTES
  // 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)
-  const apiHandler = toFetchHandler(ctx.apiProxy)
-  const requiredImageBodyBytes = Math.ceil(
-    ctx.attachments.imageLimits.maxMessageImageBytes * 4 / 3,
-  ) + REQUEST_ENVELOPE_HEADROOM_BYTES
-  const maxRequestBodyBytes = config?.maxRequestBodyBytes ?? DEFAULT_MAX_REQUEST_BODY_BYTES
-  if (maxRequestBodyBytes < requiredImageBodyBytes) {
-    throw new Error(
-      `client-connection maxRequestBodyBytes (${String(maxRequestBodyBytes)}) must be at least `
-      + `${String(requiredImageBodyBytes)} for the configured aggregate image limit`,
-    )
-  }
+  if (ctx.get('apiProxy') !== undefined) assertImageBodyCapacity(ctx, maxRequestBodyBytes)
+  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) => {
-      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) {
+      if (!isTrustedApiRequest(req, trustedHosts)) {
        res.writeHead(403)
        res.end('forbidden')
        return
      }
-      await bridge(req, res, apiHandler, maxRequestBodyBytes)
+      await bridge(req, res, fetchHandler, maxRequestBodyBytes)
    },
  }
  ctx.effect(() => ctx.httpServer.register(route), 'client-connection: /api route')
+  ctx.inject(['apiProxy'], (apiCtx) => {
+    assertImageBodyCapacity(apiCtx, maxRequestBodyBytes)
+    const downlinks = new WebSocketDownlinks(apiCtx.apiProxy)
+    const registerDownlink = (
+      path: string,
+      handle: WebUpgradeRoute['handler'],
+    ): void => {
+      apiCtx.effect(() => apiCtx.httpServer.registerUpgrade({
+        path,
+        handler: (req, socket, head) => {
+          if (!isTrustedApiRequest(req, trustedHosts)) {
+            rejectWebSocketUpgrade(socket)
+            return
+          }
+          return handle(req, socket, head)
+        },
+      }), `client-connection: ${path} WebSocket`)
+    }
+    apiCtx.effect(() => () => downlinks.close(), 'client-connection: WebSocket downlinks')
+    registerDownlink(MUX_EVENTS_PATH, (req, socket, head) => { downlinks.handleMux(req, socket, head) })
+    registerDownlink(HOST_EVENTS_PATH, (req, socket, head) => { downlinks.handleHost(req, socket, head) })
+  })
 }
--- a/packages/client/connection/src/loopback-hostname.ts
+++ b/packages/client/connection/src/loopback-hostname.ts
@@ -0,0 +1,18 @@
+/**
+ * Browser-safe, zero-dependency loopback classification shared by the `/api`
+ * Host fence and the package's `ctx.connection` state. The predicate stays
+ * package-internal; client plugins consume the derived state through Cordis.
+ */
+
+/**
+ * Whether a normalized URL hostname names the local loopback authority.
+ * @param hostname - WHATWG URL hostname (IPv6 literals retain brackets).
+ * @returns true for localhost, IPv6 loopback, or any IPv4 address in 127/8.
+ */
+export function isLoopbackHostname(hostname: string): boolean {
+  if (hostname === 'localhost' || hostname === '[::1]') return true
+  const parts = hostname.split('.')
+  return parts.length === 4
+    && parts[0] === '127'
+    && parts.every(part => /^\d{1,3}$/.test(part) && Number(part) <= 255)
+}
--- a/packages/client/connection/src/rpc-host.ts
+++ b/packages/client/connection/src/rpc-host.ts
@@ -0,0 +1,224 @@
+/** Host registry and HTTP adapter for generic Connection RPC channels. */
+
+import { Context, Service } from 'cordis'
+import type { WebRoute } from '@deepseek-ai/dsh-host-webserver'
+import {
+  clientRequestSchema,
+  RpcId,
+  type ClientRequest,
+  type RpcError,
+  type RpcErrorDetailsMap,
+  type RpcId as RpcIdType,
+  type ServerResponse as RpcServerResponse,
+} from '@deepseek-ai/dsh-host-apiproxy/api'
+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,
+  HostConnectionRpc,
+} from './rpc.ts'
+
+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.
+   * @param trustedHosts - deployment authorities accepted by trusted-host channels.
+   */
+  constructor(ctx: Context, private readonly trustedHosts: readonly string[]) {
+    super(ctx, 'connection')
+  }
+
+  /** Generic channel registry scoped to the Context reading this service. */
+  get rpc(): HostConnectionRpc {
+    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)
+      },
+    }
+  }
+
+  private register(
+    owner: Context,
+    channel: string,
+    handler: ConnectionRpcHandler,
+    options: ConnectionRpcHandlerOptions,
+  ): () => Promise<void> {
+    assertChannel(channel)
+    const trustedHosts = options.authority === 'loopback' ? [] : this.trustedHosts
+    const fetchHandler = rpcFetchHandler(channel, handler)
+    const route: WebRoute = {
+      kind: 'prefix',
+      path: channel,
+      handler: async (req, res) => {
+        if (!isTrustedApiRequest(req, trustedHosts)) {
+          res.writeHead(403)
+          res.end('forbidden')
+          return
+        }
+        await bridge(req, res, fetchHandler)
+      },
+    }
+    return owner.effect(
+      () => owner.httpServer.register(route),
+      `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,
+): FetchHandler {
+  return {
+    async fetch(request: Request): Promise<Response> {
+      const endpoint = endpointFromPath(channel, new URL(request.url).pathname)
+      if (request.method !== 'POST' || endpoint === undefined) {
+        return new Response('not found', { status: 404 })
+      }
+
+      const mediaType = request.headers.get('content-type')?.split(';', 1)[0]?.trim().toLowerCase()
+      if (mediaType !== 'application/json') {
+        return new Response('content type must be application/json', { status: 415 })
+      }
+
+      let body: unknown
+      try {
+        body = await request.json()
+      } catch {
+        return new Response('body is not JSON', { status: 400 })
+      }
+
+      const envelope = clientRequestSchema.safeParse(body)
+      if (!envelope.success) {
+        return invalidEnvelopeResponse(body, envelope.error.issues)
+      }
+      const message: ClientRequest = envelope.data
+      if (message.method !== endpoint) {
+        return errorResponse(message.rpcId, {
+          code: 'bad-request',
+          message: `method ${JSON.stringify(message.method)} does not match endpoint ${JSON.stringify(endpoint)}`,
+          details: { issues: [] },
+        })
+      }
+
+      try {
+        const result = await handler(endpoint, message.payload, request.signal)
+        return fullResponse(message.rpcId, result)
+      } catch (error) {
+        return new Response(`handler failure: ${String(error)}`, { status: 500 })
+      }
+    },
+  }
+}
+
+function invalidEnvelopeResponse(body: unknown, issues: RpcErrorDetailsMap['bad-request']['issues']): Response {
+  const rawId = (body as { rpcId?: unknown } | null)?.rpcId
+  const rpcId = typeof rawId === 'string' ? RpcId(rawId) : INVALID_REQUEST_RPC_ID
+  return errorResponse(rpcId, {
+    code: 'bad-request',
+    message: 'invalid client-request message',
+    details: { issues },
+  })
+}
+
+function endpointFromPath(channel: string, pathname: string): string | undefined {
+  if (!pathname.startsWith(`${channel}/`)) return undefined
+  const endpoint = pathname.slice(channel.length + 1)
+  const segments = endpoint.split('/')
+  if (segments.some(segment =>
+    segment === '' || segment === '.' || segment === '..' || !ENDPOINT_SEGMENT_PATTERN.test(segment))) {
+    return undefined
+  }
+  return endpoint
+}
+
+function errorResponse(rpcId: RpcIdType, error: RpcError): Response {
+  return fullResponse(rpcId, { ok: false, error })
+}
+
+function fullResponse(rpcId: RpcIdType, result: RpcServerResponse['result']): Response {
+  const body: RpcServerResponse = { type: 'server-response', rpcId, result }
+  return Response.json(body)
+}
+
+function assertChannel(channel: string): void {
+  if (!CHANNEL_PATTERN.test(channel) || channel === '/api') {
+    throw new Error(`connection: invalid or reserved RPC channel ${JSON.stringify(channel)}`)
+  }
+}
--- a/packages/client/connection/src/rpc.ts
+++ b/packages/client/connection/src/rpc.ts
@@ -0,0 +1,77 @@
+/** Generic unary RPC contracts shared by the Host and Client Connection halves. */
+
+import type { RpcResult } from '@deepseek-ai/dsh-host-apiproxy/api'
+
+/** Trust fence applied before a Host RPC channel reaches its handler. */
+export type ConnectionRpcAuthority = 'trusted-host' | 'loopback'
+
+/** Registration policy for one logical RPC channel. */
+export interface ConnectionRpcHandlerOptions {
+  /** Browser authority accepted by every endpoint in this channel. */
+  readonly authority: ConnectionRpcAuthority
+}
+
+/** Handler invoked after Connection has decoded the transport envelope. */
+export type ConnectionRpcHandler = (
+  endpoint: string,
+  payload: unknown,
+  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 `/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.
+   */
+  handle(
+    channel: string,
+    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. */
+export interface HostConnectionHandle {
+  /** Generic RPC channel registry. */
+  readonly rpc: HostConnectionRpc
+}
+
+/** Client caller for logical RPC channels carried by the current transport. */
+export interface ClientConnectionRpc {
+  /**
+   * Call one endpoint through an already registered logical channel.
+   * @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.
+   * @returns the existing RPC success/error result; correlation stays inside Connection.
+   */
+  call(
+    channel: string,
+    endpoint: string,
+    payload: unknown,
+    signal?: AbortSignal,
+  ): Promise<RpcResult<unknown>>
+}
--- a/packages/client/connection/src/websocket-downlink.ts
+++ b/packages/client/connection/src/websocket-downlink.ts
@@ -0,0 +1,153 @@
+/** Host-side WebSocket carrier for the two server-to-browser event streams. */
+
+import { randomUUID } from 'node:crypto'
+import type { IncomingMessage } from 'node:http'
+import type { Duplex } from 'node:stream'
+import WebSocket, { WebSocketServer } from 'ws'
+import type {
+  ApiProxy, HostFrame, MuxFrame, RpcRequest, ServerRequest,
+} from '@deepseek-ai/dsh-host-apiproxy/api'
+import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api'
+
+type Frame = MuxFrame | HostFrame
+
+function serverRequest(frame: RpcRequest<Frame>): ServerRequest {
+  return {
+    type: 'server-request',
+    rpcId: frame.rpcId,
+    method: frame.payload.type,
+    payload: frame.payload,
+  }
+}
+
+function send(socket: WebSocket, frame: RpcRequest<Frame>): Promise<void> {
+  return new Promise((resolve, reject) => {
+    if (socket.readyState !== WebSocket.OPEN) {
+      reject(new Error('websocket downlink closed before frame delivery'))
+      return
+    }
+    socket.send(JSON.stringify(serverRequest(frame)), (error) => {
+      if (error) reject(error)
+      else resolve()
+    })
+  })
+}
+
+function failureFrame(error: unknown): RpcRequest<Frame> {
+  return {
+    rpcId: RpcId(randomUUID()),
+    payload: {
+      type: 'stream/error',
+      error: { code: 'internal', message: String(error), details: {} },
+    },
+  }
+}
+
+/**
+ * Owns WebSocket negotiation and frame pumping for the connection plugin's
+ * two downlinks. Client messages are a protocol violation: upstream traffic
+ * remains on HTTP.
+ */
+export class WebSocketDownlinks {
+  private readonly server = new WebSocketServer({ noServer: true })
+  private readonly pumps = new Set<Promise<void>>()
+
+  /** @param api - host API supplying the typed event streams. */
+  constructor(private readonly api: ApiProxy) {}
+
+  /**
+   * Upgrade one socket and pump the mux stream until either side closes.
+   * @param req - HTTP upgrade request.
+   * @param socket - Raw socket transferred by the HTTP server.
+   * @param head - Bytes already read after the upgrade headers.
+   */
+  handleMux(req: IncomingMessage, socket: Duplex, head: Buffer): void {
+    this.upgrade(req, socket, head, signal => this.api.events.mux({
+      rpcId: RpcId(randomUUID()),
+      payload: {},
+    }, signal))
+  }
+
+  /**
+   * Upgrade one socket and pump the host stream until either side closes.
+   * @param req - HTTP upgrade request.
+   * @param socket - Raw socket transferred by the HTTP server.
+   * @param head - Bytes already read after the upgrade headers.
+   */
+  handleHost(req: IncomingMessage, socket: Duplex, head: Buffer): void {
+    this.upgrade(req, socket, head, signal => this.api.events.host({
+      rpcId: RpcId(randomUUID()),
+      payload: {},
+    }, signal))
+  }
+
+  /**
+   * Terminate owned sockets and await the no-server acceptor plus frame pumps.
+   * @returns A promise resolving after every socket and source iterator stops.
+   */
+  async close(): Promise<void> {
+    for (const socket of this.server.clients) socket.terminate()
+    await new Promise<void>((resolve, reject) => {
+      this.server.close((error) => {
+        if (error === undefined) resolve()
+        else reject(error)
+      })
+    })
+    await Promise.all(this.pumps)
+  }
+
+  private upgrade<F extends Frame>(
+    req: IncomingMessage,
+    socket: Duplex,
+    head: Buffer,
+    open: (signal: AbortSignal) => AsyncIterable<RpcRequest<F>>,
+  ): void {
+    this.server.handleUpgrade(req, socket, head, (websocket) => {
+      const abort = new AbortController()
+      websocket.once('close', () => { abort.abort() })
+      websocket.once('error', () => { abort.abort() })
+      websocket.once('message', () => {
+        websocket.close(1008, 'downlink only')
+      })
+      const pump = this.pump(websocket, open(abort.signal), abort)
+      this.pumps.add(pump)
+      void pump.then(() => { this.pumps.delete(pump) })
+    })
+  }
+
+  private async pump<F extends Frame>(
+    socket: WebSocket,
+    frames: AsyncIterable<RpcRequest<F>>,
+    abort: AbortController,
+  ): Promise<void> {
+    try {
+      for await (const frame of frames) await send(socket, frame)
+    } catch (error) {
+      if (!abort.signal.aborted) {
+        try {
+          await send(socket, failureFrame(error))
+        } catch {
+          // Socket loss won the race; no downstream remains to receive the failure frame.
+        }
+      }
+    } finally {
+      abort.abort()
+      if (socket.readyState === WebSocket.OPEN) socket.close()
+    }
+  }
+}
+
+/**
+ * Reject an untrusted upgrade before protocol negotiation.
+ * @param socket - Raw HTTP socket that remains owned by the caller.
+ */
+export function rejectWebSocketUpgrade(socket: Duplex): void {
+  socket.end([
+    'HTTP/1.1 403 Forbidden',
+    'Connection: close',
+    'Content-Type: text/plain; charset=utf-8',
+    'Content-Length: 9',
+    '',
+    'forbidden',
+  ].join('\r\n'))
+}