Merge remote-tracking branch 'origin/master' into feat/todo-multi-in-progress

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?: { 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> {
@@ -24,20 +64,28 @@ async function mount(): Promise<ConnectionHandle> {
 
 describe('connection client apply', () => {
   it('mounts ctx.connection with the real client when no ?fixture switch is present', async () => {
-    ;(globalThis as Win).location = { search: '' }
+    ;(globalThis as Win).location = { hostname: 'localhost', search: '' }
     const handle = await mount()
     expect(handle.api).toBeInstanceOf(WebApiClient)
+    expect(handle.isLoopback).toBe(true)
   })
 
   it('selects the fixture client under ?fixture (and with no location at all stays real)', async () => {
-    ;(globalThis as Win).location = { search: '?fixture' }
+    ;(globalThis as Win).location = { hostname: '127.0.0.1', search: '?fixture' }
     expect((await mount()).api).toBeInstanceOf(FixtureApiClient)
     delete (globalThis as Win).location
-    expect((await mount()).api).toBeInstanceOf(WebApiClient)
+    const handle = await mount()
+    expect(handle.api).toBeInstanceOf(WebApiClient)
+    expect(handle.isLoopback).toBe(true)
+  })
+
+  it('reports non-loopback page authority through the connection handle', async () => {
+    ;(globalThis as Win).location = { hostname: '192.0.2.20', search: '' }
+    expect((await mount()).isLoopback).toBe(false)
   })
 
   it('start() hands out one loop, rejects a second consumer, and stop() aborts the streams', async () => {
-    ;(globalThis as Win).location = { search: '?fixture' }
+    ;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' }
     const handle = await mount()
     // config omitted: the `config ?? {}` default arm is part of the surface.
     const loop = handle.start({})
@@ -45,8 +93,8 @@ describe('connection client apply', () => {
     loop.stop() // teardown must not throw; the fixture streams abort quietly
   })
 
-  it('WebApiClient carries requests over globalThis.fetch', async () => {
-    ;(globalThis as Win).location = { search: '' }
+  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
     const seen: string[] = []
@@ -57,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)
   })
 })

packages/client/connection/tests/fake-api.ts

@@ -4,7 +4,7 @@
 import type { CommandId } from '@deepseek-ai/dsh-commands/brand'
 import type {
   CommandDescriptor, HostFrame, IApiClient, ModelTarget, MuxFrame,
-  RpcRequest, RpcResponse, SessionId, SessionModels, SkillEntry,
+  RpcRequest, RpcResponse, SessionId, SessionModels, SessionSearchItem, SkillEntry,
 } from '../src/client/api.ts'
 import { RpcId } from '../src/client/api.ts'
 
@@ -44,17 +44,21 @@ export class FakeApiClient implements IApiClient {
 
   // Programmable slots (defaults answer OK-empty); reassign per case.
   onList: (payload: unknown) => Promise<RpcResponse<{ items: never[] }>> = () => Promise.resolve(ok({ items: [] }))
+  onSearch: (payload: unknown) => Promise<RpcResponse<{ items: SessionSearchItem[]; hasMore: boolean }>> =
+    () => Promise.resolve(ok({ items: [], hasMore: false }))
   onCreate: (payload: unknown) => Promise<RpcResponse<{ sessionId: SessionId }>> = () => Promise.resolve(ok({ sessionId: 'fk-new' as SessionId }))
+  onRename: (payload: unknown) => Promise<RpcResponse<{ title: string; seq: number }>> = () => Promise.resolve(ok({ title: 'fk-renamed', seq: 0 }))
+  onFork: (payload: unknown) => Promise<RpcResponse<{ sessionId: SessionId }>> = () => Promise.resolve(ok({ sessionId: 'fk-fork' as SessionId }))
   onHistory: (payload: { sessionId: SessionId; beforeSeq?: number; maxMessages?: number })
   => Promise<RpcResponse<{ events: never[]; hasMore: boolean; modelTarget: ModelTarget }>> =
     () => Promise.resolve(ok({
       events: [],
       hasMore: false,
-      modelTarget: { provider: 'deepseek', model: 'deepseek-chat' },
+      modelTarget: { provider: 'deepseek-official', model: 'deepseek-chat' },
     }))
 
   onModels: (payload: unknown) => Promise<RpcResponse<SessionModels>> = () => Promise.resolve(ok({
-    current: { provider: 'deepseek', model: 'deepseek-chat' },
+    current: { provider: 'deepseek-official', model: 'deepseek-chat' },
     groups: [],
     failures: [],
   }))
@@ -62,6 +66,7 @@ export class FakeApiClient implements IApiClient {
   => Promise<RpcResponse<{ selected: ModelTarget }>> =
     payload => Promise.resolve(ok({ selected: { provider: payload.provider, model: payload.model } }))
   onPrompt: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
+  onUpdateQueue: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
   onCancel: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
   onDescribe: (payload: unknown) => Promise<RpcResponse<{ version: string; cwd: string; attachedSessions: number }>> =
     () => Promise.resolve(ok({ version: '0-fake', cwd: '/f', attachedSessions: 0 }))
@@ -84,22 +89,44 @@ export class FakeApiClient implements IApiClient {
 
   private readonly muxConns: StreamConn<MuxFrame>[] = []
   private readonly hostConns: StreamConn<HostFrame>[] = []
+  lastSearchSignal: AbortSignal | undefined
 
   // Parameter annotations below are local structural types on purpose: the CI
   // lint lane runs without built artifacts, where IApiClient's wire types
   // (apiproxy subpath) resolve to any and inferred params trip no-unsafe-argument.
   readonly sessions: IApiClient['sessions'] = {
     list: (payload: unknown) => this.record('session.list', payload, this.onList(payload)),
+    search: (payload: unknown, signal?: AbortSignal) => {
+      this.lastSearchSignal = signal
+      return this.record('session.search', payload, this.onSearch(payload))
+    },
     create: (payload: unknown) => this.record('session.create', payload, this.onCreate(payload)),
     history: (payload: { sessionId: SessionId; beforeSeq?: number; maxMessages?: number }) =>
       this.record('session.history', payload, this.onHistory(payload)),
     models: (payload: unknown) => this.record('session.models', payload, this.onModels(payload)),
     selectModel: (payload: ModelTarget & { sessionId: SessionId }) =>
       this.record('session.selectModel', payload, this.onSelectModel(payload)),
+    rename: (payload: unknown) => this.record('session.rename', payload, this.onRename(payload)),
+    fork: (payload: unknown) => this.record('session.fork', payload, this.onFork(payload)),
     prompt: (payload: unknown) => this.record('session.prompt', payload, this.onPrompt(payload)),
+    updateQueue: (payload: unknown) => this.record('session.updateQueue', payload, this.onUpdateQueue(payload)),
     cancel: (payload: unknown) => this.record('session.cancel', payload, this.onCancel(payload)),
   }
 
+  readonly subagents: IApiClient['subagents'] = {
+    list: (payload: unknown) => this.record('subagent.list', payload, Promise.resolve(ok({
+      entries: [],
+      parentAvailable: true,
+    }))),
+    history: (payload: unknown) => this.record('subagent.history', payload, Promise.resolve(ok({
+      events: [],
+      hasMore: false,
+    }))),
+    prompt: (payload: unknown) => this.record('subagent.prompt', payload, Promise.resolve(ok({
+      messageId: 'fake-message' as never,
+    }))),
+  }
+
   readonly host: IApiClient['host'] = {
     describe: payload => this.record('host.describe', payload, this.onDescribe(payload)),
     pickDirectory: payload => this.record('host.pickDirectory', payload, this.onPickDirectory(payload)),
@@ -109,7 +136,7 @@ export class FakeApiClient implements IApiClient {
   }
 
   readonly workspace: IApiClient['workspace'] = {
-    list: (payload: unknown) => this.record('workspace.list', payload, Promise.resolve(ok({ items: [] }))),
+    list: (payload: unknown) => this.record('workspace.list', payload, Promise.resolve(ok({ items: [], archivedSessionIds: [] }))),
     create: (payload: unknown) => this.record('workspace.create', payload, Promise.resolve(ok({
       workspace: { workspaceId: 'fk-ws' as never, path: '/f/ws', title: 'ws', sessionIds: [], createdAt: '0', updatedAt: '0' },
       created: true,
@@ -121,6 +148,9 @@ export class FakeApiClient implements IApiClient {
     insertSessionBefore: (payload: unknown) => this.record('workspace.insertSessionBefore', payload, Promise.resolve(ok({
       workspace: { workspaceId: 'fk-ws' as never, path: '/f/ws', title: 'ws', sessionIds: [], createdAt: '0', updatedAt: '0' },
     }))),
+    archiveSession: (payload: unknown) => this.record('workspace.archiveSession', payload, Promise.resolve(ok({
+      archivedSessionIds: [(payload as { sessionId: SessionId }).sessionId],
+    }))),
   }
 
   // Payloads stay `unknown` (lint-lane note above); response rows are the real
@@ -150,6 +180,25 @@ export class FakeApiClient implements IApiClient {
     clear: payload => this.record('goal.clear', payload, Promise.resolve(ok({ cleared: true as const }))),
   }
 
+  readonly settings: IApiClient['settings'] = {
+    describe: payload => this.record('settings.describe', payload, Promise.resolve(ok({ writable: true, hasDocument: false, namespaces: [] }))),
+    openDocument: payload => this.record('settings.openDocument', payload, Promise.resolve(ok({ opened: true as const }))),
+    update: payload => this.record('settings.update', payload, Promise.resolve(ok({ ns: 'fake', schema: {}, value: {}, applies: 'live' as const, secrets: [], revision: 0 }))),
+    replace: payload => this.record('settings.replace', payload, Promise.resolve(ok({ ns: 'fake', schema: {}, value: {}, applies: 'live' as const, secrets: [], revision: 0 }))),
+    mutate: payload => this.record('settings.mutate', payload, Promise.resolve(ok({ ns: 'fake', schema: {}, value: {}, applies: 'live' as const, secrets: [], revision: 0 }))),
+  }
+
+  readonly credentials: IApiClient['credentials'] = {
+    describe: payload => this.record('credentials.describe', payload, Promise.resolve(ok({ credentials: {} }))),
+    set: payload => this.record('credentials.set', payload, Promise.resolve(ok({}))),
+    unset: payload => this.record('credentials.unset', payload, Promise.resolve(ok({}))),
+  }
+
+  readonly llm: IApiClient['llm'] = {
+    providers: payload => this.record('llm.providers', payload, Promise.resolve(ok({ providers: [] }))),
+    models: payload => this.record('llm.models', payload, Promise.resolve(ok({ groups: [], failures: [] }))),
+  }
+
   /** When true, streams never fire onOpen (misbehaving-carrier material for the handshake timeout guard). */
   suppressStreamOpen = false
 

packages/client/connection/tests/fixture.spec.ts

@@ -19,6 +19,20 @@ interface TimingHooks {
   failNextHistory(): void
   appendUser(id: string, msg: string): void
   appendTitle(id: string, title: string): void
+  startReasoningChunkStorm(id: string, chunkCount: number, chunksPerInterval: number, intervalMs: number): string
+  reasoningChunkStormState(): {
+    sessionId: string
+    chunkCount: number
+    chunksPerInterval: number
+    intervalMs: number
+    emitted: number
+    marker: string
+    emitting: boolean
+  } | null
+  beginModelRetry(id: string): void
+  scheduleModelRetry(id: string, retry?: number, delayMs?: number): void
+  cancelModelRetryDuringBackoff(id: string, delayMs?: number): void
+  completeModelRetry(id: string): void
   appendSilent(id: string, msg: string): void
   breakStreams(): void
 }
@@ -48,6 +62,59 @@ describe('createFixtureApi', () => {
     expect(response.result.value.items[1]?.parentSessionId).toBe('fx-alpha') // lineage material
   })
 
+  it('searches current message text with literal unicode61-style token phrases', async () => {
+    const api = createFixtureApi()
+    const signal = new AbortController().signal
+    const phrase = await api.sessions.search(req({ query: 'FIXTURE 历史消息' }), signal)
+    expect(phrase.result).toMatchObject({
+      ok: true,
+      value: {
+        items: [{ sessionId: 'fx-alpha' }],
+        hasMore: false,
+      },
+    })
+    if (!phrase.result.ok) throw new Error('search failed')
+    expect(phrase.result.value.items[0]?.snippet).toContain('fixture 历史消息')
+
+    timing().appendUser(
+      'fx-alpha',
+      `${'leading context '.repeat(20)}late café token${' trailing context'.repeat(20)}`,
+    )
+    const late = await api.sessions.search(req({ query: 'LATE CAFE TOKEN' }), signal)
+    if (!late.result.ok) throw new Error('late search failed')
+    const lateSnippet = late.result.value.items[0]?.snippet ?? ''
+    expect(lateSnippet).toContain('late café token')
+    expect(lateSnippet.startsWith('…')).toBe(true)
+    expect(lateSnippet.endsWith('…')).toBe(true)
+    expect(Array.from(lateSnippet).length).toBeLessThanOrEqual(120)
+
+    timing().appendUser('fx-alpha', 'Greek final sigma: ος')
+    const finalSigma = await api.sessions.search(req({ query: 'ΟΣ' }), signal)
+    if (!finalSigma.result.ok) throw new Error('final sigma search failed')
+    expect(finalSigma.result.value.items[0]?.snippet).toContain('ος')
+
+    const substring = await api.sessions.search(req({ query: 'ixtur' }), signal)
+    expect(substring.result).toEqual({
+      ok: true,
+      value: { items: [], hasMore: false },
+    })
+    const punctuationOnly = await api.sessions.search(req({ query: '*' }), signal)
+    expect(punctuationOnly.result).toEqual({
+      ok: true,
+      value: { items: [], hasMore: false },
+    })
+    const reasoningOnly = await api.sessions.search(req({ query: '思考过程' }), signal)
+    expect(reasoningOnly.result).toEqual({
+      ok: true,
+      value: { items: [], hasMore: false },
+    })
+
+    const aborted = new AbortController()
+    aborted.abort()
+    await expect(api.sessions.search(req({ query: 'fixture' }), aborted.signal))
+      .resolves.toMatchObject({ result: { ok: false, error: { code: 'cancelled' } } })
+  })
+
   it('pages history backwards on message-boundary cuts with seq-contiguous stitching', async () => {
     const api = createFixtureApi()
     const tail = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 10 }))
@@ -84,6 +151,14 @@ describe('createFixtureApi', () => {
         },
         plan: { active: false, pending: false },
         goal: null,
+        tokenUsage: {
+          uncachedInputTokens: 0,
+          outputTokens: 0,
+          cacheReadTokens: 0,
+          cacheWriteTokens: 0,
+        },
+        // No request ran, so neither pressure nor capacity is known yet.
+        contextPressure: {},
       } },
     })
   })
@@ -119,6 +194,36 @@ describe('createFixtureApi', () => {
     expect(JSON.stringify(after.result.value.events)).toContain('openai/gpt-5')
   })
 
+  it('serves configured DeepSeek readiness and keeps credential values write-only', async () => {
+    const api = createFixtureApi()
+    const settings = await api.settings.describe(req({}))
+    if (!settings.result.ok) throw new Error('settings describe failed')
+    expect(settings.result.value.namespaces).toMatchObject([{
+      ns: 'llm-deepseek',
+      value: { apiKeyEnv: 'DEEPSEEK_API_KEY' },
+      secrets: [{ path: ['apiKey'], set: false }],
+    }])
+
+    const initial = await api.credentials.describe(req({ refs: ['DEEPSEEK_API_KEY', 'TEST_API_KEY'] }))
+    if (!initial.result.ok) throw new Error('credential describe failed')
+    expect(initial.result.value.credentials).toEqual({
+      DEEPSEEK_API_KEY: { configured: true, source: 'file', writable: true },
+      TEST_API_KEY: { configured: false, writable: true },
+    })
+    await api.credentials.set(req({ ref: 'TEST_API_KEY', value: 'write-only-fixture-secret' }))
+    const configured = await api.credentials.describe(req({ refs: ['TEST_API_KEY'] }))
+    if (!configured.result.ok) throw new Error('credential describe failed')
+    expect(configured.result.value.credentials.TEST_API_KEY).toEqual({
+      configured: true,
+      source: 'file',
+      writable: true,
+    })
+    await api.credentials.unset(req({ ref: 'TEST_API_KEY' }))
+    const cleared = await api.credentials.describe(req({ refs: ['TEST_API_KEY'] }))
+    if (!cleared.result.ok) throw new Error('credential describe failed')
+    expect(cleared.result.value.credentials.TEST_API_KEY).toEqual({ configured: false, writable: true })
+  })
+
   it('emits the todo/write snapshot at the real tool boundary: between tool/call and tool/result, timestamps monotonic', async () => {
     const api = createFixtureApi()
     const tail = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 10 }))
@@ -192,6 +297,17 @@ describe('createFixtureApi', () => {
     expect(types).toContain('assistant/chunk')
     expect(types).toContain('assistant/message')
     expect(types.at(-1)).toBe('turn/end')
+    // Capacity is durable log state, not a transient frame: the prompt path
+    // records request/context and the projection carries it to the client.
+    expect(types).toContain('request/context')
+    expect(frames.some(frame =>
+      frame.type === 'session/projection'
+      && frame.key === 'tokenUsage'
+      && (frame.value as { outputTokens?: number }).outputTokens === 8)).toBe(true)
+    expect(frames.some(frame =>
+      frame.type === 'session/projection'
+      && frame.key === 'contextPressure'
+      && (frame.value as { contextWindow?: number }).contextWindow === 128_000)).toBe(true)
     const finalize = frames.find((f): f is Extract<MuxFrame, { type: 'session/event' }> => f.type === 'session/event' && f.event.type === 'assistant/message')
     expect(JSON.stringify(finalize?.event.data)).toContain('（已中断）')
     // Idle cancel: no replay in flight, must not explode; running flips false.
@@ -199,7 +315,7 @@ describe('createFixtureApi', () => {
     expect(idleCancel.result).toMatchObject({ ok: true })
   })
 
-  it('steer during a replay inserts a steering message and the replay continues to completion', async () => {
+  it('steer during a replay lands a user/message inside the current turn and the replay continues', async () => {
     const api = createFixtureApi()
     const created = await api.sessions.create(req({}))
     if (!created.result.ok) throw new Error('create failed')
@@ -212,7 +328,7 @@ describe('createFixtureApi', () => {
     await api.sessions.prompt(req({ sessionId: id, mode: 'steer' as const, content: [{ type: 'text' as const, text: '插话' }] }))
     const frames = await framesPromise
     const types = frames.filter((f): f is Extract<MuxFrame, { type: 'session/event' }> => f.type === 'session/event').map(f => f.event.type)
-    expect(types).toContain('steering/message')
+    expect(JSON.stringify(frames)).toContain('插话')
     expect(types.at(-1)).toBe('turn/end') // steer did not restart the turn
   })
 
@@ -223,7 +339,7 @@ describe('createFixtureApi', () => {
       const envelopes: RpcRequest<MuxFrame>[] = []
       for await (const envelope of api.events.mux(req({}), abort.signal)) {
         envelopes.push(envelope)
-        if (envelopes.length >= 8) abort.abort()
+        if (envelopes.length >= 10) abort.abort()
       }
       return envelopes
     }
@@ -231,16 +347,18 @@ describe('createFixtureApi', () => {
     const second = await openOnce()
     expect(first[0]?.payload).toMatchObject({ type: 'session/subscribed', sessionId: 'fx-alpha' })
     expect((first[0]?.payload as { lastSeq: number }).lastSeq).toBeGreaterThan(0)
-    // Projection baseline frames follow the subscribed frame (title + todos + permissions + plan + goal units).
+    // Projection baseline frames follow subscribed (domain units + token usage).
     expect(first[1]?.payload).toMatchObject({ type: 'session/projection', sessionId: 'fx-alpha', key: 'title', value: 'Fixture 历史会话' })
     expect(first[2]?.payload).toMatchObject({ type: 'session/projection', sessionId: 'fx-alpha', key: 'todos' })
     expect(first[3]?.payload).toMatchObject({ type: 'session/projection', sessionId: 'fx-alpha', key: 'permissions' })
     expect(first[4]?.payload).toMatchObject({ type: 'session/projection', sessionId: 'fx-alpha', key: 'plan', value: { active: false, pending: false } })
     expect(first[5]?.payload).toMatchObject({ type: 'session/projection', sessionId: 'fx-alpha', key: 'goal', value: null })
-    expect(first[6]?.payload).toMatchObject({ type: 'approval/requested', toolName: 'dangerous_tool' })
-    expect(second[6]?.rpcId).toBe(first[6]?.rpcId) // stable rpcId across replays (host replay semantics)
-    expect(first[7]?.payload).toMatchObject({ type: 'question/requested', sessionId: 'fx-alpha' })
-    expect(second[7]?.rpcId).toBe(first[7]?.rpcId)
+    expect(first[6]?.payload).toMatchObject({ type: 'session/projection', sessionId: 'fx-alpha', key: 'tokenUsage' })
+    expect(first[7]?.payload).toMatchObject({ type: 'session/projection', sessionId: 'fx-alpha', key: 'contextPressure' })
+    expect(first[8]?.payload).toMatchObject({ type: 'approval/requested', toolName: 'dangerous_tool' })
+    expect(second[8]?.rpcId).toBe(first[8]?.rpcId) // stable rpcId across replays (host replay semantics)
+    expect(first[9]?.payload).toMatchObject({ type: 'question/requested', sessionId: 'fx-alpha' })
+    expect(second[9]?.rpcId).toBe(first[9]?.rpcId)
   })
 
   it('steer with no replay in flight falls through to a fresh queued turn; non-text blocks stringify empty', async () => {
@@ -258,7 +376,7 @@ describe('createFixtureApi', () => {
     }))
     const frames = await framesPromise
     const types = frames.filter((f): f is Extract<MuxFrame, { type: 'session/event' }> => f.type === 'session/event').map(f => f.event.type)
-    expect(types[0]).toBe('turn/start') // idle steer degraded to a queued turn, not a steering insert
+    expect(types[0]).toBe('turn/start') // idle steer degraded to a queued turn, not an in-turn insert
   })
 
   it('gamma interval flip emits host/session-status and a running log-less session subscribes at lastSeq -1', async () => {
@@ -468,6 +586,47 @@ describe('createFixtureApi', () => {
     expect(seen.map(f => f.type)).toEqual(['host/workspace-changed', 'host/workspace-changed'])
   })
 
+  it('session.rename covers not-found, blank title, and the accepted append + title frame', async () => {
+    const api = createFixtureApi()
+    const abort = new AbortController()
+    const framesPromise = (async () => {
+      const frames: MuxFrame[] = []
+      for await (const envelope of api.events.mux(req({}), abort.signal)) {
+        frames.push(envelope.payload)
+        if (frames.some(f => f.type === 'session/projection' && f.key === 'title' && f.value === '重命名')) abort.abort()
+      }
+      return frames
+    })()
+    await new Promise(resolve => setTimeout(resolve, 10))
+
+    const missing = await api.sessions.rename(req({ sessionId: sid('fx-void'), title: 'x' }))
+    expect(missing.result).toMatchObject({ ok: false, error: { code: 'session-not-found', details: { sessionId: 'fx-void' } } })
+
+    const blank = await api.sessions.rename(req({ sessionId: sid('fx-alpha'), title: '   ' }))
+    expect(blank.result).toMatchObject({ ok: false, error: { code: 'title-invalid', details: { sessionId: 'fx-alpha' } } })
+
+    const renamed = await api.sessions.rename(req({ sessionId: sid('fx-alpha'), title: '  重命名  ' }))
+    if (!renamed.result.ok) throw new Error('rename failed')
+    expect(renamed.result.value.title).toBe('重命名')
+    const acceptedSeq = renamed.result.value.seq
+    // The response seq addresses the appended title event (the client plane
+    // has no session/title in its event union — titles ride the projection —
+    // so the event is located by seq and its payload checked structurally).
+    const history = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 100 }))
+    if (!history.result.ok) throw new Error('history failed')
+    const appended = history.result.value.events.find(entry => entry.event.seq === acceptedSeq)
+    expect(appended?.event).toMatchObject({
+      type: 'session/title',
+      data: { title: '重命名', messageSeqs: [], source: { kind: 'user' } },
+    })
+    // Beyond the subscribe-time baseline replay, the append emitted exactly
+    // one title projection frame carrying the new value at the response seq.
+    const frames = await framesPromise
+    const titleFrames = frames.filter(f => f.type === 'session/projection' && f.key === 'title' && f.sessionId === sid('fx-alpha') && f.value === '重命名')
+    expect(titleFrames).toHaveLength(1)
+    expect(titleFrames[0]).toMatchObject({ seq: acceptedSeq })
+  })
+
   it('workspace.insertSessionBefore moves, appends, no-ops, and rejects invalid ids', async () => {
     const api = createFixtureApi()
     const wsid = 'fx-ws-fixture' as WorkspaceId
@@ -694,8 +853,18 @@ describe('createFixtureApi', () => {
     hooks.appendSilent('fx-alpha', '静默丢帧')
     hooks.appendUser('fx-alpha', '正常直播')
     hooks.appendTitle('fx-alpha', 'Fixture 修订标题')
+    hooks.beginModelRetry('fx-alpha')
+    hooks.scheduleModelRetry('fx-alpha')
+    hooks.completeModelRetry('fx-alpha')
+    hooks.beginModelRetry('fx-alpha')
+    hooks.cancelModelRetryDuringBackoff('fx-alpha')
     await vi.waitFor(() => {
       expect(seen.some(f => f.type === 'session/event' && JSON.stringify(f.event.data).includes('正常直播'))).toBe(true)
+      expect(seen.some(f => f.type === 'session/event' && (f.event as { type: string }).type === 'llm/retry')).toBe(true)
+      expect(seen.some(f => f.type === 'session/event' && JSON.stringify(f.event.data).includes('重试后的完整回复'))).toBe(true)
+      expect(seen.some(f => f.type === 'session/event'
+        && f.event.type === 'turn/end'
+        && f.event.data.reason.kind === 'aborted')).toBe(true)
       expect(seen.some(f => f.type === 'session/projection' && f.key === 'title' && f.value === 'Fixture 修订标题')).toBe(true)
     })
     expect(seen.some(f => f.type === 'session/event' && JSON.stringify(f.event.data).includes('静默丢帧'))).toBe(false)
@@ -718,6 +887,50 @@ describe('createFixtureApi', () => {
     expect(abort.signal.aborted).toBe(false)
     expect(habort.signal.aborted).toBe(false)
   })
+
+  it('paces the opt-in reasoning stress hook from an external interval', async () => {
+    vi.useFakeTimers()
+    vi.setSystemTime(0)
+    const api = createFixtureApi()
+    const hooks = timing()
+    expect(hooks.reasoningChunkStormState()).toBeNull()
+    expect(() => hooks.startReasoningChunkStorm('fx-alpha', 0, 1, 16)).toThrow(/chunk count/)
+    expect(() => hooks.startReasoningChunkStorm('fx-alpha', 1, 0, 16)).toThrow(/chunks per interval/)
+    expect(() => hooks.startReasoningChunkStorm('fx-alpha', 1, 1, 0)).toThrow(/reasoning interval/)
+    const abort = new AbortController()
+    try {
+      const streamed = collect(api.events.mux(req({}), abort.signal), abort, frames => frames.some(frame => (
+        frame.type === 'session/event'
+        && frame.event.type === 'assistant/chunk'
+        && frame.event.data.chunk.type === 'reasoning-delta'
+        && frame.event.data.chunk.text.includes('REASONING_STRESS_COMPLETE')
+      )))
+      const marker = hooks.startReasoningChunkStorm('fx-alpha', 3, 2, 16)
+      expect(() => hooks.startReasoningChunkStorm('fx-alpha', 1, 1, 16)).toThrow(/already running/)
+      expect(hooks.reasoningChunkStormState()).toMatchObject({ emitted: 0, emitting: true, marker })
+
+      await vi.advanceTimersByTimeAsync(0)
+      expect(hooks.reasoningChunkStormState()).toMatchObject({ emitted: 2, emitting: true })
+      await vi.advanceTimersByTimeAsync(16)
+      expect(hooks.reasoningChunkStormState()).toEqual({
+        sessionId: 'fx-alpha', chunkCount: 3, chunksPerInterval: 2, intervalMs: 16,
+        emitted: 3, marker, emitting: false,
+      })
+
+      const frames = await streamed
+      const deltas = frames.flatMap(frame => (
+        frame.type === 'session/event'
+        && frame.event.type === 'assistant/chunk'
+        && frame.event.data.chunk.type === 'reasoning-delta'
+          ? [frame.event.data.chunk.text]
+          : []
+      ))
+      expect(deltas).toEqual(['推理', '推理', `\n${marker}`])
+    } finally {
+      abort.abort()
+      vi.useRealTimers()
+    }
+  })
 })
 
 describe('FixtureApiClient (protocol-level fake carrier)', () => {
@@ -752,6 +965,10 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
 
   it('covers the whole unary dispatch table', async () => {
     const client = new FixtureApiClient()
+    expect((await client.sessions.search(
+      { query: 'fixture' },
+      new AbortController().signal,
+    )).result.ok).toBe(true)
     const created = await client.sessions.create({})
     if (!created.result.ok) throw new Error('create failed')
     const id = created.result.value.sessionId
@@ -795,6 +1012,21 @@ describe('FixtureApiClient (protocol-level fake carrier)', () => {
     // complete → complete is an invalid transition.
     expect((await client.goals.complete({ sessionId: id, ref })).result.ok).toBe(false)
     expect((await client.goals.clear({ sessionId: id, ref })).result).toEqual({ ok: true, value: { cleared: true } })
+
+    const goalHistory = await client.sessions.history({ sessionId: id })
+    if (!goalHistory.result.ok) throw new Error('goal history failed')
+    const goalEvents = goalHistory.result.value.events.map(entry => entry.event as unknown as {
+      type: string
+      data: {
+        operation?: string
+        source?: { kind?: string; round?: number }
+      }
+    })
+    const goalChanges = goalEvents.filter(event => event.type === 'goal/change')
+    expect(goalChanges.map(event => event.data.operation))
+      .toEqual(['create', 'edit', 'pause', 'resume', 'complete', 'clear'])
+    expect(goalEvents.some(event => event.type === 'user/message'
+      && event.data.source?.kind === 'goal' && event.data.source.round === 0)).toBe(false)
   })
 
   it('maps empty, prompt-reject, and workspace-first query scenarios', async () => {

packages/client/connection/tests/loopback-hostname.spec.ts

@@ -0,0 +1,18 @@
+/** Shared loopback-hostname semantics for the Host fence and browser UI. */
+
+import { describe, expect, it } from 'vitest'
+import { isLoopbackHostname } from '../src/loopback-hostname.ts'
+
+describe('isLoopbackHostname', () => {
+  it('accepts localhost, IPv6 loopback, and the whole IPv4 127/8 block', () => {
+    for (const hostname of ['localhost', '[::1]', '127.0.0.1', '127.8.9.10', '127.255.255.255']) {
+      expect(isLoopbackHostname(hostname)).toBe(true)
+    }
+  })
+
+  it('refuses malformed and non-loopback hostnames', () => {
+    for (const hostname of ['remote.localhost', '::1', '128.0.0.1', '127.0.0', '127.0.0.256', '127.0.0.-1']) {
+      expect(isLoopbackHostname(hostname)).toBe(false)
+    }
+  })
+})

packages/client/connection/tests/node-half.spec.ts

@@ -1,29 +1,38 @@
 /** Node half: registers the /api prefix route bridging to the api gateway. */
-import { EventEmitter } from 'node:events'
-import { Readable } from 'node:stream'
+import { EventEmitter, once } from 'node:events'
+import { createServer, request as httpRequest } from 'node:http'
+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,
   }
 }
 
 /** Bodyless GET carrying the given headers (enough for the trust fence + bridge). */
-function fakeRequest(headers: Record<string, string>): IncomingMessage {
+function fakeRequest(headers: Record<string, string>, url = `${API_PATH}/session.list`): IncomingMessage {
   const request = Readable.from([]) as unknown as IncomingMessage
-  Object.assign(request, { url: `${API_PATH}/session.list`, method: 'GET', headers })
+  Object.assign(request, { url, method: 'GET', headers })
   return request
 }
 
@@ -43,47 +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)
-    // The apply throw also escapes cordis as a late rejection — the shape the
-    // boot's installFailLoud is contracted to catch. Capture it so the run
-    // stays clean, same pattern as the webserver bind-failure test.
-    const rejections: unknown[] = []
-    const onUnhandled = (err: unknown): void => { rejections.push(err) }
-    process.on('unhandledRejection', onUnhandled)
-    try {
-      const fiber = ctx.plugin({ inject: [...inject], apply }, { trustedHosts: ['harness.internal/path'] })
-      await expect(fiber.await()).rejects.toThrow(/not a bare host\[:port\] authority/)
-      expect(routes).toHaveLength(0)
-      for (let i = 0; i < 100 && rejections.length === 0; i++) {
-        await new Promise(resolve => setTimeout(resolve, 10))
-      }
-      expect(rejections.map(String).join('\n')).toContain('not a bare host[:port] authority')
-    } finally {
-      process.off('unhandledRejection', onUnhandled)
-    }
+    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 () => {
@@ -97,6 +126,31 @@ describe('connection node half', () => {
     await dispose()
   })
 
+  it('pins privileged methods to loopback even for a declared trusted authority', async () => {
+    const { routes, dispose } = await mounted({ trustedHosts: ['harness.example'] })
+    // The privileged set: native dialogs plus the whole settings/credential
+    // configuration plane, reads included. The same declared authority reaches
+    // ordinary reads (carrier-level 404 from the empty proxy proves the fence
+    // passed), but each privileged method stays loopback-only and 403s.
+    for (const method of [
+      'host.pickDirectory', 'host.openPath',
+      'settings.describe', 'settings.openDocument', 'settings.update', 'settings.replace', 'settings.mutate',
+      'credentials.describe', 'credentials.set', 'credentials.unset',
+    ]) {
+      const denied = fakeResponse()
+      await routes[0]!.handler(
+        fakeRequest({ host: 'harness.example' }, `${API_PATH}/${method}`),
+        denied.response,
+      )
+      expect(denied.state.status).toBe(403)
+      expect(denied.state.body).toBe('forbidden')
+    }
+    const read = fakeResponse()
+    await routes[0]!.handler(fakeRequest({ host: 'harness.example' }), read.response)
+    expect(read.state.status).not.toBe(403)
+    await dispose()
+  })
+
   it('passes loopback and declared-authority requests through to the bridge', async () => {
     const { routes, dispose } = await mounted({ trustedHosts: ['harness.example:3080', '192.168.1.5'] })
     // Loopback, no browser markers (curl shape): the fence passes; the carrier
@@ -118,3 +172,69 @@ describe('connection node half', () => {
     await dispose()
   })
 })
+
+describe('connection node half over a real HTTP server', () => {
+  /** Serve the registered prefix route from a real server and return its port. */
+  async function serve(routes: WebRoute[]): Promise<{ port: number; close: () => Promise<void> }> {
+    const server = createServer((request, response) => {
+      void routes[0]!.handler(request, response)
+    })
+    await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
+    const address = server.address() as AddressInfo
+    return {
+      port: address.port,
+      close: () => new Promise<void>((resolve, reject) => {
+        server.close((error) => {
+          if (error === undefined || error === null) resolve()
+          else reject(error)
+        })
+      }),
+    }
+  }
+
+  /** One real request; `host` spoofs the authority the way a LAN client's browser would send it. */
+  function call(port: number, method: string, host: string): Promise<number> {
+    return new Promise((resolve, reject) => {
+      const request = httpRequest(
+        { host: '127.0.0.1', port, path: `${API_PATH}/${method}`, method: 'GET', headers: { host } },
+        (response) => {
+          response.resume()
+          response.on('end', () => { resolve(response.statusCode ?? 0) })
+        },
+      )
+      request.on('error', reject)
+      request.end()
+    })
+  }
+
+  it('answers a declared LAN authority with 403 on every configuration method, over real HTTP', async () => {
+    // The fence's input is a real IncomingMessage parsed by Node from the
+    // wire, not a hand-assembled object: the Host header a LAN browser sends
+    // is exactly what decides loopback-only here, so the boundary is asserted
+    // against the parse the server actually performs.
+    const { routes, dispose } = await mounted({ trustedHosts: ['harness.example'] })
+    const { port, close } = await serve(routes)
+    try {
+      // Reads are as privileged as writes: describe returns the exposed
+      // configuration, and credentials.describe probes arbitrary env-var names.
+      for (const method of [
+        'settings.describe', 'settings.openDocument', 'settings.update', 'settings.replace', 'settings.mutate',
+        'credentials.describe', 'credentials.set', 'credentials.unset',
+        'host.pickDirectory', 'host.openPath',
+      ]) {
+        expect([method, await call(port, method, 'harness.example')]).toEqual([method, 403])
+      }
+      // The model catalog stays reachable for the same authority: a LAN
+      // client's model picker needs it, and it carries no key or endpoint
+      // state (404 is the empty proxy's carrier answer — the fence passed).
+      for (const method of ['llm.providers', 'llm.models']) {
+        expect([method, await call(port, method, 'harness.example')]).toEqual([method, 404])
+      }
+      // Loopback reaches everything, configuration included.
+      expect(await call(port, 'settings.describe', `127.0.0.1:${String(port)}`)).toBe(404)
+    } finally {
+      await close()
+      await dispose()
+    }
+  })
+})

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