Merge remote-tracking branch 'origin/master' into xtr/react-loop-simplification

# Conflicts: # packages/client/ui-conversation/src/client/queue/QueueDock.tsx
2026-08-04 19:37:51 +08:00
parent 09e909b96b 6f10f9c01c
commit 674278cb17
108 changed files with 2136 additions and 496 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
+    }
+  })
+})