Merge remote-tracking branch 'origin/master' into worktree/open-settings-config-file

# Conflicts:
#	packages/client/connection/README.i18n.yaml
#	packages/client/connection/README.md
#	packages/client/connection/README.zh.md

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`

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')

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).

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()
+    }
+  }
 }

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'
@@ -77,6 +78,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,
@@ -93,8 +95,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) })
 }

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'))
+}