fix(web): gate Remote WebSockets on application readiness

apps/web/tests/fixtures/restart-startup.js

@@ -0,0 +1,30 @@
+/** Hold one Host service until the restart test releases application startup. */
+export const name = 'web-restart-startup'
+
+/** Keep the carrier services available while Session Controller startup is held. */
+export const inject = ['typertGateway', 'connection', 'webServer']
+
+/**
+ * Provide the test-only dependency after startup has been released.
+ * @param {import('@deepseek-ai/cordis').Context} ctx - Host plugin context.
+ * @returns {Promise<void>} Once the barrier is released or the plugin is disposed.
+ */
+export async function apply(ctx) {
+  if (process.env.DSH_WEB_RESTART_HOLD_STARTUP === '1') {
+    if (typeof process.send !== 'function') throw new Error('web restart startup fixture requires an IPC channel')
+    const released = Promise.withResolvers()
+    const resume = (message) => {
+      if (message === 'resume-startup') released.resolve(true)
+    }
+    ctx.effect(() => {
+      process.on('message', resume)
+      return () => {
+        process.off('message', resume)
+        released.resolve(false)
+      }
+    }, 'web-restart: startup barrier')
+    process.send('startup-blocked')
+    if (!await released.promise) return
+  }
+  ctx.reflect.provide('webRestartReady', {})
+}

apps/web/tests/fixtures/restart-startup.overlay.yml

@@ -0,0 +1,5 @@
+- id: session-controller
+  inject: [webRestartReady]
+- insert:
+    - id: web-restart-startup
+      name: ./restart-startup.js

apps/web/tests/server-restart.e2e.ts

@@ -14,16 +14,18 @@ class RestartableServer {
   private child: ChildProcess | undefined
   private closed: Promise<void> | undefined
   private output = ''
+  readonly startupBlocked = Promise.withResolvers<undefined>()
 
   constructor(private readonly world: string, private readonly modelUrl: string) {}
 
-  async start(port: number): Promise<string> {
+  async start(port: number, holdStartup = false): Promise<string> {
     if (this.child !== undefined) throw new Error('Server is already running')
     this.output = ''
     const ready = Promise.withResolvers<string>()
     const child = spawn(process.execPath, [
       join(REPO_ROOT, 'apps/cli/lib/bin.js'), '--profile', 'web',
       '--patch', fileURLToPath(new URL('./pin-browse-picker.overlay.yml', import.meta.url)),
+      '--patch', fileURLToPath(new URL('./fixtures/restart-startup.overlay.yml', import.meta.url)),
       '--no-open', '--port', String(port),
     ], {
       cwd: this.world,
@@ -32,8 +34,9 @@ class RestartableServer {
         DSH_HOME: join(this.world, 'home'), DSH_AGENTS_HOME: join(this.world, 'agents'),
         DSH_TELEMETRY_DISABLED: '1', DEEPSEEK_API_KEY: 'keyless-server-restart-fixture',
         DEEPSEEK_BASE_URL: this.modelUrl,
+        DSH_WEB_RESTART_HOLD_STARTUP: holdStartup ? '1' : '0',
       },
-      stdio: ['ignore', 'pipe', 'pipe'],
+      stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
     })
     this.child = child
     const receive = (data: Buffer): void => {
@@ -43,6 +46,9 @@ class RestartableServer {
     }
     child.stdout?.on('data', receive)
     child.stderr?.on('data', receive)
+    child.on('message', (message) => {
+      if (message === 'startup-blocked') this.startupBlocked.resolve(undefined)
+    })
     child.once('error', (error) => { ready.reject(error) })
     this.closed = new Promise((resolve) => {
       child.once('close', (code, signal) => {
@@ -58,6 +64,7 @@ class RestartableServer {
   async stop(): Promise<void> {
     const child = this.child
     if (child === undefined) return
+    this.resumeStartup()
     let forced = false
     const timer = setTimeout(() => { forced = true; child.kill('SIGKILL') }, 15_000)
     try {
@@ -70,10 +77,14 @@ class RestartableServer {
     if (forced) throw new Error('Server did not stop after SIGTERM\n' + this.logs())
   }
 
+  resumeStartup(): void {
+    if (this.child?.connected) this.child.send('resume-startup')
+  }
+
   logs(): string { return this.output.replace(/([?&]token=)[A-Za-z0-9_-]+/gu, '$1<redacted>') }
 }
 
-it('keeps an opened Session and draft when the server restarts on the same port', async () => {
+it.each([false, true])('keeps the same revision, Session and page across a server restart (delayed startup: %s)', async (holdStartup) => {
   const world = await mkdtemp(join(tmpdir(), 'dsh-server-restart-'))
   onTestFinished(() => rm(world, { recursive: true, force: true }))
   const model = createServer((request, response) => {
@@ -126,6 +137,7 @@ it('keeps an opened Session and draft when the server restarts on the same port'
   onTestFinished(() => browser.close())
   const page = await newEnglishPage(browser)
   const errors: string[] = []
+  const streamErrors: unknown[] = []
   const graphs: { rev: string; entries: { id: string; rev: string }[] }[] = []
   let readyFrames = 0
   let sessionBaselines = 0
@@ -140,7 +152,8 @@ it('keeps an opened Session and draft when the server restarts on the same port'
     })
     socket.on('framereceived', ({ payload }) => {
       if (typeof payload !== 'string') return
-      const message = JSON.parse(payload) as { type?: string; streamId: string; value?: { type?: string } }
+      const message = JSON.parse(payload) as { type?: string; streamId: string; value?: { type?: string }; error?: unknown }
+      if (message.type === 'error') streamErrors.push(message.error)
       if (message.type === 'item' && message.value?.type === 'ready') readyFrames++
       if (message.type === 'item' && endpoints.get(message.streamId) === 'session/follow'
         && message.value?.type === 'snapshot') sessionBaselines++
@@ -154,7 +167,7 @@ it('keeps an opened Session and draft when the server restarts on the same port'
   })
   onTestFailed(async () => {
     await saveFailureShot(page, 'web-server-restart-' + scenarioId)
-    await writeFile(join(REPO_ROOT, '.artifacts', 'server-restart-' + scenarioId + '.json'), JSON.stringify({ errors, graphs, server: server.logs() }, null, 2))
+    await writeFile(join(REPO_ROOT, '.artifacts', 'server-restart-' + scenarioId + '.json'), JSON.stringify({ errors, streamErrors, graphs, server: server.logs() }, null, 2))
   })
   await page.goto(url, { waitUntil: 'load' })
   await page.locator('[data-slot="root"]').waitFor({ state: 'attached', timeout: 20_000 })
@@ -177,7 +190,29 @@ it('keeps an opened Session and draft when the server restarts on the same port'
   page.on('framenavigated', () => { navigations++ })
 
   await server.stop()
-  const restarted = await server.start(port)
+  let restarted: string
+  const restarting = server.start(port, holdStartup)
+  try {
+    if (holdStartup) {
+      await Promise.race([
+        server.startupBlocked.promise,
+        restarting.then(() => { throw new Error('the restarted Host did not hold controller startup') }),
+      ])
+      const opened = await page.evaluate(() => new Promise<boolean>((resolve) => {
+        const endpoint = new URL('api/remote.mux', location.href)
+        endpoint.protocol = 'ws:'
+        const socket = new WebSocket(endpoint)
+        let accepted = false
+        socket.addEventListener('open', () => { accepted = true; socket.close() })
+        socket.addEventListener('close', () => { resolve(accepted) })
+      }))
+      expect(opened).toBe(false)
+      expect(readyFrames).toBe(beforeReady)
+    }
+  } finally {
+    server.resumeStartup()
+    restarted = await restarting
+  }
   expect(new URL(restarted).origin).toBe(new URL(url).origin)
   await expect.poll(() => readyFrames, { timeout: 30_000 }).toBeGreaterThan(beforeReady)
   await expect.poll(() => sessionBaselines, { timeout: 30_000 }).toBeGreaterThan(beforeBaseline)
@@ -187,7 +222,8 @@ it('keeps an opened Session and draft when the server restarts on the same port'
   }))
 
   expect(errors).toEqual([])
-  expect(graphs.at(-1)).toEqual(graphs[beforeGraphs - 1])
+  expect(streamErrors).toEqual([])
+  for (const graph of graphs.slice(beforeGraphs)) expect(graph).toEqual(graphs[beforeGraphs - 1])
   expect(await originalRoot!.evaluate(element => element.isConnected)).toBe(true)
   expect(await composer.textContent()).toBe(draft)
   expect(await response.isVisible()).toBe(true)
@@ -195,5 +231,6 @@ it('keeps an opened Session and draft when the server restarts on the same port'
   await composer.press('Enter')
   await expect.poll(() => response.count(), { timeout: 20_000 }).toBe(2)
   expect(errors).toEqual([])
+  expect(streamErrors).toEqual([])
   expect(navigations).toBe(0)
 })

docs/config-catalog.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 docs/config-catalog.md
-config-catalog.md: 563c8d21896e93a76394937b55f841036ee74c42
-config-catalog.zh.md: f3505a577660a534e6c5a3d255d4c46a4cfaea05
+config-catalog.md: b9e75f6679d43aef41808a30ed370e26070c1331
+config-catalog.zh.md: d76c88fb806998abee414347822a3d11c3b2b7de

docs/config-catalog.md

@@ -194,7 +194,7 @@ export interface Config {
 }
 ```
 
-Source: [`packages/api/gateway/src/index.ts:144`](../packages/api/gateway/src/index.ts)
+Source: [`packages/api/gateway/src/index.ts:145`](../packages/api/gateway/src/index.ts)
 
 <a id="deepseek-aidsh-api-job-controller"></a>
 

docs/config-catalog.zh.md

@@ -196,7 +196,7 @@ export interface Config {
 }
 ```
 
-来源：[`packages/api/gateway/src/index.ts:144`](../packages/api/gateway/src/index.ts)
+来源：[`packages/api/gateway/src/index.ts:145`](../packages/api/gateway/src/index.ts)
 
 <a id="deepseek-aidsh-api-job-controller"></a>
 

packages/api/gateway/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/api/gateway/README.md
-README.md: b5b64d3fa073c4f992529b63bc2e02117a28cecf
-README.zh.md: 87919703fdb2a1f9e32603bf6a1d152d4cc5404e
+README.md: 1a746a83eb966fba6ae40b61061d42bf478e1c84
+README.zh.md: 7b37548f6ad894b761457df80278eeed7aa6ca7b

packages/api/gateway/README.md

@@ -34,6 +34,8 @@ A cancellation-aware Remote method declares `signal: AbortSignal` as its final H
 
 A stream Remote uses `@Remote({ mode: 'stream' })` and returns an `Iterable` or `AsyncIterable`. `ctx.typertGateway.stream()` applies the same endpoint, argument, lookup, and cancellation checks as unary invocation, then returns a cancellation-aware iterable over the business items. The Client opens the Gateway-owned `/api/remote.mux` WebSocket when its plugin activates and keeps it connected while idle. Connection owns the retry schedule; before each retry it asks the mux to cancel any candidate or active socket and make exactly one fresh physical attempt. The Host sends Ping control frames at the configured `websocketHeartbeatIntervalMs` interval (two seconds by default), and the browser answers Pong at the WebSocket protocol layer, so idle network intermediaries see traffic without any Remote stream frame. A socket that has not answered the previous Ping is terminated at the next interval. Independently cancellable logical streams share that socket; an in-process Connection carrier provides equivalent streams directly without opening it.
 
+When the launcher supplies `ctx.appReady`, Gateway registers the WebSocket upgrade route only after successful application startup. Earlier connection attempts remain carrier failures handled by Connection's retry policy, so a restarting Host cannot accept streams while its controllers are still initializing. Unloading cancels a pending readiness subscription and closes any registered carrier. Embedded Hosts without `appReady` register immediately and own their startup ordering; in-process invocation is unchanged.
+
 Every stream also carries a Client-to-Host uplink on the same logical stream. A method declares the uplink item type as the second type argument of its return type, `RemoteStream<Out, In>` from [`dsh-typert-protocol`](../../typert/protocol/README.md); `In` defaults to `never`, and a method that returns `Iterable`, `AsyncIterable`, or `RemoteStream<Out>` declares that it reads none. The running method reads its call as `this.ctx.invocation`: the Gateway derives the receiver from a Context carrying that `RemoteInvocation` — `request`, `service`, `peer`, `signal`, and `uplink()` — so nothing enters the parameter list, and `ctx.invocation` is `undefined` on a Context no Remote call derived. `uplink()` is available once per call. With a descriptor codec it decodes every item before delivery; without one — an SRC method, or a method whose `In` is `never` — it delivers JSON-safe `unknown` values. The Client sends each item as an `item` frame on the open logical stream, a top-level `undefined` item as an `item` frame without `value`, and its half-close as an `end` frame; the method observes the half-close as the end of the `uplink()` iteration and cancellation through `signal`. A rejected item fails the whole stream with `gateway/input-invalid` (field `uplink`). Uplink items are validated one by one at the Host because they arrive from the browser; downlink items are values the Host method produced and pass through unchecked. When the method finishes its downlink the Gateway returns the uplink iterator and drops unread items; items sent to a method that never takes its uplink wait in the inbox until then. Each logical stream buffers at most `streamInboxBytes` (262144 by default) of uplink frame bytes; overflow fails the stream with `gateway/uplink-overflow`, and an item after `end` fails it with `gateway/protocol`, in both cases without closing the socket. `item`, `end`, and `cancel` frames for a stream id the Host has already finished are dropped; only a duplicate `open` closes the socket. Gateway-owned streams such as `$events` return their uplink as soon as they open, so their `item` frames are dropped instead of buffered. A unary method may also call `uplink()`; its items are readable only while the method runs. Every stream speaks for the operator Peer: a WebSocket's streams for the Peer Connection admitted at upgrade, whose scope disposal closes the socket, and an in-process carrier's for the operator directly.
 
 Host composition can register one application event source through `registerRemoteEvents()`. Gateway reserves the internal `$events` logical endpoint for that source, accepts only empty `args`, and aborts streams opened by the registration when the source is withdrawn. API Remotes owns the event selection, argument validation, per-Client queues, and the Host home sent in the opening `{ type: 'ready', clientId, host: { home } }` frame. Its source factory attaches incremental listeners synchronously, so the Client publishes the generation and starts baseline reads only after incremental delivery is ready.

packages/api/gateway/README.zh.md

@@ -34,6 +34,8 @@ Connection 可用时，Host 入口会在 Connection 共享的 `/api` FetchHandle
 
 流式 Remote 使用 `@Remote({ mode: 'stream' })` 并返回 `Iterable` 或 `AsyncIterable`。`ctx.typertGateway.stream()` 执行与一元调用相同的 endpoint、参数、lookup 和取消校验，再返回可取消的业务项 iterable。Client 插件激活时打开 Gateway 自有的 `/api/remote.mux` WebSocket，并让它在空闲时保持连接。Connection 拥有重试调度；每次 retry 前，它要求 mux 取消候选或活动 socket，并且只做一次全新的物理连接尝试。Host 按配置的 `websocketHeartbeatIntervalMs` 间隔（默认 2 秒）发送 Ping 控制帧，浏览器在 WebSocket 协议层自动回复 Pong，使空闲网络中间层持续看到流量，而不新增 Remote 流帧。若 socket 尚未回复上一次 Ping，Host 会在下一间隔终止它。可独立取消的逻辑流共享这条连接；进程内 Connection 载体直接提供等价的流，不打开该 WebSocket。
 
+启动器提供 `ctx.appReady` 时，Gateway 仅在应用成功启动后注册 WebSocket 升级路由。此前的连接尝试仍属于载体故障，由 Connection 的重试策略处理，因此重启中的 Host 不会在控制器仍在初始化时接受流。卸载会取消尚未触发的就绪订阅，并关闭已注册的载体。不提供 `appReady` 的嵌入式 Host 会立即注册，并自行负责启动顺序；进程内调用不变。
+
 每条流还在同一条逻辑流上携带从 Client 到 Host 的上行。方法把上行项类型声明为返回类型的第二个类型参数——来自 [`dsh-typert-protocol`](../../typert/protocol/README.zh.md) 的 `RemoteStream<Out, In>`；`In` 缺省为 `never`，返回 `Iterable`、`AsyncIterable` 或 `RemoteStream<Out>` 的方法即声明自己不读上行。运行中的方法通过 `this.ctx.invocation` 读取本次调用：Gateway 从携带该 `RemoteInvocation`（`request`、`service`、`peer`、`signal` 与 `uplink()`）的 Context 派生接收者，因此没有任何东西进入参数列表，而非 Remote 调用派生的 Context 上 `ctx.invocation` 为 `undefined`。`uplink()` 每次调用只能取一次。描述符带 codec 时它在交付前逐项解码；不带时——SRC 方法，或 `In` 为 `never` 的方法——它交付经 JSON 安全校验的 `unknown` 值。Client 在已打开的逻辑流上把每一项作为 `item` 帧发送（顶层 `undefined` 项是不带 `value` 的 `item` 帧），以 `end` 帧表示半关闭；方法通过 `uplink()` 迭代结束感知半关闭，通过 `signal` 感知取消。被拒绝的项使整条流以 `gateway/input-invalid`（字段 `uplink`）失败。上行项在 Host 逐项校验，因为它们来自浏览器；下行项是 Host 方法产出的值，不经校验原样透传。方法结束下行时 Gateway 调用 uplink 迭代器的 return 并丢弃未读的项；发给从未取用 uplink 的方法的项在 inbox 里等到那时为止。每条逻辑流最多缓冲 `streamInboxBytes`（默认 262144）字节的上行帧；超限使流以 `gateway/uplink-overflow` 失败，`end` 之后的项使流以 `gateway/protocol` 失败，两种情况都不关闭 socket。发给 Host 已结束的流 id 的 `item`、`end` 与 `cancel` 帧被丢弃；只有重复的 `open` 会关闭 socket。`$events` 这类 Gateway 自有的流在打开时立即 return 其 uplink，因此它们的 `item` 帧被丢弃而不缓冲。一元方法也可以调用 `uplink()`；其项只在方法运行期间可读。每条流都代表操作者 Peer：WebSocket 上的流代表 Connection 在升级时接纳的 Peer，该 scope 释放时 socket 关闭；进程内载体的流直接代表操作者。
 
 Host 组合可通过 `registerRemoteEvents()` 注册唯一的应用事件 source。Gateway 为它保留内部 `$events` logical endpoint，只接受空 `args`，并在 source 撤回时中止该注册打开的流。事件名单、参数校验、每个 Client 的队列及 opening `{ type: 'ready', clientId, host: { home } }` frame 中的 Host home 由 API Remotes 拥有。source factory 在返回 iterable 前同步挂好增量 listener，因此 Client 只在增量投递就绪后发布 generation 并开始 baseline 读取。

packages/api/gateway/package.json

@@ -67,6 +67,7 @@
   },
   "devDependencies": {
     "@deepseek-ai/dsh-brand": "workspace:^",
+    "@deepseek-ai/dsh-cmdline": "workspace:^",
     "@deepseek-ai/dsh-client-connection": "workspace:^",
     "@deepseek-ai/dsh-host-webserver": "workspace:^",
     "@deepseek-ai/dsh-remote-mock": "workspace:^",

packages/api/gateway/src/index.ts

@@ -15,6 +15,7 @@ import {
 import { Deque } from '@deepseek-ai/dsh-deque'
 import type { WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver'
 import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
+import type {} from '@deepseek-ai/dsh-cmdline'
 import z from '@deepseek-ai/schemastery'
 export type { TypertGatewayFaultDetails } from './remote-error-codes.ts'
 import {
@@ -217,6 +218,8 @@ export class TypertGatewayService extends Service implements TypertGateway {
 
   /**
    * Register the Gateway against the active Typert registry.
+   * WebSocket admission waits for launcher-owned application readiness when supplied;
+   * direct invocation and in-process streams remain available independently.
    * @param ctx - owning Host Context with Typert registry access.
    * @param config - validated Gateway transport configuration.
    */
@@ -234,31 +237,42 @@ export class TypertGatewayService extends Service implements TypertGateway {
       )
     })
     ctx.inject(['connection', 'webServer'], (webCtx) => {
-      const mux = new RemoteStreamMuxServer(
-        (endpoint, payload, uplink, peer, control) =>
-          this.openWireStream(endpoint, payload, uplink, peer, control.signal, control),
-        this.wireStream.failure,
-        resolved.websocketHeartbeatIntervalMs,
-        resolved.streamInboxBytes,
-      )
-      webCtx.effect(() => {
-        const route: WebUpgradeRoute = {
-          path: REMOTE_STREAM_MUX_PATH,
-          handler: (req, socket, head) => {
-            const admission = webCtx.connection.admit(req)
-            if ('rejection' in admission) {
-              rejectRemoteStreamUpgrade(socket, admission.rejection)
-              return
-            }
-            mux.handleUpgrade(req, socket, head, admission.peer)
-          },
+      const listen = (): void => {
+        const mux = new RemoteStreamMuxServer(
+          (endpoint, payload, uplink, peer, control) =>
+            this.openWireStream(endpoint, payload, uplink, peer, control.signal, control),
+          this.wireStream.failure,
+          resolved.websocketHeartbeatIntervalMs,
+          resolved.streamInboxBytes,
+        )
+        webCtx.effect(function* () {
+          yield () => mux.close()
+          const route: WebUpgradeRoute = {
+            path: REMOTE_STREAM_MUX_PATH,
+            handler: (req, socket, head) => {
+              const admission = webCtx.connection.admit(req)
+              if ('rejection' in admission) {
+                rejectRemoteStreamUpgrade(socket, admission.rejection)
+                return
+              }
+              mux.handleUpgrade(req, socket, head, admission.peer)
+            },
+          }
+          yield webCtx.webServer.registerUpgrade(route)
+        }, `api-gateway: ${REMOTE_STREAM_MUX_PATH} WebSocket`)
+      }
+      // Existing pages reconnect before the new Host prints its URL. No stream
+      // may enter until the launcher has activated and audited its controllers.
+      const ready = webCtx.get('appReady')
+      if (ready === undefined) listen()
+      else webCtx.effect(() => {
+        let closed = false
+        const cancel = ready.onReady(() => { if (!closed) listen() })
+        return () => {
+          closed = true
+          cancel()
         }
-        const unregister = webCtx.webServer.registerUpgrade(route)
-        return async () => {
-          unregister()
-          await mux.close()
-        }
-      }, `api-gateway: ${REMOTE_STREAM_MUX_PATH} WebSocket`)
+      }, 'api-gateway: application readiness')
     })
   }
 

packages/api/gateway/tests/gateway-stream.host.spec.ts

@@ -6,6 +6,7 @@ import { Context, symbols } from '@deepseek-ai/cordis'
 import { apply as applyConnection, inject as connectionInject } from '@deepseek-ai/dsh-client-connection'
 import WebServer from '@deepseek-ai/dsh-host-webserver'
 import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
+import type { AppReady } from '@deepseek-ai/dsh-cmdline'
 import {
   Remote,
   remoteErrorOf,
@@ -229,6 +230,30 @@ class FeedService extends TypertRemoteService {
 
 const roots: Context[] = []
 
+class StartupProbe implements AppReady {
+  private ready = false
+  private readonly listeners = new Set<() => void>()
+  lastListener: (() => void) | undefined
+
+  get pending(): number { return this.listeners.size }
+
+  onReady(listener: () => void): () => void {
+    this.lastListener = listener
+    if (this.ready) {
+      listener()
+      return () => {}
+    }
+    this.listeners.add(listener)
+    return () => { this.listeners.delete(listener) }
+  }
+
+  commit(): void {
+    this.ready = true
+    for (const listener of this.listeners) listener()
+    this.listeners.clear()
+  }
+}
+
 class RemoteEventSourceProbe {
   readonly source = (signal: AbortSignal): AsyncIterable<TypertRemoteEventDispatch> => {
     this.signal = signal
@@ -307,6 +332,30 @@ afterEach(async () => {
 })
 
 describe('Typert Remote streams', () => {
+  it.each([false, true])('accepts WebSockets only after application readiness (already ready: %s)', async (alreadyReady) => {
+    const startup = new StartupProbe()
+    if (alreadyReady) startup.commit()
+    const { ctx } = await setup(true, {}, startup)
+    if (!alreadyReady) {
+      expect(startup.pending).toBe(1)
+      expect(await acceptsSocket(ctx)).toBe(false)
+      startup.commit()
+    }
+    expect(await acceptsSocket(ctx)).toBe(true)
+    expect(startup.pending).toBe(0)
+  })
+
+  it('withdraws a pending WebSocket startup subscription when the Gateway unloads', async () => {
+    const startup = new StartupProbe()
+    const { ctx } = await setup(true, {}, startup)
+    expect(startup.pending).toBe(1)
+    await ctx.fiber.dispose()
+    expect(startup.pending).toBe(0)
+    // A launcher commit can already hold a copy of the cancelled listener.
+    expect(() => { startup.lastListener?.() }).not.toThrow()
+    expect(() => { startup.commit() }).not.toThrow()
+  })
+
   it('validates the WebSocket heartbeat timer range and the stream inbox bound', () => {
     expect(TypertGatewayService.Config({})).toEqual({ websocketHeartbeatIntervalMs: 2_000, streamInboxBytes: 262_144 })
     expect(TypertGatewayService.Config({ websocketHeartbeatIntervalMs: MAX_TIMER_DELAY_MS, streamInboxBytes: 1 }))
@@ -1341,9 +1390,11 @@ describe('Typert Remote streams', () => {
 async function setup(
   transport: boolean,
   gatewayConfig: GatewayConfig = {},
+  ready?: AppReady,
 ): Promise<{ readonly ctx: Context; readonly service: FeedService }> {
   const ctx = new Context()
   roots.push(ctx)
+  if (ready !== undefined) ctx.provide('appReady', ready)
   if (transport) {
     await ctx.plugin(WebServer, { host: '127.0.0.1', port: 0 })
     provideBrowserCredentials(ctx)
@@ -1365,6 +1416,17 @@ async function setup(
   return { ctx, service: receiver[symbols.original] ?? receiver }
 }
 
+async function acceptsSocket(ctx: Context): Promise<boolean> {
+  const socket = new WebSocket(`ws://127.0.0.1:${String(ctx.webServer.port)}/api/remote.mux`, {
+    headers: { cookie: browserCookie(ctx) },
+  })
+  const closed = new Promise<void>((resolve) => { socket.once('close', () => { resolve() }) })
+  const opened = await once(socket, 'open').then(() => true, () => false)
+  if (opened) socket.close()
+  await closed
+  return opened
+}
+
 function descriptors(): InvocationDescriptor[] {
   const label = {
     name: 'label',

packages/api/gateway/tsconfig.host.json

@@ -13,6 +13,9 @@
     "src/types.ts"
   ],
   "references": [
+    {
+      "path": "../../boot/cmdline"
+    },
     {
       "path": "../../../vendor/cosmokit"
     },

pnpm-lock.yaml

@@ -1034,6 +1034,9 @@ importers:
       '@deepseek-ai/dsh-client-connection':
         specifier: workspace:^
         version: link:../../client/connection
+      '@deepseek-ai/dsh-cmdline':
+        specifier: workspace:^
+        version: link:../../boot/cmdline
       '@deepseek-ai/dsh-host-webserver':
         specifier: workspace:^
         version: link:../../host/webserver