Merge origin/master into task/command-feedback-master

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
@@ -14,6 +14,7 @@
  */
 
 import type { IncomingHttpHeaders } from 'node:http'
+import { isLoopbackHostname } from './loopback-hostname.ts'
 
 /** The request facts the fence reads (structural subset of IncomingMessage). */
 interface ApiTrustRequest {
@@ -25,14 +26,6 @@ function header(headers: IncomingHttpHeaders, name: string): string | undefined
   return typeof value === 'string' ? value : undefined
 }
 
-function isLoopbackHostname(hostname: string): boolean {
-  if (hostname === 'localhost' || hostname === '[::1]') return true
-  const parts = hostname.split('.')
-  return parts.length === 4
-    && parts[0] === '127'
-    && parts.every(part => /^\d{1,3}$/.test(part) && Number(part) <= 255)
-}
-
 /** Normalized URL of a Host-header authority (hostname lowercased, default port stripped, IPv6 bracketed), or undefined when unparsable. */
 function parseAuthority(authority: string): URL | undefined {
   try {
@@ -104,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/api.ts

@@ -1,19 +1,22 @@
 // Central contract re-export point: every contract import inside
 // web-runtime goes through this single file.
-// Types are type-only imports from the apiproxy api/ layer (zero Node deps, browser-safe);
-// the only runtime values are the RpcId constructor and the AbstractApiClient seam.
+// Types and runtime protocol helpers/bounds come from the apiproxy api/ layer
+// (zero Node deps, browser-safe); AbstractApiClient is the client seam.
 // NEVER import the package root: it drags bootHost/cordis into the browser bundle.
 // The ./api and ./client subpath exports are the browser-safe channels added for this.
 
 export type {
-  ApiProxy, SessionsApi, SessionSummary, HostApi, EventsApi, MuxFrame, HostFrame,
+  ApiProxy, SessionsApi, SessionSearchItem, SessionSummary, HostApi, EventsApi, MuxFrame, HostFrame,
   ApprovalResponsePayload, QuestionResponsePayload, HistoryEntry, ToolEventView,
   DirectoryEntry, DirectoryListing,
   WorkspaceApi, WorkspaceId, WorkspaceView,
   CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
   ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
-  ModelReasoningEffort, ModelTarget, SessionModels,
+  ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
   GoalsApi, GoalRef,
+  SettingsApi, SettingsNamespaceView, SettingsPathOpView, SettingsSecretView,
+  CredentialsApi, CredentialView, ConfigurableProviderView, LlmApi,
+  SubagentsApi, SubagentAddress, SubagentCatalog, SubagentListEntry, SubagentPromptReceipt,
 } from '@deepseek-ai/dsh-host-apiproxy/api'
 export type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'
 export type {
@@ -23,10 +26,15 @@ export type {
 // transportError moved down to the apiproxy api layer (it belongs beside
 // RpcResult, its subject); re-exported here so connection consumers keep one
 // contract entry point.
-export { RpcId, transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
+export {
+  RpcId,
+  SESSION_SEARCH_RESULT_LIMIT,
+  transportError,
+} from '@deepseek-ai/dsh-host-apiproxy/api'
 export { AbstractApiClient } from '@deepseek-ai/dsh-host-apiproxy/client'
 export type { IApiClient } from '@deepseek-ai/dsh-host-apiproxy/client'
 export type { SessionId, SessionEvent } from '@deepseek-ai/dsh-session/types'
+export type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 export type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm/types'
 
 import type { RpcResponse, RpcResult } from '@deepseek-ai/dsh-host-apiproxy/api'

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/index.ts

@@ -8,22 +8,30 @@ import type { IApiClient } from './api.ts'
 import { ConnectionController, type ConnectionConfig, type ConnectionSinks, type ConnectionState } from './connection.ts'
 import { FixtureApiClient } from './fixture.ts'
 import { WebApiClient } from './web-api-client.ts'
+import { isLoopbackHostname } from '../loopback-hostname.ts'
 
 // ---- Contract re-exports (browser-safe apiproxy channels + core types) ----
 export type {
-  ApiProxy, SessionsApi, SessionSummary, HostApi, EventsApi, MuxFrame, HostFrame,
+  ApiProxy, SessionsApi, SessionSearchItem, SessionSummary, HostApi, EventsApi, MuxFrame, HostFrame,
   ApprovalResponsePayload, QuestionResponsePayload, HistoryEntry, ToolEventView,
   DirectoryEntry, DirectoryListing,
   ToolCallView, ToolResultView, WorkspaceApi, WorkspaceId, WorkspaceView,
   CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
   ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
-  ModelReasoningEffort, ModelTarget, SessionModels,
+  MessageId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels,
+  SubagentsApi, SubagentAddress, SubagentCatalog, SubagentListEntry, SubagentPromptReceipt,
   RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
   ClientRequest, ServerResponse, ServerRequest, ClientResponse, RpcMessage, RpcReceipt,
   IApiClient, SessionId, SessionEvent, ContentBlock, StreamChunk,
   GoalsApi, GoalRef,
+  SettingsApi, SettingsNamespaceView, SettingsPathOpView, SettingsSecretView,
+  CredentialsApi, CredentialView, ConfigurableProviderView, LlmApi,
+} from './api.ts'
+export {
+  RpcId,
+  AbstractApiClient,
+  transportError,
 } from './api.ts'
-export { RpcId, AbstractApiClient, transportError } from './api.ts'
 
 // Connection loop types are public through ConnectionHandle.start; the
 // controller remains package-internal.
@@ -41,6 +49,8 @@ export const inject: string[] = []
 export interface ConnectionHandle {
   /** Shared api client (fixture or real, decided at boot from the page URL). */
   readonly api: IApiClient
+  /** Whether the current page authority is loopback; non-browser contexts default to true. */
+  readonly isLoopback: boolean
   /**
    * Start the connect/pump/reconnect loop with the consumer's frame sinks.
    * One consumer owns the streams (the runtime object layer); a second call
@@ -57,11 +67,13 @@ export interface ConnectionHandle {
  * @param ctx - client cordis context.
  */
 export function apply(ctx: Context): void {
-  const fixture = typeof location !== 'undefined' && new URLSearchParams(location.search).has('fixture')
+  const pageLocation = typeof location === 'undefined' ? undefined : location
+  const fixture = pageLocation !== undefined && new URLSearchParams(pageLocation.search).has('fixture')
   const api: IApiClient = fixture ? new FixtureApiClient() : new WebApiClient()
   let started = false
   const handle: ConnectionHandle = {
     api,
+    isLoopback: pageLocation === undefined || isLoopbackHostname(pageLocation.hostname),
     start(sinks, config) {
       if (started) throw new Error('connection: the stream loop is already owned by another consumer')
       started = true

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'
@@ -33,10 +34,40 @@ export const Config: z<ConnectionConfig> = z.object({
   trustedHosts: z.array(String).default([]),
 })
 
+/**
+ * Methods gated to loopback even on a trusted-host deployment. Native dialogs
+ * act on the host machine; the settings and credential domains mutate the
+ * user's configuration and secret store, and READING them is equally
+ * privileged — `settings.describe` returns every exposed namespace's
+ * configuration and `credentials.describe` reports whether an arbitrary
+ * environment-variable name is configured and where from, which is
+ * reconnaissance no anonymous caller should have. `trustedHosts` is a
+ * DNS-rebinding fence, explicitly not authentication, so the whole
+ * configuration plane stays loopback-same-origin until a real authentication
+ * layer exists. The model catalog (`llm.providers`, `llm.models`) is
+ * deliberately NOT here: it carries provider ids, display names, and model
+ * lists — no endpoints, keys, or key state — and a LAN client's model picker
+ * legitimately needs it.
+ */
+const PRIVILEGED_METHODS = new Set([
+  'host.pickDirectory',
+  'host.openPath',
+  'settings.describe',
+  'settings.openDocument',
+  'settings.update',
+  'settings.replace',
+  'settings.mutate',
+  'credentials.describe',
+  'credentials.set',
+  'credentials.unset',
+])
+
 /**
  * Mounts the API gateway under the browser transport prefix. Every request on
  * the prefix passes the browser-trust fence first (DNS-rebinding and
- * cross-site defense — [api-request-trust](./api-request-trust.ts)).
+ * cross-site defense — [api-request-trust](./api-request-trust.ts));
+ * privileged methods additionally pass it with an empty trust list, which
+ * pins them to loopback.
  * @param ctx - Host plugin context.
  * @param config - resolved plugin config (schema defaults applied).
  */
@@ -47,17 +78,48 @@ 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,
     handler: async (req, res) => {
-      if (!isTrustedApiRequest(req, trustedHosts)) {
+      const pathname = new URL(req.url ?? '/', 'http://dsh.internal').pathname
+      const method = pathname.startsWith(`${API_PATH}/`)
+        ? pathname.slice(API_PATH.length + 1)
+        : undefined
+      const allowed = method !== undefined && PRIVILEGED_METHODS.has(method)
+        ? isTrustedApiRequest(req, [])
+        : isTrustedApiRequest(req, trustedHosts)
+      if (!allowed) {
         res.writeHead(403)
         res.end('forbidden')
         return
       }
+      if (req.method === 'GET' && (pathname === MUX_EVENTS_PATH || pathname === HOST_EVENTS_PATH)) {
+        res.writeHead(426, { connection: 'Upgrade', upgrade: 'websocket' })
+        res.end('upgrade required')
+        return
+      }
       await bridge(req, res, apiHandler)
     },
   }
   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/loopback-hostname.ts

@@ -0,0 +1,18 @@
+/**
+ * Browser-safe, zero-dependency loopback classification shared by the `/api`
+ * Host fence and the package's `ctx.connection` state. The predicate stays
+ * package-internal; client plugins consume the derived state through Cordis.
+ */
+
+/**
+ * Whether a normalized URL hostname names the local loopback authority.
+ * @param hostname - WHATWG URL hostname (IPv6 literals retain brackets).
+ * @returns true for localhost, IPv6 loopback, or any IPv4 address in 127/8.
+ */
+export function isLoopbackHostname(hostname: string): boolean {
+  if (hostname === 'localhost' || hostname === '[::1]') return true
+  const parts = hostname.split('.')
+  return parts.length === 4
+    && parts[0] === '127'
+    && parts.every(part => /^\d{1,3}$/.test(part) && Number(part) <= 255)
+}

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