Merge pull request #1527 from deepseek-harness/worktree-websocket

feat(web): move connection downlinks to WebSocket
2026-08-04 18:18:32 +08:00
parent 7f3ea406e1 b0b50be64f
commit 6f10f9c01c
44 changed files with 1114 additions and 108 deletions
--- a/packages/client/connection/README.i18n.yaml
+++ b/packages/client/connection/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/client/connection/README.md
-README.md: f537fee3273e3b5d2411197cf1a1a6e0d34af5f9
-README.zh.md: a29d2c00e7df3f6290a03ffdad59b70b43702aca
+README.md: faf093964a740092983e13bf88f2cccd853c3e36
+README.zh.md: b06ab245dedbde13957aa416be044ef107b2753c
--- a/packages/client/connection/README.md
+++ b/packages/client/connection/README.md
@@ -2,11 +2,15 @@

 English | [中文](README.zh.md)

-Wire consumer layer: the client plugin's apply mounts `ctx.connection` (shared api client + current-page loopback state + single-consumer stream-loop starter); the export face carries the wire contract types, the `AbstractApiClient` seam, and the loop's sink/config types. Loopback hostname classification stays package-internal: the `/api` Host fence uses it directly, while other client plugins consume the derived `ctx.connection.isLoopback` state. The node half's `/api` route pins the privileged method set (`host.pickDirectory`, `host.openPath`, and the whole configuration plane — `settings.describe`/`update`/`replace`/`mutate` and `credentials.describe`/`set`/`unset`, reads included, since describing returns the exposed configuration and probing an arbitrary reference reports where a credential comes from) to loopback by passing the trust fence with an empty trust list — a declared `trustedHosts` authority reaches every other method, while these stay loopback-local until a real authentication layer exists. The platform subclasses (WebApiClient/FixtureApiClient), the ConnectionController loop, and the fixture data source are package-internal — apply selects and drives them; tests reach them via src. Contract: api-contracts v3 §3.
+Wire consumer layer: the client plugin's apply mounts `ctx.connection` (shared api client + current-page loopback state + single-consumer stream-loop starter); the export face carries the wire contract types, the `AbstractApiClient` seam, and the loop's sink/config types. The real browser carrier uses HTTP POST for unary and respond operations and opens one downlink-only WebSocket each for `events.mux` and `events.host`; the fixture and in-process carriers continue to satisfy the same two-stream abstraction. Loopback hostname classification stays package-internal: the `/api` Host fence and WebSocket upgrades use it directly, while other client plugins consume the derived `ctx.connection.isLoopback` state. The node half's `/api` route pins the privileged method set (`host.pickDirectory`, `host.openPath`, and the whole configuration plane — `settings.describe`/`update`/`replace`/`mutate` and `credentials.describe`/`set`/`unset`, reads included, since describing returns the exposed configuration and probing an arbitrary reference reports where a credential comes from) to loopback by passing the trust fence with an empty trust list — a declared `trustedHosts` authority reaches every other method, while these stay loopback-local until a real authentication layer exists. The platform subclasses (WebApiClient/FixtureApiClient), the ConnectionController loop, and the fixture data source are package-internal — apply selects and drives them; tests reach them via src. The downlink boundary is documented in the [WebSocket downlink carrier Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md); the protocol contract is api-contracts v3 §3.

 ## /api browser-trust fence

-The node half guards every request under `/api` before bridging (`src/api-request-trust.ts`). Every request — browser-marked or not — must present a `Host` that is a loopback authority or matches a `trustedHosts` entry: exact on `host:port` entries, any port on port-less entries, both sides compared through WHATWG normalization (DNS-rebinding defense). There is deliberately no shortcut for requests without browser markers: over plain HTTP a browser attaches neither `Origin` nor Fetch-Metadata to reads (EventSource, images, navigations — those headers go only to trustworthy destinations), so an unmarked request may still be a rebound browser read with a readable response, and Host is the one header rebinding cannot forge; non-browser clients pass the same fence via loopback, the CLI-derived LAN IP literals, or a declared authority. When markers are present, an attached `Origin` must equal the Host authority, and an explicit `sec-fetch-site: cross-site` marker is refused. A `trustedHosts` entry that is not a bare, canonical `host[:port]` authority — one WHATWG parsing reads back exactly as written — fails the plugin load loudly: parsing would otherwise quietly authorize the hostname inside `harness.internal/path`, or broaden a dangling-colon or zero-padded port to an any-port grant. Failures answer plain 403 before any RPC dispatch. A non-loopback (`--host 0.0.0.0`) deployment therefore needs its serving authorities trusted: the dsh CLI derives the machine's LAN IP literals itself and its `--trusted-host` flag declares named ones, so `trustedHosts` in cordis.yml is for compositions the CLI does not boot. The fence is deliberately not an authentication layer — reachability policy stays with the webserver binding, and auth remains deferred work. Decision record: [the api browser-trust boundary Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md).
+The node half guards every entry under `/api` before bridging or upgrading (`src/api-request-trust.ts`). Every request — browser-marked or not — must present a `Host` that is a loopback authority or matches a `trustedHosts` entry: exact on `host:port` entries, any port on port-less entries, both sides compared through WHATWG normalization (DNS-rebinding defense). There is deliberately no shortcut for unmarked HTTP requests: over plain HTTP a browser attaches neither `Origin` nor Fetch-Metadata to image and navigation reads, so an unmarked request may still be a rebound browser read with a readable response, and Host is the one header rebinding cannot forge; a browser WebSocket handshake carries `Origin` and passes the same comparison. Non-browser clients pass the same fence via loopback, the CLI-derived LAN IP literals, or a declared authority. When markers are present, an attached `Origin` must equal the Host authority, and an explicit `sec-fetch-site: cross-site` marker is refused. A `trustedHosts` entry that is not a bare, canonical `host[:port]` authority — one WHATWG parsing reads back exactly as written — fails the plugin load loudly: parsing would otherwise quietly authorize the hostname inside `harness.internal/path`, or broaden a dangling-colon or zero-padded port to an any-port grant. HTTP failures answer plain 403 before any RPC dispatch; upgrade failures reject the handshake before any event stream starts. A non-loopback (`--host 0.0.0.0`) deployment therefore needs its serving authorities trusted: the dsh CLI derives the machine's LAN IP literals itself and its `--trusted-host` flag declares named ones, so `trustedHosts` in cordis.yml is for compositions the CLI does not boot. The fence is deliberately not an authentication layer — reachability policy stays with the webserver binding, and auth remains deferred work. Decision record: [the api browser-trust boundary Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md).
+
+## `/api` WebSocket downlinks
+
+`/api/events.mux` and `/api/events.host` each accept a WebSocket upgrade and send only the corresponding `ServerRequest` text messages to the browser; the client sends no application data over these sockets. If either socket ends, the current connection generation fails and rebuilds both streams; readiness still requires both sockets to be open and the `host.describe` HTTP call to succeed. Host teardown terminates both sockets, aborts their sources, and waits for source cleanup before returning. Ordinary network GETs to these paths return 426 with no SSE fallback; `toFetchHandler`'s SSE codec serves only the isomorphic in-process carrier.

 ## Keyless fixture

--- a/packages/client/connection/README.zh.md
+++ b/packages/client/connection/README.zh.md
@@ -2,11 +2,15 @@

 [English](README.md) | 中文

-协议消费层：客户端插件的 apply 会挂载 `ctx.connection`（共享 API 客户端 + 当前页面的 loopback 状态 + 单消费方流循环启动器）；导出表层携带协议契约类型、`AbstractApiClient` seam，以及循环的 sink／配置类型。Loopback hostname 判定逻辑留在包内部：`/api` Host fence 会直接使用它，其他客户端插件则消费派生的 `ctx.connection.isLoopback` 状态。node 半侧的 `/api` 路由让特权方法集（`host.pickDirectory`、`host.openPath`，以及整个配置面——`settings.describe`/`update`/`replace`/`mutate` 与 `credentials.describe`/`set`/`unset`，读取也在内，因为 describe 会返回已暴露的配置，而探测任意引用会报出某条凭据来自何处）以空信任表过信任 fence，从而钉在回环——已声明的 `trustedHosts` 授权可达其余全部方法，而这些方法在真正的认证层出现之前仍只限回环本机。平台子类（WebApiClient/FixtureApiClient）、ConnectionController 循环和 fixture 数据源都属于包内部：apply 负责选择并驱动它们，测试则通过 src 访问。契约：api-contracts v3 §3。
+协议消费层：客户端插件的 apply 会挂载 `ctx.connection`（共享 API 客户端 + 当前页面的 loopback 状态 + 单消费方流循环启动器）；导出表层携带协议契约类型、`AbstractApiClient` seam，以及循环的 sink／配置类型。真实浏览器载体以 HTTP POST 发送 unary／respond，并为 `events.mux` 与 `events.host` 各开一条只下行的 WebSocket；fixture 与进程内载体继续满足同一双流抽象。Loopback hostname 判定逻辑留在包内部：`/api` Host fence 与 WebSocket upgrade 会直接使用它，其他客户端插件则消费派生的 `ctx.connection.isLoopback` 状态。node 半侧的 `/api` 路由让特权方法集（`host.pickDirectory`、`host.openPath`，以及整个配置面——`settings.describe`/`update`/`replace`/`mutate` 与 `credentials.describe`/`set`/`unset`，读取也在内，因为 describe 会返回已暴露的配置，而探测任意引用会报出某条凭据来自何处）以空信任表过信任 fence，从而钉在回环——已声明的 `trustedHosts` 授权可达其余全部方法，而这些方法在真正的认证层出现之前仍只限回环本机。平台子类（WebApiClient/FixtureApiClient）、ConnectionController 循环和 fixture 数据源都属于包内部：apply 负责选择并驱动它们，测试则通过 src 访问。下行边界见 [WebSocket 下行载体 Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md)；协议契约见 api-contracts v3 §3。

 ## /api 浏览器信任栅栏

-node 半侧在桥接前守卫 `/api` 下的每个请求（`src/api-request-trust.ts`）。每个请求——无论是否带浏览器标记——`Host` 都必须是回环地址权威，或与某个 `trustedHosts` 条目匹配：带端口的 `host:port` 条目精确匹配，不带端口的条目匹配任意端口，两侧均经 WHATWG 归一化后比较（DNS rebinding 防御）。刻意不为无浏览器标记的请求开捷径：明文 HTTP 下浏览器的读取（EventSource、图片、导航——这些头只发给可信目标）既不带 `Origin` 也不带 Fetch-Metadata，因此无标记请求仍可能是被重绑页面发起的、响应可被读走的读取，而 Host 是重绑唯一伪造不了的请求头；非浏览器客户端经由回环地址、CLI 推导的 LAN IP 字面量或已声明的权威通过同一道栅栏。当标记存在时，`Origin` 必须与 Host 权威完全一致；显式的 `sec-fetch-site: cross-site` 标记一律拒绝。不是纯的、规范形 `host[:port]` 权威的 `trustedHosts` 条目——即 WHATWG 解析读回后与原文不完全一致的——会让插件加载大声失败：否则解析会悄悄授权 `harness.internal/path` 这类笔误里的 hostname，或把悬空冒号、补零端口放大成任意端口授权。失败在任何 RPC 分发之前以纯 403 应答。因此非回环（`--host 0.0.0.0`）部署需要让自己的服务权威被信任：dsh CLI 会自行推导本机的 LAN IP 字面量，其 `--trusted-host` flag 用于声明具名权威，所以 cordis.yml 中的 `trustedHosts` 面向 CLI 不参与引导的组合。这道栅栏刻意不承担认证职责——可达性策略归 webserver 绑定配置，认证仍是延期工作。决策记录：[api 浏览器信任边界 Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md)。
+node 半侧在桥接或 upgrade 前守卫 `/api` 下的每个入口（`src/api-request-trust.ts`）。每个请求——无论是否带浏览器标记——`Host` 都必须是回环地址权威，或与某个 `trustedHosts` 条目匹配：带端口的 `host:port` 条目精确匹配，不带端口的条目匹配任意端口，两侧均经 WHATWG 归一化后比较（DNS rebinding 防御）。刻意不为无浏览器标记的 HTTP 请求开捷径：明文 HTTP 下浏览器的图片与导航读取既不带 `Origin` 也不带 Fetch-Metadata，因此无标记请求仍可能是被重绑页面发起的、响应可被读走的读取，而 Host 是重绑唯一伪造不了的请求头；WebSocket 浏览器握手会带 `Origin` 并通过同一道比较。非浏览器客户端经由回环地址、CLI 推导的 LAN IP 字面量或已声明的权威通过同一道栅栏。当标记存在时，`Origin` 必须与 Host 权威完全一致；显式的 `sec-fetch-site: cross-site` 标记一律拒绝。不是纯的、规范形 `host[:port]` 权威的 `trustedHosts` 条目——即 WHATWG 解析读回后与原文不完全一致的——会让插件加载大声失败：否则解析会悄悄授权 `harness.internal/path` 这类笔误里的 hostname，或把悬空冒号、补零端口放大成任意端口授权。HTTP 失败在任何 RPC 分发之前以纯 403 应答，upgrade 失败在启动任何 event stream 前拒绝握手。因此非回环（`--host 0.0.0.0`）部署需要让自己的服务权威被信任：dsh CLI 会自行推导本机的 LAN IP 字面量，其 `--trusted-host` flag 用于声明具名权威，所以 cordis.yml 中的 `trustedHosts` 面向 CLI 不参与引导的组合。这道栅栏刻意不承担认证职责——可达性策略归 webserver 绑定配置，认证仍是延期工作。决策记录：[api 浏览器信任边界 Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md)。
+
+## `/api` WebSocket 下行
+
+`/api/events.mux` 与 `/api/events.host` 各接受一条 WebSocket upgrade，并只向浏览器发送对应的 `ServerRequest` text message；客户端不会在这些 socket 上发送业务数据。任一 socket 结束都会使当前 connection generation 失败并重建两条流，连接就绪仍要求两条 socket open 且 `host.describe` HTTP 调用成功。Host teardown 会终止两条 socket、中止各自的 source，并等待 source 清理完成后再返回。普通网络 GET 这些路径会返回 426，不保留 SSE 回退；`toFetchHandler` 的 SSE 编解码只服务进程内同构载体。

 ## 无密钥 fixture

--- a/packages/client/connection/package.json
+++ b/packages/client/connection/package.json
@@ -1,6 +1,6 @@
 {
  "name": "@deepseek-ai/dsh-client-connection",
-  "description": "Wire consumer layer: IApiClient subclasses, ConnectionController (SSE dual-stream + reconnect), fixture api (no cordis)",
+  "description": "Wire consumer layer: HTTP-up/WebSocket-down client, ConnectionController dual streams with reconnect, and fixture api",
  "version": "0.0.1",
  "private": true,
  "type": "module",
@@ -34,7 +34,8 @@
    "@deepseek-ai/dsh-llm": "workspace:^",
    "@deepseek-ai/dsh-session": "workspace:^",
    "@deepseek-ai/dsh-tools": "workspace:^",
-    "schemastery": "^3.18.0"
+    "schemastery": "^3.18.0",
+    "ws": "^8.21.0"
  },
  "files": [
    "lib/index.js",
@@ -52,6 +53,7 @@
  "devDependencies": {
    "@deepseek-ai/dsh-host-webserver": "workspace:^",
    "@deepseek-ai/dsh-invariants": "workspace:^",
+    "@types/ws": "^8.18.1",
    "cordis": "^4.0.0-rc.7"
  }
 }
--- 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
@@ -97,7 +97,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/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/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/index.ts
+++ b/packages/client/connection/src/index.ts
@@ -2,13 +2,14 @@
 import type { Context } from 'cordis'
 import z from 'schemastery'
 // 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 { rejectWebSocketUpgrade, WebSocketDownlinks } from './websocket-downlink.ts'

-export { API_PATH } from './api-path.ts'
+export { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'

 /** Stable Cordis plugin name. */
 export const name = 'client-connection'
@@ -76,6 +77,7 @@ export function apply(ctx: Context, config?: ConnectionConfig): void {
  // silently authorizing its hostname prefix at request time.
  for (const entry of trustedHosts) assertTrustedAuthority(entry)
  const apiHandler = toFetchHandler(ctx.apiProxy)
+  const downlinks = new WebSocketDownlinks(ctx.apiProxy)
  const route: WebRoute = {
    kind: 'prefix',
    path: API_PATH,
@@ -92,8 +94,31 @@ export function apply(ctx: Context, config?: ConnectionConfig): void {
        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)
    },
  }
  ctx.effect(() => ctx.httpServer.register(route), 'client-connection: /api route')
+  const registerDownlink = (
+    path: string,
+    handle: WebUpgradeRoute['handler'],
+  ): void => {
+    ctx.effect(() => ctx.httpServer.registerUpgrade({
+      path,
+      handler: (req, socket, head) => {
+        if (!isTrustedApiRequest(req, trustedHosts)) {
+          rejectWebSocketUpgrade(socket)
+          return
+        }
+        return handle(req, socket, head)
+      },
+    }), `client-connection: ${path} WebSocket`)
+  }
+  ctx.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/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'))
+}
--- a/packages/client/connection/tests/client-apply.spec.ts
+++ b/packages/client/connection/tests/client-apply.spec.ts
@@ -3,15 +3,55 @@
 * selection off the page URL, and the single-consumer stream-loop ownership.
 */
 import { Context } from 'cordis'
-import { afterEach, describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { apply, type ConnectionHandle } from '../src/client/index.ts'
+import type { RpcMessage } from '../src/client/api.ts'
+import { RpcId } from '../src/client/api.ts'
 import { FixtureApiClient } from '../src/client/fixture.ts'
 import { WebApiClient } from '../src/client/web-api-client.ts'

-type Win = { location?: { hostname: string; search: string } }
+type Win = { location?: { hostname: string; search: string; origin?: string } }
+type WebSocketGlobal = { WebSocket?: typeof WebSocket }
+
+const originalWebSocket = globalThis.WebSocket
+const sockets: FakeWebSocket[] = []
+
+class FakeWebSocket extends EventTarget {
+  static readonly CONNECTING = 0
+  static readonly OPEN = 1
+  static readonly CLOSING = 2
+  static readonly CLOSED = 3
+
+  readonly url: string
+  readyState = FakeWebSocket.CONNECTING
+
+  constructor(url: string | URL) {
+    super()
+    this.url = String(url)
+    sockets.push(this)
+    queueMicrotask(() => {
+      if (this.readyState !== FakeWebSocket.CONNECTING) return
+      this.readyState = FakeWebSocket.OPEN
+      this.dispatchEvent(new Event('open'))
+    })
+  }
+
+  close(): void {
+    if (this.readyState === FakeWebSocket.CLOSED) return
+    this.readyState = FakeWebSocket.CLOSED
+    this.dispatchEvent(new Event('close'))
+  }
+
+  receive(data: unknown): void {
+    this.dispatchEvent(new MessageEvent('message', { data }))
+  }
+}

 afterEach(() => {
  delete (globalThis as Win).location
+  sockets.length = 0
+  if (originalWebSocket === undefined) delete (globalThis as WebSocketGlobal).WebSocket
+  else globalThis.WebSocket = originalWebSocket
 })

 async function mount(): Promise<ConnectionHandle> {
@@ -53,7 +93,7 @@ describe('connection client apply', () => {
    loop.stop() // teardown must not throw; the fixture streams abort quietly
  })

-  it('WebApiClient carries requests over globalThis.fetch', async () => {
+  it('WebApiClient keeps unary calls and respond on globalThis.fetch', async () => {
    ;(globalThis as Win).location = { hostname: 'localhost', search: '' }
    const handle = await mount()
    const original = globalThis.fetch
@@ -65,9 +105,102 @@ describe('connection client apply', () => {
    try {
      // Schema rejection is fine — the transport hop is the assertion.
      await (handle.api as WebApiClient).host.describe({}).catch(() => undefined)
+      await handle.api.respond({
+        type: 'client-response',
+        rpcId: RpcId('response-over-http'),
+        result: { ok: true, value: {} },
+      }).catch(() => undefined)
    } finally {
      globalThis.fetch = original
    }
-    expect(seen.some(u => u.includes('/api/'))).toBe(true)
+    expect(seen.some(u => u.includes('/api/host.describe'))).toBe(true)
+    expect(seen.some(u => u.includes('/api/respond'))).toBe(true)
+  })
+
+  it('opens one WebSocket per downlink, parses frames, and aborts both without using fetch', async () => {
+    ;(globalThis as Win).location = {
+      hostname: 'localhost', search: '', origin: 'http://localhost:3080',
+    }
+    ;(globalThis as WebSocketGlobal).WebSocket = FakeWebSocket as unknown as typeof WebSocket
+    const fetch = vi.spyOn(globalThis, 'fetch')
+    const client = (await mount()).api as WebApiClient
+    const envelopes: RpcMessage[][] = []
+    client.subscribeEnvelopes((batch) => { envelopes.push([...batch]) })
+    const opened: string[] = []
+    const muxAbort = new AbortController()
+    const hostAbort = new AbortController()
+    const mux = client.events.mux({}, muxAbort.signal, () => { opened.push('mux') })[Symbol.asyncIterator]()
+    const host = client.events.host({}, hostAbort.signal, () => { opened.push('host') })[Symbol.asyncIterator]()
+    const muxFrame = mux.next()
+    const hostFrame = host.next()
+    await vi.waitFor(() => { expect(sockets).toHaveLength(2) })
+    expect(sockets.map(socket => socket.url)).toEqual([
+      'ws://localhost:3080/api/events.mux',
+      'ws://localhost:3080/api/events.host',
+    ])
+    await vi.waitFor(() => { expect(opened).toEqual(['mux', 'host']) })
+
+    const errors = vi.spyOn(console, 'error').mockImplementation(() => {})
+    sockets[0]!.receive(new Uint8Array([1, 2, 3]))
+    sockets[1]!.receive(JSON.stringify({ type: 'server-request', rpcId: 'bad', method: 'host/session-status', payload: {} }))
+    sockets[0]!.receive(JSON.stringify({
+      type: 'server-request',
+      rpcId: 'mux-browser',
+      method: 'session/subscribed',
+      payload: { type: 'session/subscribed', sessionId: 'session-browser', lastSeq: 8 },
+    }))
+    sockets[1]!.receive(JSON.stringify({
+      type: 'server-request',
+      rpcId: 'host-browser',
+      method: 'host/commands-changed',
+      payload: { type: 'host/commands-changed' },
+    }))
+    expect(await muxFrame).toMatchObject({
+      value: { rpcId: 'mux-browser', payload: { type: 'session/subscribed', lastSeq: 8 } },
+    })
+    expect(await hostFrame).toMatchObject({
+      value: { rpcId: 'host-browser', payload: { type: 'host/commands-changed' } },
+    })
+    expect(errors).toHaveBeenCalledTimes(2)
+    await vi.waitFor(() => { expect(envelopes.flat()).toHaveLength(2) })
+    expect(fetch).not.toHaveBeenCalled()
+
+    const muxEnd = mux.next()
+    const hostEnd = host.next()
+    muxAbort.abort()
+    hostAbort.abort()
+    await expect(muxEnd).resolves.toMatchObject({ done: true })
+    await expect(hostEnd).resolves.toMatchObject({ done: true })
+    expect(sockets.every(socket => socket.readyState === FakeWebSocket.CLOSED)).toBe(true)
+    errors.mockRestore()
+    fetch.mockRestore()
+  })
+
+  it('maps an HTTPS page origin to a secure WebSocket URL', async () => {
+    ;(globalThis as Win).location = {
+      hostname: 'harness.example', search: '', origin: 'https://harness.example',
+    }
+    ;(globalThis as WebSocketGlobal).WebSocket = FakeWebSocket as unknown as typeof WebSocket
+    const client = (await mount()).api
+    const abort = new AbortController()
+    const iterator = client.events.mux({}, abort.signal)[Symbol.asyncIterator]()
+    const pending = iterator.next()
+    await vi.waitFor(() => { expect(sockets[0]?.url).toBe('wss://harness.example/api/events.mux') })
+    abort.abort()
+    await expect(pending).resolves.toMatchObject({ done: true })
+  })
+
+  it('closes a WebSocket immediately when its signal was already aborted', async () => {
+    ;(globalThis as Win).location = {
+      hostname: 'localhost', search: '', origin: 'http://localhost:3080',
+    }
+    ;(globalThis as WebSocketGlobal).WebSocket = FakeWebSocket as unknown as typeof WebSocket
+    const client = (await mount()).api
+    const abort = new AbortController()
+    abort.abort()
+    const iterator = client.events.mux({}, abort.signal)[Symbol.asyncIterator]()
+    await expect(iterator.next()).resolves.toMatchObject({ done: true })
+    expect(sockets).toHaveLength(1)
+    expect(sockets[0]?.readyState).toBe(FakeWebSocket.CLOSED)
  })
 })
--- a/packages/client/connection/tests/node-half.spec.ts
+++ b/packages/client/connection/tests/node-half.spec.ts
@@ -1,22 +1,29 @@
 /** Node half: registers the /api prefix route bridging to the api gateway. */
-import { EventEmitter } from 'node:events'
+import { EventEmitter, once } from 'node:events'
 import { createServer, request as httpRequest } from 'node:http'
-import { Readable } from 'node:stream'
+import { PassThrough, Readable } from 'node:stream'
 import { Context } from 'cordis'
 import { describe, expect, it } from 'vitest'
 import type { AddressInfo } from 'node:net'
 import type { IncomingMessage, ServerResponse } from 'node:http'
 import type { ApiProxy } from '@deepseek-ai/dsh-host-apiproxy/api'
-import type { HttpServerService, WebRoute } from '@deepseek-ai/dsh-host-webserver'
-import { API_PATH, apply, inject } from '../src/index.ts'
+import type { HttpServerService, WebRoute, WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver'
+import { API_PATH, apply, HOST_EVENTS_PATH, inject, MUX_EVENTS_PATH } from '../src/index.ts'

-/** Structural httpServer fake: the plugin only touches register(). */
-function fakeHttpServer(routes: WebRoute[]): Pick<HttpServerService, 'register' | 'tapIndex' | 'port'> {
+/** Structural httpServer fake recording both route registries. */
+function fakeHttpServer(
+  routes: WebRoute[],
+  upgrades: WebUpgradeRoute[],
+): Pick<HttpServerService, 'register' | 'registerUpgrade' | 'tapIndex' | 'port'> {
  return {
    register(route) {
      routes.push(route)
      return () => { routes.splice(routes.indexOf(route), 1) }
    },
+    registerUpgrade(route) {
+      upgrades.push(route)
+      return () => { upgrades.splice(upgrades.indexOf(route), 1) }
+    },
    tapIndex: () => () => {},
    port: 0,
  }
@@ -45,33 +52,67 @@ function fakeResponse(): { response: ServerResponse; state: { status?: number; b
  return { response, state }
 }

-async function mounted(config?: { trustedHosts?: string[] }): Promise<{ routes: WebRoute[]; dispose: () => Promise<void> }> {
+async function mounted(config?: { trustedHosts?: string[] }): Promise<{
+  routes: WebRoute[]
+  upgrades: WebUpgradeRoute[]
+  dispose: () => Promise<void>
+}> {
  const ctx = new Context()
  const routes: WebRoute[] = []
-  ctx.provide('httpServer', fakeHttpServer(routes) as HttpServerService)
+  const upgrades: WebUpgradeRoute[] = []
+  ctx.provide('httpServer', fakeHttpServer(routes, upgrades) as HttpServerService)
  ctx.provide('apiProxy', {} as unknown as ApiProxy)
  const fiber = ctx.plugin({ inject: [...inject], apply }, config)
  await fiber.await()
-  return { routes, dispose: () => fiber.dispose() }
+  return { routes, upgrades, dispose: () => fiber.dispose() }
 }

 describe('connection node half', () => {
  it('fails the load on a trustedHosts entry that is not a bare authority', async () => {
    const routes: WebRoute[] = []
+    const upgrades: WebUpgradeRoute[] = []
    const ctx = new Context()
-    ctx.provide('httpServer', fakeHttpServer(routes) as HttpServerService)
+    ctx.provide('httpServer', fakeHttpServer(routes, upgrades) as HttpServerService)
    ctx.provide('apiProxy', {} as unknown as ApiProxy)
    const fiber = ctx.plugin({ inject: [...inject], apply }, { trustedHosts: ['harness.internal/path'] })
    await expect(fiber).rejects.toThrow(/not a bare host\[:port\] authority/)
    expect(routes).toHaveLength(0)
+    expect(upgrades).toHaveLength(0)
  })

-  it('registers the /api prefix route and removes it with the fiber', async () => {
-    const { routes, dispose } = await mounted()
+  it('registers one HTTP route plus one upgrade route per downlink and removes all three with the fiber', async () => {
+    const { routes, upgrades, dispose } = await mounted()
    expect(routes).toHaveLength(1)
    expect(routes[0]).toMatchObject({ kind: 'prefix', path: API_PATH })
+    expect(upgrades.map(route => route.path)).toEqual([MUX_EVENTS_PATH, HOST_EVENTS_PATH])
    await dispose()
    expect(routes).toHaveLength(0)
+    expect(upgrades).toHaveLength(0)
+  })
+
+  it('requires WebSocket upgrade for network GETs to either event path', async () => {
+    const { routes, dispose } = await mounted()
+    for (const path of [MUX_EVENTS_PATH, HOST_EVENTS_PATH]) {
+      const { response, state } = fakeResponse()
+      await routes[0]!.handler(fakeRequest({ host: '127.0.0.1:3080' }, path), response)
+      expect(state.status).toBe(426)
+      expect(state.body).toBe('upgrade required')
+    }
+    await dispose()
+  })
+
+  it('rejects an untrusted WebSocket upgrade before protocol negotiation', async () => {
+    const { upgrades, dispose } = await mounted()
+    const socket = new PassThrough()
+    const chunks: Buffer[] = []
+    socket.on('data', (chunk: Buffer) => { chunks.push(chunk) })
+    const ended = once(socket, 'end')
+    await upgrades[0]!.handler(fakeRequest({
+      host: 'harness.example', origin: 'http://harness.example', 'sec-fetch-site': 'same-origin',
+    }, MUX_EVENTS_PATH), socket, Buffer.alloc(0))
+    await ended
+    expect(Buffer.concat(chunks).toString()).toContain('HTTP/1.1 403 Forbidden')
+    await dispose()
  })

  it('refuses an untrusted Host on any /api path before the bridge runs', async () => {
--- a/packages/client/connection/tests/websocket-downlink.spec.ts
+++ b/packages/client/connection/tests/websocket-downlink.spec.ts
@@ -0,0 +1,308 @@
+import { once } from 'node:events'
+import { createServer } from 'node:http'
+import type { AddressInfo } from 'node:net'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import WebSocket 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'
+import { HOST_EVENTS_PATH, MUX_EVENTS_PATH } from '../src/api-path.ts'
+import { WebSocketDownlinks } from '../src/websocket-downlink.ts'
+
+type MuxSource = (signal: AbortSignal) => AsyncIterable<RpcRequest<MuxFrame>>
+type HostSource = (signal: AbortSignal) => AsyncIterable<RpcRequest<HostFrame>>
+
+const running: (() => Promise<void>)[] = []
+
+afterEach(async () => {
+  await Promise.all(running.splice(0).map(close => close()))
+})
+
+function untilAbort(signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return Promise.resolve()
+  return new Promise((resolve) => {
+    signal.addEventListener('abort', () => { resolve() }, { once: true })
+  })
+}
+
+async function * idle<F>(signal: AbortSignal): AsyncGenerator<RpcRequest<F>> {
+  await untilAbort(signal)
+}
+
+function api(mux: MuxSource, host: HostSource): ApiProxy {
+  return {
+    events: {
+      mux: (_request, signal) => mux(signal),
+      host: (_request, signal) => host(signal),
+    },
+  } as ApiProxy
+}
+
+async function serve(downlinks: WebSocketDownlinks): Promise<{
+  origin: string
+  close: () => Promise<void>
+}> {
+  const server = createServer()
+  server.on('upgrade', (request, socket, head) => {
+    const pathname = new URL(request.url ?? '/', 'http://dsh.internal').pathname
+    if (pathname === MUX_EVENTS_PATH) downlinks.handleMux(request, socket, head)
+    else if (pathname === HOST_EVENTS_PATH) downlinks.handleHost(request, socket, head)
+    else socket.destroy()
+  })
+  await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
+  const port = (server.address() as AddressInfo).port
+  return {
+    origin: `ws://127.0.0.1:${String(port)}`,
+    close: async () => {
+      await downlinks.close()
+      await new Promise<void>(resolve => server.close(() => { resolve() }))
+    },
+  }
+}
+
+function read(socket: WebSocket): Promise<ServerRequest> {
+  return once(socket, 'message').then(([data]) => JSON.parse(String(data)) as ServerRequest)
+}
+
+async function acceptedSocket(downlinks: WebSocketDownlinks): Promise<WebSocket> {
+  const server = (downlinks as unknown as { server: { clients: Set<WebSocket> } }).server
+  let accepted: WebSocket | undefined
+  await vi.waitFor(() => {
+    accepted = server.clients.values().next().value
+    expect(accepted).toBeDefined()
+  })
+  return accepted as WebSocket
+}
+
+describe('WebSocket downlinks', () => {
+  it('carries mux and host over independent downstream sockets and cancels each source on close', async () => {
+    let muxAborted = false
+    let hostAborted = false
+    const downlinks = new WebSocketDownlinks(api(
+      async function * (signal) {
+        try {
+          yield {
+            rpcId: RpcId('mux-1'),
+            payload: { type: 'session/subscribed', sessionId: 'session-1' as never, lastSeq: 4 },
+          }
+          await untilAbort(signal)
+        } finally {
+          muxAborted = true
+        }
+      },
+      async function * (signal) {
+        try {
+          yield { rpcId: RpcId('host-1'), payload: { type: 'host/commands-changed' } }
+          await untilAbort(signal)
+        } finally {
+          hostAborted = true
+        }
+      },
+    ))
+    const host = await serve(downlinks)
+    running.push(host.close)
+
+    const mux = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    const hostSocket = new WebSocket(`${host.origin}${HOST_EVENTS_PATH}`)
+    const muxFrame = read(mux)
+    const hostFrame = read(hostSocket)
+    expect(await muxFrame).toEqual({
+      type: 'server-request',
+      rpcId: 'mux-1',
+      method: 'session/subscribed',
+      payload: { type: 'session/subscribed', sessionId: 'session-1', lastSeq: 4 },
+    })
+    expect(await hostFrame).toEqual({
+      type: 'server-request',
+      rpcId: 'host-1',
+      method: 'host/commands-changed',
+      payload: { type: 'host/commands-changed' },
+    })
+
+    const muxClosed = once(mux, 'close')
+    const hostClosed = once(hostSocket, 'close')
+    mux.close()
+    hostSocket.close()
+    await Promise.all([muxClosed, hostClosed])
+    await vi.waitFor(() => {
+      expect(muxAborted).toBe(true)
+      expect(hostAborted).toBe(true)
+    })
+  })
+
+  it('rejects client messages because upstream remains HTTP', async () => {
+    let aborted = false
+    const downlinks = new WebSocketDownlinks(api(
+      async function * (signal) {
+        try {
+          await untilAbort(signal)
+        } finally {
+          aborted = true
+        }
+      },
+      idle,
+    ))
+    const host = await serve(downlinks)
+    running.push(host.close)
+    const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    await once(socket, 'open')
+    const closed = once(socket, 'close')
+    socket.send('upstream payload')
+    const [code, reason] = await closed as [number, Buffer]
+    expect(code).toBe(1008)
+    expect(String(reason)).toBe('downlink only')
+    await vi.waitFor(() => { expect(aborted).toBe(true) })
+  })
+
+  it('sends stream/error before closing when a source fails', async () => {
+    const downlinks = new WebSocketDownlinks(api(
+      async function * () {
+        throw new Error('mux source failed')
+      },
+      idle,
+    ))
+    const host = await serve(downlinks)
+    running.push(host.close)
+    const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    const failure = read(socket)
+    const closed = once(socket, 'close')
+    expect((await failure).payload).toEqual({
+      type: 'stream/error',
+      error: { code: 'internal', message: 'Error: mux source failed', details: {} },
+    })
+    await closed
+  })
+
+  it('aborts the source when an accepted socket reports a transport error', async () => {
+    let aborted = false
+    const downlinks = new WebSocketDownlinks(api(
+      async function * (signal) {
+        try {
+          await untilAbort(signal)
+        } finally {
+          aborted = true
+        }
+      },
+      idle,
+    ))
+    const host = await serve(downlinks)
+    running.push(host.close)
+    const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    await once(socket, 'open')
+    const accepted = await acceptedSocket(downlinks)
+    const closed = once(socket, 'close')
+    accepted.emit('error', new Error('transport failed'))
+    await closed
+    expect(aborted).toBe(true)
+  })
+
+  it('drops a source frame that races after the client has closed', async () => {
+    let release!: () => void
+    const gate = new Promise<void>((resolve) => { release = resolve })
+    let finish!: () => void
+    const finished = new Promise<void>((resolve) => { finish = resolve })
+    let sourceSignal: AbortSignal | undefined
+    const downlinks = new WebSocketDownlinks(api(
+      async function * (signal) {
+        sourceSignal = signal
+        try {
+          await gate
+          yield {
+            rpcId: RpcId('late'),
+            payload: { type: 'session/subscribed', sessionId: 'session-late' as never, lastSeq: 0 },
+          }
+        } finally {
+          finish()
+        }
+      },
+      idle,
+    ))
+    const host = await serve(downlinks)
+    running.push(host.close)
+    const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    await once(socket, 'open')
+    const closed = once(socket, 'close')
+    socket.close()
+    await closed
+    await vi.waitFor(() => { expect(sourceSignal?.aborted).toBe(true) })
+    release()
+    await finished
+  })
+
+  it('contains socket send callback failures and closes the downlink', async () => {
+    let release!: () => void
+    const gate = new Promise<void>((resolve) => { release = resolve })
+    const downlinks = new WebSocketDownlinks(api(
+      async function * () {
+        await gate
+        yield {
+          rpcId: RpcId('send-failure'),
+          payload: { type: 'session/subscribed', sessionId: 'session-send' as never, lastSeq: 0 },
+        }
+      },
+      idle,
+    ))
+    const host = await serve(downlinks)
+    running.push(host.close)
+    const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    await once(socket, 'open')
+    const accepted = await acceptedSocket(downlinks)
+    const send = vi.spyOn(accepted, 'send').mockImplementation(((
+      _data: unknown,
+      optionsOrCallback?: unknown,
+      callback?: (error?: Error) => void,
+    ) => {
+      const done = typeof optionsOrCallback === 'function'
+        ? optionsOrCallback as (error?: Error) => void
+        : callback
+      done?.(new Error('socket send failed'))
+    }) as WebSocket['send'])
+    const closed = once(socket, 'close')
+    release()
+    await closed
+    expect(send).toHaveBeenCalledTimes(2)
+    send.mockRestore()
+  })
+
+  it('rejects when its acceptor has already closed', async () => {
+    const downlinks = new WebSocketDownlinks(api(idle, idle))
+    await downlinks.close()
+    await expect(downlinks.close()).rejects.toThrow('The server is not running')
+  })
+
+  it('waits for source cleanup before teardown resolves', async () => {
+    let cleanupStarted!: () => void
+    const started = new Promise<void>((resolve) => { cleanupStarted = resolve })
+    let releaseCleanup!: () => void
+    const cleanupGate = new Promise<void>((resolve) => { releaseCleanup = resolve })
+    let cleaned = false
+    const downlinks = new WebSocketDownlinks(api(
+      async function * (signal) {
+        try {
+          await untilAbort(signal)
+        } finally {
+          cleanupStarted()
+          await cleanupGate
+          cleaned = true
+        }
+      },
+      idle,
+    ))
+    const host = await serve(downlinks)
+    const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
+    await once(socket, 'open')
+    let closed = false
+    const closing = host.close().then(() => { closed = true })
+    try {
+      await started
+      expect(closed).toBe(false)
+      releaseCleanup()
+      await closing
+      expect(cleaned).toBe(true)
+    } finally {
+      releaseCleanup()
+      await closing
+    }
+  })
+})
--- a/packages/cordis/tool-cordis/src/api-catalog.ts
+++ b/packages/cordis/tool-cordis/src/api-catalog.ts
@@ -388,6 +388,10 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
        signature: 'register(route: WebRoute): () => void',
        jsDoc: '/**\n * Register a named route. Duplicate (kind, path) throws — route patterns are\n * a composition-level contract, so a collision is a misconfiguration.\n * @param route - kind, path, and the owning handler.\n * @returns the disposer removing the route.\n */',
      },
+      {
+        signature: 'registerUpgrade(route: WebUpgradeRoute): () => void',
+        jsDoc: '/**\n * Register an exact-path HTTP upgrade route. Duplicate paths throw because\n * one socket can have only one protocol owner.\n * @param route - pathname and handler owning negotiation plus socket use.\n * @returns the disposer removing the route.\n */',
+      },
      {
        signature: 'tapIndex(transform: (html: string) => string): () => void',
        jsDoc: '/**\n * Register an index.html transform, applied to every index response in\n * registration order.\n * @param transform - pure html-to-html function.\n * @returns the disposer removing the transform.\n */',
@@ -3129,6 +3133,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
    name: 'WebSource',
    declaration: 'export interface WebSource {\n    url: string;\n    title?: string;\n    snippet?: string;\n    publishedAt?: string;\n}',
  },
+  {
+    name: 'WebUpgradeRoute',
+    declaration: 'export interface WebUpgradeRoute {\n    path: string;\n    handler: (req: IncomingMessage, socket: Duplex, head: Buffer) => void | Promise<void>;\n}',
+  },
  {
    name: 'WorkflowMeta',
    declaration: 'export interface WorkflowMeta {\n    name: string;\n    description: string;\n    whenToUse?: string;\n    phases?: WorkflowPhase[];\n}',
--- a/packages/host/apiproxy/src/api/events.ts
+++ b/packages/host/apiproxy/src/api/events.ts
@@ -1,5 +1,5 @@
 /**
- * events domain contract: signatures and frame unions for the two SSE
+ * events domain contract: signatures and frame unions for the two logical
 * streams. Four-quadrant: streams yield the narrow form `RpcRequest<Frame>` (server-request
 * view) — rpcId must be exposed to the business layer, because responses to answerable frames
 * (approval/question requested) echo it; for pure pushes it identifies that one push.
@@ -42,7 +42,7 @@ export interface QueuedInboxItem {
  message: Message
 }

-/** Streaming face of the contract: the two SSE stream openers (mux + host). */
+/** Streaming face of the contract: the two logical stream openers (mux + host). */
 export interface EventsApi {
  /**
   * All-session aggregated mux stream. On open, emits a subscribed control frame for every
--- a/packages/host/apiproxy/src/api/index.ts
+++ b/packages/host/apiproxy/src/api/index.ts
@@ -1,7 +1,7 @@
 /**
 * apiproxy contract-layer barrel. api/ has zero Node dependencies and is
- * importable from the browser; the TS interfaces are the authoritative contract, HTTP/SSE are
- * merely physical channels (four-quadrant message model).
+ * importable from the browser; the TS interfaces are the authoritative contract, while HTTP,
+ * WebSocket, and in-process SSE are merely physical channels (four-quadrant message model).
 */

 import type { SessionsApi } from './sessions.ts'
--- a/packages/host/apiproxy/src/api/rpc.ts
+++ b/packages/host/apiproxy/src/api/rpc.ts
@@ -1,7 +1,7 @@
 /**
- * Four-quadrant RPC message model. Channels and messages are
- * decoupled: HTTP is the client→server physical channel, SSE the server→client one; logical
- * messages are channel-independent, and the wire full form is a four-member discriminated union.
+ * Four-quadrant RPC message model. Channels and messages are decoupled: HTTP,
+ * WebSocket, and in-process SSE are physical carriers, while logical messages
+ * are channel-independent and form a four-member discriminated union.
 * api/ contract layer: zero Node dependencies, importable from the browser.
 */

@@ -147,7 +147,7 @@ export interface ServerResponse {
 }

 /**
- * Message initiated by the server (wire carrier: SSE frame). Answerable interactions
+ * Message initiated by the server (wire carrier: downstream stream frame). Answerable interactions
 * (approval/question requested — stable rpcId, reused on replay) and pure pushes
 * (session/event etc. — rpcId identifies that one push) share this shape; whether a
 * response is expected is determined statically by method (a strict dichotomy, no third kind).
--- a/packages/host/apiproxy/src/fetch/client.ts
+++ b/packages/host/apiproxy/src/fetch/client.ts
@@ -1,6 +1,6 @@
 /**
 * Client side of the fetch carrier. AbstractApiClient holds every protocol invariant: rpcId minting,
- * four-quadrant envelope wrap/unwrap, zod parsing, SSE frame decoding, and the payload-direct
+ * four-quadrant envelope wrap/unwrap, zod parsing, in-process SSE frame decoding, and the payload-direct
 * IApiClient domain methods (business code never mints). Platform differences ride two aspects:
 * abstract doFetch (transport) + overridable onEnvelope (tap). ApiProxy (the impl face) is untouched.
 */
@@ -69,8 +69,8 @@ import {
 * Bounded calls merge it with the instance timeout via AbortSignal.any; user-paced calls
 * carry only that external signal. In both cases the signal rides beside the request, never
 * on the wire, like the stream signatures.
- * Stream methods accept an optional onOpen callback: it fires once the SSE transport is
- * readable (response headers received, before any frame) — the "stream established" signal
+ * Stream methods accept an optional onOpen callback: it fires once the physical transport is
+ * readable (before any frame) — the "stream established" signal
 * connection controllers need for the readiness handshake. Generators are lazy, so the
 * underlying fetch (and therefore onOpen) only happens once iteration starts.
 * Relationship: ApiProxy is the narrow-form signature contract the impl side implements;
--- a/packages/host/apiproxy/src/index.ts
+++ b/packages/host/apiproxy/src/index.ts
@@ -5,7 +5,7 @@
 * platform subclasses on the client side), and the host-side implementation
 * (api-proxy.ts: createApiProxy + the ApiProxyService gateway plugin providing
 * `ctx.apiProxy`). Transport-agnostic by design: this package registers no
- * routes — carriers (HTTP today, IPC later) wrap `ctx.apiProxy` themselves.
+ * routes — physical carriers wrap `ctx.apiProxy` themselves.
 */

 import { resolve } from 'node:path'
--- a/packages/host/webserver/README.i18n.yaml
+++ b/packages/host/webserver/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/host/webserver/README.md
-README.md: c3c7b222683bc7731a6c21f2fffd325225099bab
-README.zh.md: 99c0560eb74dc8076772ba1deef3034000f5f0db
+README.md: 196f350d87c5322cd3e9cda6e40587d35acd08c4
+README.zh.md: 0ae0470eab0aae2f6b539404621c611d95827977
--- a/packages/host/webserver/README.md
+++ b/packages/host/webserver/README.md
@@ -2,17 +2,17 @@

 English | [中文](README.zh.md)

-Plain HTTP route-registration plugin (default-exported `HttpServerService`, config `{host, port, distIndex}`): a `node:http` server that listens on activation and provides `ctx.httpServer` — `register(route)` adds a named `exact`/`prefix` route (duplicate `(kind, path)` throws: route patterns are a composition-level contract, so a collision is a misconfiguration; the returned disposer removes the route), `tapIndex(transform)` adds an index.html transform applied in registration order, `port` reads the listening port (the OS-assigned value when `port` is 0), and `host` reads the configured bind host (composition-time facts other plugins adapt to, e.g. the directory-picker chooser). The match order is fixed — exact over the whole table, then longest prefix, then the static dist fallback with the locked semantics: traversal outside the dist root is 403, any miss falls back to `index.html` with HTTP 200 (SPA routing), unknown extensions ship as octet-stream, non-GET/HEAD is 405. Registration order carries no request-facing semantics.
+Web HTTP and upgrade-route registration plugin (default-exported `HttpServerService`, config `{host, port, distIndex}`): a `node:http` server that listens on activation and provides `ctx.httpServer`. `register(route)` adds a named `exact`/`prefix` HTTP route; `registerUpgrade(route)` adds an upgrade route for an exact pathname. A duplicate path within either table throws because route patterns are a composition-level contract and a collision is a misconfiguration; both methods return a disposer that removes the registration. `tapIndex(transform)` adds an index.html transform applied in registration order, `port` reads the listening port (the OS-assigned value when `port` is 0), and `host` reads the configured bind host (composition-time facts other plugins adapt to, e.g. the directory-picker chooser). HTTP match order is fixed: exact over the whole table, then longest prefix, then the static dist fallback with the locked semantics: traversal outside the dist root is 403, any miss falls back to `index.html` with HTTP 200 (SPA routing), unknown extensions ship as octet-stream, and non-GET/HEAD is 405. Upgrades match exactly and unmatched connections are closed; registration order carries no request-facing semantics.

-The package knows no harness concepts: the `/api` bridge is the connection plugin's route, plugin bundles and the HMR event stream are the modules/hmr plugins' routes. `host` accepts only `127.0.0.1` (default posture) and `0.0.0.0` (deliberate network exposure); `distIndex` is an assembly fact the composing app resolves and injects, never self-resolved (dist location is workspace knowledge of the app). Web (browser) shape only — Electron loads dist over `file://` and carries fetch over an IPC bridge, not this server. This package never prints; the URL line belongs to the shell.
+The package knows no harness concepts: the `/api` HTTP bridge and downlink WebSockets are routes owned by the connection plugin, while plugin bundles and the HMR event stream are routes owned by the modules/hmr plugins. The upgrade handler owns the protocol handshake and connection contents; the webserver only delivers the raw socket and request. `host` accepts only `127.0.0.1` (default posture) and `0.0.0.0` (deliberate network exposure); `distIndex` is an assembly fact the composing app resolves and injects, never self-resolved (dist location is workspace knowledge of the app). Web (browser) shape only — Electron loads dist over `file://` and carries fetch over an IPC bridge, not this server. This package never prints; the URL line belongs to the shell.

-A listen failure (EADDRINUSE…) throws out of activation and rejects Loader composition with the bind diagnostic; the failed candidate fiber is disposed. A request whose handling throws (a malformed %-escape hitting `decodeURIComponent`, a client dropping mid-body) is answered 400 — or the socket destroyed when headers are already out — and logged as a warning; it never exits the process. Disposal pairs `close()` with `closeAllConnections()` because held-open responses (SSE) never end on their own.
+A listen failure (EADDRINUSE…) throws out of activation and rejects Loader composition with the bind diagnostic; the failed candidate fiber is disposed. An HTTP request whose handling throws (a malformed %-escape hitting `decodeURIComponent`, a client dropping mid-body) is answered 400 — or the socket destroyed when headers are already out — and logged as a warning; it never exits the process. An upgrade-handler exception or upgraded-socket transport error is logged as a warning and destroys its socket. Disposal starts `close()` and `closeAllConnections()`, destroys every tracked upgraded socket, and returns only after the HTTP server and those sockets have closed.

 In development, the client-plugin registry synchronously captures each built bundle's stat baseline before it returns, then polls those baselines and re-hashes changed content. Each rescan stages its candidate table, graph, and watch map before publishing them, so a baseline failure preserves the prior graph. An immediate rebuild therefore cannot disappear into an asynchronously established watch baseline; a rename window marks the path dirty, retains the last successful baseline, and forces a re-hash when the bundle reappears even with identical metadata.

 ## Model Experience

-None, as the package is a pure HTTP carrier between the browser and the routes other plugins register; nothing here reaches a model request.
+None, as the package is a Web carrier between the browser and the HTTP/upgrade routes other plugins register; nothing here reaches a model request.

 #### KV Cache effect

--- a/packages/host/webserver/README.zh.md
+++ b/packages/host/webserver/README.zh.md
@@ -2,17 +2,17 @@

 [English](README.md) | 中文

-朴素的 HTTP 路由注册插件（默认导出 `HttpServerService`，配置为 `{host, port, distIndex}`）：一个在激活时开始监听的 `node:http` 服务器，提供 `ctx.httpServer`。`register(route)` 添加具名的 `exact`／`prefix` 路由；重复的 `(kind, path)` 会抛错，因为路由模式是组合层契约，冲突即配置错误；返回的 disposer 会移除该路由。`tapIndex(transform)` 添加按注册顺序应用的 index.html 转换，`port` 读取正在监听的端口（当 `port` 为 0 时读取 OS 分配的值），`host` 读取配置的绑定宿主（这些是其他插件据以自适应的组合期事实，例如 directory-picker 选择器）。匹配顺序固定不变：先在整张表中匹配精确路由，再匹配最长前缀，最后回退到静态 dist，并遵循固定语义：越出 dist 根目录的遍历返回 403，任何未命中项都以 HTTP 200 回退到 `index.html`（SPA 路由），未知扩展名按 octet-stream 提供，GET／HEAD 之外的方法返回 405。注册顺序不承载任何面向请求的语义。
+Web HTTP 与 upgrade route 注册插件（默认导出 `HttpServerService`，配置为 `{host, port, distIndex}`）：一个在激活时开始监听的 `node:http` 服务器，提供 `ctx.httpServer`。`register(route)` 添加具名的 `exact`／`prefix` HTTP route；`registerUpgrade(route)` 添加精确 pathname 的 upgrade route；同一张表内的重复路径会抛错，因为 route 模式是组合层契约，冲突即配置错误；两者返回的 disposer 都会移除注册。`tapIndex(transform)` 添加按注册顺序应用的 index.html 转换，`port` 读取正在监听的端口（当 `port` 为 0 时读取 OS 分配的值），`host` 读取配置的绑定宿主（这些是其他插件据以自适应的组合期事实，例如 directory-picker 选择器）。HTTP 匹配顺序固定不变：先在整张表中匹配精确 route，再匹配最长前缀，最后回退到静态 dist，并遵循固定语义：越出 dist 根目录的遍历返回 403，任何未命中项都以 HTTP 200 回退到 `index.html`（SPA 路由），未知扩展名按 octet-stream 提供，GET／HEAD 之外的方法返回 405。upgrade 只做精确匹配，未命中连接直接关闭；注册顺序不承载任何面向请求的语义。

-该包不了解任何 harness 概念：`/api` 桥接是 connection 插件的路由，插件 bundle 与 HMR（热模块替换）事件流则是 modules／hmr 插件的路由。`host` 只接受 `127.0.0.1`（默认姿态）和 `0.0.0.0`（有意向网络开放）；`distIndex` 是由组合应用解析并注入的组装事实，绝不会自行解析，因为 dist 位置属于应用的工作区知识。该服务器只服务 Web（浏览器）形态；Electron 通过 `file://` 加载 dist，并经 IPC 桥接承载 fetch，而不使用本服务器。该包从不打印内容；URL 行属于 shell。
+该包不了解任何 harness 概念：`/api` HTTP 桥接与下行 WebSocket 是 connection 插件的 route，插件 bundle 与 HMR（热模块替换）事件流则是 modules／hmr 插件的 route。upgrade handler 拥有协议握手与连接内容；webserver 只交付原始 socket 与 request。`host` 只接受 `127.0.0.1`（默认姿态）和 `0.0.0.0`（有意向网络开放）；`distIndex` 是由组合应用解析并注入的组装事实，绝不会自行解析，因为 dist 位置属于应用的工作区知识。该服务器只服务 Web（浏览器）形态；Electron 通过 `file://` 加载 dist，并经 IPC 桥接承载 fetch，而不使用本服务器。该包从不打印内容；URL 行属于 shell。

-监听失败（EADDRINUSE……）会从激活过程抛出，以 bind 诊断使 Loader 组合 reject；失败的候选 fiber 会被 dispose（资源释放）。处理请求时抛错（例如格式错误的百分号转义传入 `decodeURIComponent`，或客户端在请求体传输中途断开）时，服务器会响应 400；若响应头已经发出，则销毁 socket，并记录 warning，但绝不会退出进程。资源释放会把 `close()` 与 `closeAllConnections()` 配对，因为一直保持打开的 SSE（Server-Sent Events）响应不会自行结束。
+监听失败（EADDRINUSE……）会从激活过程抛出，以 bind 诊断使 Loader 组合 reject；失败的候选 fiber 会被 dispose（资源释放）。处理 HTTP 请求时抛错（例如格式错误的百分号转义传入 `decodeURIComponent`，或客户端在请求体传输中途断开）时，服务器会响应 400；若响应头已经发出，则销毁 socket，并记录 warning，但绝不会退出进程。upgrade handler 抛错或升级 socket 出现传输错误时，会记录 warning 并销毁对应 socket。资源释放会启动 `close()` 与 `closeAllConnections()`，销毁所有受跟踪的升级 socket，并仅在 HTTP server 与这些 socket 均已关闭后返回。

 在开发环境中，客户端插件注册表会在返回前同步捕获每个已构建 bundle 的 stat 基线，随后轮询这些基线，并在内容变化后重新计算哈希。每次重新扫描都会先暂存候选表、图和监听 map，再统一发布，因此基线失败会保留先前的图。这样，即时重建不会消失在异步建立的监听基线中；重命名窗口会把路径标记为脏，保留最近一次成功基线，并在 bundle 重新出现时强制重新计算哈希，即使其元数据完全相同也不例外。

 ## 模型体验

-无。该包只是浏览器与其他插件所注册路由之间的纯 HTTP 载体，其中没有任何内容会进入模型请求。
+无。该包只是浏览器与其他插件所注册 HTTP／upgrade route 之间的 Web 载体，其中没有任何内容会进入模型请求。

 #### KV 缓存影响

--- a/packages/host/webserver/package.json
+++ b/packages/host/webserver/package.json
@@ -1,6 +1,6 @@
 {
  "name": "@deepseek-ai/dsh-host-webserver",
-  "description": "Plain HTTP route-registration plugin: named-route registry (webServer service) + index transform taps + static dist fallback; knows no harness concepts",
+  "description": "Web route-registration plugin: HTTP and upgrade routes, index transform taps, and static dist fallback; knows no harness concepts",
  "version": "0.0.1",
  "private": true,
  "type": "module",
--- a/packages/host/webserver/src/index.ts
+++ b/packages/host/webserver/src/index.ts
@@ -1,10 +1,9 @@
 /**
- * @deepseek-ai/dsh-host-webserver — plain HTTP route-registration plugin: a
- * node:http server plus the `httpServer` service (named-route registry + index
- * transform taps + static dist fallback). Knows no harness concepts — every
- * feature surface (API bridge, plugin bundles, SSE) is a route some other
- * plugin registers. Web (browser) shape only — Electron loads dist over
- * file:// and carries fetch over an IPC bridge, not this server. This package
+ * @deepseek-ai/dsh-host-webserver — Web route-registration plugin: a node:http
+ * server plus the `httpServer` service (HTTP and upgrade route registries,
+ * index transform taps, and static dist fallback). Knows no harness concepts;
+ * feature plugins own every registered protocol. Web shape only — Electron
+ * loads dist over file:// and carries fetch over an IPC bridge. This package
 * never prints: the URL line belongs to the shell.
 */

@@ -12,6 +11,7 @@ import { createServer } from 'node:http'
 import type { IncomingMessage, ServerResponse, Server } from 'node:http'
 import { readFile } from 'node:fs/promises'
 import type { AddressInfo } from 'node:net'
+import type { Duplex } from 'node:stream'
 import { dirname } from 'node:path'
 import { Context, Service } from 'cordis'
 import z from 'schemastery'
@@ -35,6 +35,14 @@ export interface WebRoute {
  handler: (req: IncomingMessage, res: ServerResponse) => void | Promise<void>
 }

+/** One exact-path HTTP upgrade registration. */
+export interface WebUpgradeRoute {
+  /** Absolute pathname, no trailing slash. */
+  path: string
+  /** Owns protocol negotiation and the upgraded socket after dispatch. */
+  handler: (req: IncomingMessage, socket: Duplex, head: Buffer) => void | Promise<void>
+}
+
 /** Gateway config: listen address plus the static dist anchor (injected by the composing app, never self-resolved). */
 export interface Config {
  /** Listen host; the two supported values are loopback and all-interfaces. */
@@ -61,6 +69,8 @@ export class HttpServerService extends Service {

  private readonly exact = new Map<string, WebRoute>()
  private readonly prefixes = new Map<string, WebRoute>()
+  private readonly upgrades = new Map<string, WebUpgradeRoute>()
+  private readonly upgradedSockets = new Set<Duplex>()
  private readonly indexTaps: ((html: string) => string)[] = []
  private readonly distRoot: string
  private readonly distIndex: string
@@ -98,6 +108,20 @@ export class HttpServerService extends Service {
    return () => { table.delete(route.path) }
  }

+  /**
+   * Register an exact-path HTTP upgrade route. Duplicate paths throw because
+   * one socket can have only one protocol owner.
+   * @param route - pathname and handler owning negotiation plus socket use.
+   * @returns the disposer removing the route.
+   */
+  registerUpgrade(route: WebUpgradeRoute): () => void {
+    if (this.upgrades.has(route.path)) {
+      throw new Error(`webserver: duplicate upgrade route "${route.path}"`)
+    }
+    this.upgrades.set(route.path, route)
+    return () => { this.upgrades.delete(route.path) }
+  }
+
  /**
   * Register an index.html transform, applied to every index response in
   * registration order.
@@ -147,6 +171,40 @@ export class HttpServerService extends Service {
        res.end()
      })
    })
+    this.server.on('upgrade', (req, socket, head) => {
+      const onError = (error: Error): void => {
+        this.ctx.logger.warn(error)
+        socket.destroy()
+      }
+      socket.on('error', onError)
+      socket.once('close', () => {
+        socket.off('error', onError)
+        this.upgradedSockets.delete(socket)
+      })
+      let route: WebUpgradeRoute | undefined
+      try {
+        /* v8 ignore next -- node:http always sets url on server requests. */
+        route = this.upgrades.get(new URL(req.url ?? '/', 'http://x').pathname)
+      } catch (error) {
+        this.ctx.logger.warn(error instanceof Error ? error : new Error(String(error)))
+        socket.destroy()
+        return
+      }
+      if (route === undefined) {
+        socket.destroy()
+        return
+      }
+      this.upgradedSockets.add(socket)
+      try {
+        Promise.resolve(route.handler(req, socket, head)).catch((error: unknown) => {
+          this.ctx.logger.warn(error instanceof Error ? error : new Error(String(error)))
+          socket.destroy()
+        })
+      } catch (error) {
+        this.ctx.logger.warn(error instanceof Error ? error : new Error(String(error)))
+        socket.destroy()
+      }
+    })

    await new Promise<void>((resolve, reject) => {
      this.server.once('error', reject)
@@ -158,12 +216,19 @@ export class HttpServerService extends Service {
      })
    })

-    // close + closeAllConnections: held-open responses (SSE) never end on
-    // their own; without the force-close, close() would hang teardown.
-    this.ctx.effect(() => () => new Promise<void>((resolve) => {
-      this.server.close(() => { resolve() })
+    // Node does not include upgraded sockets in closeAllConnections(), so the
+    // service tracks and destroys them as part of the same ownership boundary.
+    this.ctx.effect(() => async () => {
+      const serverClosed = new Promise<void>((resolve) => {
+        this.server.close(() => { resolve() })
+      })
      this.server.closeAllConnections()
-    }), 'httpServer.listen')
+      const upgradedClosed = [...this.upgradedSockets].map(socket => new Promise<void>((resolve) => {
+        socket.once('close', () => { resolve() })
+        socket.destroy()
+      }))
+      await Promise.all([serverClosed, ...upgradedClosed])
+    }, 'httpServer.listen')
  }

  /** Longest-prefix-wins over the prefix table after an exact-table miss. */
--- a/packages/host/webserver/src/invariant.ts
+++ b/packages/host/webserver/src/invariant.ts
@@ -15,7 +15,7 @@ export const name = 'host-webserver-invariant'
 export const inject = ['invariants']

 /**
- * Owned relation: route registrations and their disposers must stay
+ * Owned relation: HTTP and upgrade route registrations and their disposers must stay
 * symmetric — after the owning fiber of a registered route unloads, the
 * route table must no longer answer for its path (a stale route would keep
 * serving a disposed plugin's handler). Checked on every fiber teardown
@@ -26,7 +26,10 @@ export const inject = ['invariants']
 const install: InvariantInstaller = (ctx, fail) => {
  ctx.on('internal/plugin', () => {
    const server = ctx.get('httpServer') as
-      | { register(route: { kind: 'exact'; path: string; handler: () => void }): () => void }
+      | {
+        register(route: { kind: 'exact'; path: string; handler: () => void }): () => void
+        registerUpgrade(route: { path: string; handler: () => void }): () => void
+      }
      | undefined
    if (server === undefined) return // no webserver row in this composition
    // Register/dispose probe on a reserved path: if dispose leaves the route
@@ -37,8 +40,11 @@ const install: InvariantInstaller = (ctx, fail) => {
    try {
      server.register(probe)()
      server.register(probe)()
+      const upgradeProbe = { path: '/__dsh_invariant_upgrade_probe__', handler: () => {} }
+      server.registerUpgrade(upgradeProbe)()
+      server.registerUpgrade(upgradeProbe)()
    } catch {
-      fail('httpServer.register() disposer left the route registered — route table and fiber lifecycles diverged')
+      fail('httpServer route disposer left a route registered — route tables and fiber lifecycles diverged')
    }
  }, { global: true })
 }
--- a/packages/host/webserver/tests/webserver.spec.ts
+++ b/packages/host/webserver/tests/webserver.spec.ts
@@ -7,6 +7,8 @@

 import { mkdtemp, rm, writeFile } from 'node:fs/promises'
 import { mkdir } from 'node:fs/promises'
+import { once } from 'node:events'
+import { connect } from 'node:net'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { pathToFileURL } from 'node:url'
@@ -72,6 +74,24 @@ async function request(port: number, path: string, init?: RequestInit): Promise<
  return { status: response.status, body: (await response.text()).slice(0, 80) }
 }

+/** Open one raw upgrade request and return after the handler writes its response. */
+async function upgrade(port: number, path: string): Promise<ReturnType<typeof connect>> {
+  const socket = connect(port, '127.0.0.1')
+  await once(socket, 'connect')
+  const response = once(socket, 'data')
+  socket.write([
+    `GET ${path} HTTP/1.1`,
+    `Host: 127.0.0.1:${String(port)}`,
+    'Connection: Upgrade',
+    'Upgrade: dsh-test',
+    '',
+    '',
+  ].join('\r\n'))
+  const [data] = await response as [Buffer]
+  expect(String(data)).toContain('101 Switching Protocols')
+  return socket
+}
+
 describe('real Loader composition', () => {
  // Real-Loader composition resolves workspace packages through tsx at test
  // time; first resolution after the host/client program split is slow enough
@@ -131,8 +151,51 @@ describe('real Loader composition', () => {
    expect((await request(port, '/once')).body).toContain('shell') // back to the SPA fallback
    expect(() => server.register({ kind: 'exact', path: '/once', handler: () => {} })).not.toThrow()

-    // Teardown: fiber dispose closes the socket and severs held connections.
+    // Upgrade routes match exact pathnames, reject duplicate ownership, and
+    // become registrable again after disposal. The accepted socket stays open
+    // so the teardown assertion also covers upgraded-connection ownership.
+    let upgradedServerClosed = false
+    const disposeUpgrade = server.registerUpgrade({
+      path: '/events',
+      handler: (_req, socket) => {
+        socket.once('close', () => { upgradedServerClosed = true })
+        socket.write('HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: dsh-test\r\n\r\n')
+      },
+    })
+    expect(() => server.registerUpgrade({ path: '/events', handler: () => {} }))
+      .toThrow(/duplicate upgrade route/)
+    const upgraded = await upgrade(port, '/events?stream=mux')
+    disposeUpgrade()
+    expect(() => server.registerUpgrade({ path: '/events', handler: () => {} })).not.toThrow()
+
+    // The webserver contains raw-socket errors even before an upgrade handler
+    // has installed its protocol implementation.
+    server.registerUpgrade({
+      path: '/upgrade-error',
+      handler: async (_req, socket) => {
+        await Promise.resolve()
+        socket.destroy(new Error('test upgrade transport failure'))
+      },
+    })
+    const failedUpgrade = connect(port, '127.0.0.1')
+    failedUpgrade.on('error', () => { /* The server-side reset is the fixture outcome. */ })
+    await once(failedUpgrade, 'connect')
+    const failedUpgradeClosed = once(failedUpgrade, 'close')
+    failedUpgrade.write([
+      'GET /upgrade-error HTTP/1.1',
+      `Host: 127.0.0.1:${String(port)}`,
+      'Connection: Upgrade',
+      'Upgrade: dsh-test',
+      '',
+      '',
+    ].join('\r\n'))
+    await failedUpgradeClosed
+    expect(await request(port, '/probe')).toMatchObject({ status: 200, body: 'EXACT' })
+
+    // Teardown closes both ordinary and upgraded sockets before it resolves.
    await loaded.fiber.dispose()
+    expect(upgradedServerClosed).toBe(true)
+    upgraded.destroy()
    await expect(request(port, '/probe')).rejects.toThrow()
  })