Merge PR #500 into CI optimization

packages/client/connection/README.md

@@ -0,0 +1,16 @@
+# @deepseek-ai/dsh-client-connection
+
+Wire consumer layer (moved verbatim from web-runtime): IApiClient family (WebApiClient/FixtureApiClient), ConnectionController (SSE dual-stream + backoff reconnect), WEB_EVENTS. Contract: api-contracts v3 §3, export inventory in §3.2.
+
+## Model Experience
+
+None, as the wire consumer layer moves already-composed messages between browser and host; nothing here reaches a model request.
+
+#### KV Cache effect
+
+None; this package neither assembles nor sends a provider request.
+
+## Known Limitations and Deferred Work
+
+- **history's implicit resume is arguable** — opening history on an unattached session pulls an agent up host-side; the pure-persistence-read alternative is recorded in the rt-core reconciliation ledger, unchanged in P-I. This package's consumers see it as latency on first open.
+- **`ToolEventView`/`ToolCallView`/`ToolResultView` re-exports are scheduled for removal** — they fall when the toolview migration deletes the host `viewFor` line (presentation belongs to the client); the fixture keeps a local `viewFor` mirror until then.

packages/client/connection/package.json

@@ -0,0 +1,53 @@
+{
+  "name": "@deepseek-ai/dsh-client-connection",
+  "description": "Wire consumer layer: IApiClient subclasses, ConnectionController (SSE dual-stream + reconnect), fixture api (no cordis)",
+  "version": "0.0.1",
+  "private": true,
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/types/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/types/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./invariant": {
+      "types": "./lib/types/invariant.d.ts",
+      "default": "./lib/invariant.js"
+    },
+    "./client": {
+      "types": "./lib/types/client/index.d.ts",
+      "default": "./lib/client.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "dshClient": {
+    "inject": [],
+    "platform": "web",
+    "immediately": true
+  },
+  "license": "BSD-3-Clause",
+  "dependencies": {
+    "@deepseek-ai/dsh-host-apiproxy": "workspace:^",
+    "@deepseek-ai/dsh-llm": "workspace:^",
+    "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-tools": "workspace:^"
+  },
+  "files": [
+    "lib/index.js",
+    "lib/invariant.js",
+    "lib/client.js",
+    "lib/types/**/*.d.ts",
+    "lib/types/**/*.d.ts.map",
+    "src"
+  ],
+  "peerDependencies": {
+    "@deepseek-ai/dsh-invariants": "^0.0.1",
+    "cordis": "^4.0.0-rc.7"
+  },
+  "devDependencies": {
+    "@deepseek-ai/dsh-invariants": "workspace:^",
+    "cordis": "^4.0.0-rc.7"
+  }
+}

packages/client/connection/src/client/api.ts

@@ -0,0 +1,46 @@
+// 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.
+// 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,
+  ApprovalResponsePayload, QuestionResponsePayload, HistoryEntry, ToolEventView,
+} from '@deepseek-ai/dsh-host-apiproxy/api'
+export type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'
+export type {
+  RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
+  ClientRequest, ServerResponse, ServerRequest, ClientResponse, RpcMessage, RpcReceipt,
+} from '@deepseek-ai/dsh-host-apiproxy/api'
+export { RpcId } 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 { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm/types'
+
+import type { RpcResponse, RpcResult } from '@deepseek-ai/dsh-host-apiproxy/api'
+
+/**
+ * Unwrap a unary response: RpcResponse<T> -> RpcResult<T> (business code only
+ * cares about the result slot).
+ * @param response - the unary response.
+ * @returns its result slot.
+ */
+export function resultOf<T>(response: RpcResponse<T>): RpcResult<T> {
+  return response.result
+}
+
+/**
+ * Fold a transport exception into the RpcResult error branch (unified error
+ * surface; 'internal' as the catch-all code).
+ * @param error - the thrown value from the carrier.
+ * @returns the error branch of an RpcResult.
+ */
+export function transportError<T>(error: unknown): RpcResult<T> {
+  return {
+    ok: false,
+    error: { code: 'internal', message: error instanceof Error ? error.message : String(error), details: {} },
+  }
+}

packages/client/connection/src/client/connection.ts

@@ -0,0 +1,190 @@
+import type { IApiClient, HostFrame, MuxFrame, RpcRequest } from './api.ts'
+
+/** Reconnect/backoff tunables (deployment-varying — no hardcoded tunables; web-cordis §B.1 lists
+ *  these as the future `ctx.connection` plugin Config). All fields optional; defaults below. */
+export interface ConnectionConfig {
+  /** First-retry backoff cap in ms (jittered: actual delay is cap/2..cap). */
+  backoffBaseMs?: number
+  /** Exponential growth factor per consecutive failed attempt. */
+  backoffFactor?: number
+  /** Upper bound for the backoff cap in ms. */
+  backoffMaxMs?: number
+  /** Cap on waiting for both streams' onOpen before onConnected, in ms. The strict handshake
+   *  (audit C2) waits for mux+host stream establishment plus describe; a carrier that never
+   *  fires onOpen (misbehaving proxy) must not wedge the connection forever — on timeout the
+   *  generation proceeds as connected and the live-gap repair path (audit S3) covers stragglers. */
+  streamOpenTimeoutMs?: number
+}
+
+const CONNECTION_DEFAULTS: Required<ConnectionConfig> = {
+  backoffBaseMs: 500,
+  backoffFactor: 2,
+  backoffMaxMs: 10_000,
+  streamOpenTimeoutMs: 3_000,
+}
+
+function sleep(ms: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve) => {
+    const t = setTimeout(done, ms)
+    signal.addEventListener('abort', done, { once: true })
+    function done(): void {
+      clearTimeout(t)
+      signal.removeEventListener('abort', done)
+      resolve()
+    }
+  })
+}
+
+/** Coarse connection state for the UI (audit C1): 'connected' after each generation's handshake,
+ *  'reconnecting' the moment the generation fails (covers the whole backoff+retry span). */
+export type ConnectionState = 'connected' | 'reconnecting'
+
+/** Frame sink callbacks: the Controller owns the physical streams; business dispatch belongs to
+ *  SessionManager. */
+export interface ConnectionSinks {
+  onMuxEnvelope?: (envelope: RpcRequest<MuxFrame>) => void
+  onHostEnvelope?: (envelope: RpcRequest<HostFrame>) => void
+  /** After each connection generation is established (both streams open + describe succeeded), first connect included. */
+  onConnected?: () => void
+  /** Coarse state transitions (deduplicated: fires only on change). The initial pre-connect
+   *  span reports nothing — the UI treats "no state yet" as connecting, not as an outage. */
+  onStateChange?: (state: ConnectionState) => void
+}
+
+/**
+ * Opens both streams and keeps iterating (pull mode: nothing reads the socket and the tap
+ * never fires unless someone for-awaits), reconnecting with exponential backoff on loss.
+ * State (generation/attempt) is instance-private, never in the store.
+ * The pump body feeds each frame to a sink (sink exceptions must
+ * not kill the pump — a broken business layer must not drag down the connection layer).
+ */
+export class ConnectionController {
+  private generation = 0
+  private attempt = 0
+  private current: AbortController | null = null
+  private running = false
+  private lastState: ConnectionState | null = null
+  private readonly config: Required<ConnectionConfig>
+
+  constructor(
+    private readonly api: IApiClient,
+    private readonly sinks: ConnectionSinks = {},
+    config: ConnectionConfig = {},
+  ) {
+    this.config = { ...CONNECTION_DEFAULTS, ...config }
+  }
+
+  /** Idempotent: begin the connect/pump/reconnect loop. */
+  start(): void {
+    if (this.running) return
+    this.running = true
+    void this.loop()
+  }
+
+  /** Stop the loop and abort the current generation's streams. */
+  stop(): void {
+    this.running = false
+    this.current?.abort()
+    this.current = null
+  }
+
+  private backoffDelay(attempt: number): number {
+    const { backoffBaseMs, backoffFactor, backoffMaxMs } = this.config
+    const cap = Math.min(backoffMaxMs, backoffBaseMs * backoffFactor ** Math.max(0, attempt - 1))
+    return cap / 2 + Math.random() * (cap / 2)
+  }
+
+  /** Read through a method: stop() flips the flag across awaits, so narrowing from the loop condition must not stick. */
+  private isRunning(): boolean {
+    return this.running
+  }
+
+  private async loop(): Promise<void> {
+    while (this.running) {
+      const gen = ++this.generation
+      const ac = new AbortController()
+      this.current = ac
+
+      /* v8 ignore next -- initializer placeholder: the Promise executor
+       * below runs synchronously and replaces it before anyone can call it. */
+      let muxOpened = (): void => {}
+      /* v8 ignore next -- same placeholder pattern as muxOpened. */
+      let hostOpened = (): void => {}
+      const streamsOpen = Promise.all([
+        new Promise<void>((resolve) => { muxOpened = resolve }),
+        new Promise<void>((resolve) => { hostOpened = resolve }),
+      ])
+
+      const failed = new Promise<void>((resolve) => {
+        const settle = (): void => {
+          if (gen === this.generation && !ac.signal.aborted) ac.abort()
+          resolve()
+        }
+        void this.pumpStream(this.api.events.mux({}, ac.signal, muxOpened), this.sinks.onMuxEnvelope, settle)
+        void this.pumpStream(this.api.events.host({}, ac.signal, hostOpened), this.sinks.onHostEnvelope, settle)
+      })
+
+      try {
+        // Strict readiness handshake (audit C2): describe proves unary reachability, onOpen
+        // proves each SSE transport is established (response headers in, 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).
+        const timeout = new AbortController()
+        await Promise.all([
+          this.api.host.describe({}),
+          Promise.race([streamsOpen, sleep(this.config.streamOpenTimeoutMs, timeout.signal)]),
+        ])
+        timeout.abort()
+        if (ac.signal.aborted) throw new Error('generation aborted during readiness handshake')
+        this.attempt = 0
+        this.emitState('connected')
+        this.callSink(this.sinks.onConnected)
+      } catch {
+        // Transport failure: treat as generation failure, fall through to the shared backoff.
+        if (!ac.signal.aborted) ac.abort()
+      }
+
+      await failed
+      if (!this.isRunning()) return
+      this.emitState('reconnecting')
+      this.attempt += 1
+      console.warn(`[web-runtime] connection lost, retry #${this.attempt}`)
+      const idle = new AbortController()
+      await sleep(this.backoffDelay(this.attempt), idle.signal)
+    }
+  }
+
+  /** Deduplicated state emission (sink isolation applies). */
+  private emitState(state: ConnectionState): void {
+    if (this.lastState === state) return
+    this.lastState = state
+    this.callSink(() => this.sinks.onStateChange?.(state))
+  }
+
+  private async pumpStream<F extends { type: string }>(
+    stream: AsyncIterable<RpcRequest<F>>,
+    sink: ((envelope: RpcRequest<F>) => void) | undefined,
+    onEnd: () => void,
+  ): Promise<void> {
+    try {
+      for await (const envelope of stream) {
+        if (envelope.payload.type === 'stream/error') break
+        if (sink !== undefined) this.callSink(() => { sink(envelope) })
+      }
+    } catch {
+      // Stream loss: converge on onEnd, which triggers the shared reconnect.
+    }
+    onEnd()
+  }
+
+  /** Sink exception isolation: a business-layer throw is logged only, never affecting pump or reconnect semantics. */
+  private callSink(fn: (() => void) | undefined): void {
+    if (fn === undefined) return
+    try {
+      fn()
+    } catch (error) {
+      console.error('[web-runtime] connection sink threw:', error)
+    }
+  }
+}

packages/client/connection/src/client/fixture.ts

@@ -0,0 +1,560 @@
+// FixtureApi: standalone UI development without a server. Real contract shape: unary takes
+// RpcRequest<P> and returns RpcResponse<T> (echoing the rpcId); streams yield RpcRequest<frame>
+// (the fixture IS the fake server, so it mints frame rpcIds); root respond takes ClientResponse
+// and returns RpcReceipt. fx-alpha carries a hand-built history script (60 turns, pageable);
+// prompt triggers a chunked streaming replay; cancel stops the replay; one resident pending
+// approval (placeholder-card material, subscribed-baseline-replay semantics: stable rpcId reuse).
+
+import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
+import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
+import type {
+  ApiProxy, ClientRequest, ClientResponse, HistoryEntry, HostFrame, MuxFrame, RpcReceipt,
+  RpcRequest, RpcResponse, RpcResult, ServerRequest, ServerResponse, SessionSummary,
+  ToolCallView, ToolEventView, ToolResultView,
+} from './api.ts'
+import type { RequestPayload, ResponseValue, RpcMethodMap } from '@deepseek-ai/dsh-host-apiproxy/api'
+import { AbstractApiClient, RpcId } from './api.ts'
+
+/** The fake carrier mints like a real one (business code never mints). */
+function rpcRequest<P>(payload: P): RpcRequest<P> {
+  return { rpcId: RpcId(crypto.randomUUID()), payload }
+}
+
+function text(t: string): ContentBlock[] {
+  return [{ type: 'text', text: t }]
+}
+
+function sid(id: string): SessionId {
+  return id as SessionId
+}
+
+/** fx-alpha history script: 60 turns (~130+ messages -> 3 pages at PAGE_MESSAGES=50),
+ *  mixing reasoning blocks / tool call+result / steering / context. */
+function buildAlphaLog(): SessionEvent[] {
+  const events: Record<string, unknown>[] = []
+  let time = Date.now() - 3_600_000
+  const push = (e: Record<string, unknown>): number => {
+    const seq = events.length
+    events.push({ seq, time: (time += 800), ...e })
+    return seq
+  }
+  for (let turn = 0; turn < 60; turn++) {
+    push({ type: 'turn/start', data: { turn, trigger: { kind: 'message', source: { kind: 'user' } } } })
+    push({ type: 'user/message', surfaceOp: 'append', data: { content: text(`问题 ${turn}：fixture 历史消息，用于翻页与渲染验收。`), source: { kind: 'user' } } })
+    if (turn % 9 === 4) {
+      push({ type: 'context/message', surfaceOp: 'append', data: { content: text(`[fixture] 上下文注入（turn ${turn}）`), source: { kind: 'plugin', plugin: 'fixture' } } })
+    }
+    push({ type: 'step/start', data: { turn, step: 0 } })
+    const withTool = turn % 5 === 2
+    const withReasoning = turn % 3 === 1
+    const blocks: ContentBlock[] = []
+    if (withReasoning) blocks.push({ type: 'reasoning', text: `思考过程 ${turn}：这是一段可折叠的 reasoning 内容。` })
+    blocks.push({ type: 'text', text: `回答 ${turn}：这是 fixture 生成的历史回复正文。` })
+    if (withTool) {
+      const callId = `fx-call-${turn}`
+      blocks.push({ type: 'tool-call', id: callId, name: 'echo', arguments: `{"text":"turn ${turn}"}` } as ContentBlock)
+      push({ type: 'assistant/message', surfaceOp: 'append', data: { turn, step: 0, content: blocks, provenance: { provider: 'fixture', model: 'fx-1' } } })
+      push({ type: 'tool/call', data: { turn, step: 0, callId, name: 'echo', arguments: `{"text":"turn ${turn}"}` } })
+      push({ type: 'tool/result', surfaceOp: 'append', data: { turn, step: 0, callId, content: text(`ECHO: TURN ${turn}`), isError: turn % 25 === 12 } })
+      push({ type: 'step/end', data: { turn, step: 0 } })
+      push({ type: 'step/start', data: { turn, step: 1 } })
+      push({ type: 'assistant/message', surfaceOp: 'append', data: { turn, step: 1, content: text(`工具结果已消化（turn ${turn}）。`), provenance: { provider: 'fixture', model: 'fx-1' } } })
+      push({ type: 'step/end', data: { turn, step: 1 } })
+    } else {
+      push({ type: 'assistant/message', surfaceOp: 'append', data: { turn, step: 0, content: blocks, provenance: { provider: 'fixture', model: 'fx-1' } } })
+      push({ type: 'step/end', data: { turn, step: 0 } })
+    }
+    if (turn % 13 === 6) {
+      push({ type: 'steering/message', surfaceOp: 'append', data: { turn, content: text(`插话 ${turn}：fixture steering 消息。`), source: { kind: 'user' } } })
+    }
+    push({ type: 'turn/end', data: { turn, reason: { kind: 'completed' } } })
+  }
+  // Three view-sample turns (60-62) for the tool-card wire acceptance: one per built-in card
+  // type. `echo` above stays presenter-less on purpose — it is the no-view fallback sample.
+  const toolTurn = (turn: number, name: string, args: string, resultText: string): void => {
+    const callId = `fx-call-${turn}`
+    push({ type: 'turn/start', data: { turn, trigger: { kind: 'message', source: { kind: 'user' } } } })
+    push({ type: 'user/message', surfaceOp: 'append', data: { content: text(`问题 ${turn}：${name} 样本。`), source: { kind: 'user' } } })
+    push({ type: 'step/start', data: { turn, step: 0 } })
+    push({
+      type: 'assistant/message', surfaceOp: 'append',
+      data: { turn, step: 0, content: [{ type: 'tool-call', id: callId, name, arguments: args } as ContentBlock], provenance: { provider: 'fixture', model: 'fx-1' } },
+    })
+    push({ type: 'tool/call', data: { turn, step: 0, callId, name, arguments: args } })
+    push({ type: 'tool/result', surfaceOp: 'append', data: { turn, step: 0, callId, content: text(resultText), isError: false } })
+    push({ type: 'step/end', data: { turn, step: 0 } })
+    push({ type: 'turn/end', data: { turn, reason: { kind: 'completed' } } })
+  }
+  toolTurn(60, 'fx-bash', '{"command":"ls -la","cwd":"/tmp/fixture"}', 'total 2\ndrwxr-xr-x fixture\n-rw-r--r-- demo.txt')
+  toolTurn(61, 'fx-write', '{"path":"notes/demo.txt","content":"hello fixture\\n"}', 'wrote notes/demo.txt')
+  toolTurn(62, 'fx-note', '{"note":"三型卡验收样本"}', '已记录')
+  return events as unknown as SessionEvent[]
+}
+
+/** Narrows a parsed-JSON field to string; fixture args are authored in-file, so non-strings only mean a typo here. */
+/* v8 ignore next -- the fallback arm is the same in-file-typo guard as the JSON.parse catch above. */
+const str = (value: unknown, fallback = ''): string => typeof value === 'string' ? value : fallback
+
+/** Fixture presenter registry (mirrors host viewFor): pure derivation, undefined = no view. */
+function presentCall(name: string, argsRaw: string): ToolCallView | undefined {
+  let args: Record<string, unknown>
+  try {
+    args = JSON.parse(argsRaw) as Record<string, unknown>
+  } catch {
+    /* v8 ignore next 2 -- defensive: fixture args are authored in-file as valid JSON; only an in-file typo could reach the catch. */
+    return undefined
+  }
+  switch (name) {
+    case 'fx-bash':
+      return { card: 'terminal', title: str(args.command), cwd: str(args.cwd, '/tmp/fixture'), description: 'fixture 终端样本' }
+    case 'fx-write':
+      return {
+        card: 'diff', title: `Write ${str(args.path)}`,
+        diffs: [{ path: str(args.path), oldText: null, newText: str(args.content) }],
+      }
+    case 'fx-note':
+      return { card: 'generic', title: '记录笔记', kind: 'edit', rawInput: args }
+    default:
+      return undefined // echo et al: the documented no-view fallback path
+  }
+}
+
+function presentResult(name: string, argsRaw: string, resultText: string): ToolResultView | undefined {
+  const call = presentCall(name, argsRaw)
+  if (call === undefined) return undefined
+  switch (call.card) {
+    case 'terminal':
+      return { card: 'terminal', output: resultText, exitCode: 0 }
+    case 'diff':
+      return { card: 'diff', diffs: call.diffs }
+    case 'generic':
+      return { card: 'generic', content: text(resultText) }
+  }
+}
+
+/** Host-side viewFor mirror: tool/call presents from its own args; tool/result back-scans the log for the paired call. */
+function viewFor(event: SessionEvent, log: readonly SessionEvent[]): ToolEventView | undefined {
+  if (event.type === 'tool/call') {
+    const view = presentCall(event.data.name, event.data.arguments)
+    return view === undefined ? undefined : { for: 'call', view }
+  }
+  if (event.type === 'tool/result') {
+    const callId = String(event.data.callId)
+    for (let i = log.length - 1; i >= 0; i--) {
+      const candidate = log[i]
+      /* v8 ignore next -- dense-array guard: i stays within [0, log.length),
+      so the undefined arm needs a sparse log no code path builds. */
+      if (candidate !== undefined && candidate.type === 'tool/call' && String(candidate.data.callId) === callId) {
+        const resultText = event.data.content.map(b => (b.type === 'text' ? b.text : '')).join('')
+        const view = presentResult(candidate.data.name, candidate.data.arguments, resultText)
+        return view === undefined ? undefined : { for: 'result', view }
+      }
+    }
+    return undefined // cross-page unpaired: documented default
+  }
+  return undefined
+}
+
+/**
+ * Message-boundary paging (mirrors the host's paging contract): count
+ * maxMessages messages
+ *  backwards from end, cut at a turn/start boundary.
+ Entries carry pagination-time views
+ *  (the host analogue computes viewFor per entry at page time). */
+function pageOf(
+  log: readonly SessionEvent[],
+  beforeSeq: number | undefined,
+  maxMessages: number,
+): { events: HistoryEntry[]; hasMore: boolean } {
+  const end = beforeSeq === undefined ? log.length : Math.max(0, Math.min(beforeSeq, log.length))
+  let start = 0
+  let messages = 0
+  for (let i = end - 1; i >= 0; i--) {
+    const event = log[i]
+    /* v8 ignore next -- dense-array guard: log seqs are array indexes, i stays within [0, end). */
+    if (event === undefined) break
+    if (event.type === 'user/message' || event.type === 'assistant/message' || event.type === 'steering/message') messages++
+    if (event.type === 'turn/start' && messages >= maxMessages) {
+      start = i
+      break
+    }
+  }
+  const events = log.slice(start, end).map((event): HistoryEntry => {
+    const view = viewFor(event, log)
+    return view === undefined ? { event } : { event, view }
+  })
+  return { events, hasMore: start > 0 }
+}
+
+interface StreamConn<F> {
+  push(envelope: RpcRequest<F>): void
+}
+
+/** Inbox pump shared by both stream generators (FrameQueue pattern: ONE abort listener hung
+ *  outside the loop — a per-iteration {once:true} listener never fires for non-final rounds and
+ *  piles up for the stream's lifetime, audit C5). breakNow force-ends the stream without the
+ *  client's signal (timing hook: simulated connection loss). */
+class FxInbox<F> implements StreamConn<F> {
+  private readonly inbox: RpcRequest<F>[] = []
+  private wake: (() => void) | null = null
+  private broken = false
+
+  push(envelope: RpcRequest<F>): void {
+    this.inbox.push(envelope)
+    this.wake?.()
+  }
+
+  breakNow(): void {
+    this.broken = true
+    this.wake?.()
+  }
+
+  /** Read through a method: breakNow()/abort flip state across yields, so narrowing from the loop condition must not stick. */
+  private isLive(signal: AbortSignal): boolean {
+    return !signal.aborted && !this.broken
+  }
+
+  async *drain(signal: AbortSignal): AsyncGenerator<RpcRequest<F>> {
+    const onAbort = (): void => this.wake?.()
+    signal.addEventListener('abort', onAbort)
+    try {
+      while (this.isLive(signal)) {
+        while (this.inbox.length > 0) yield this.inbox.shift() as RpcRequest<F>
+        if (!this.isLive(signal)) break
+        await new Promise<void>((resolve) => {
+          this.wake = resolve
+        })
+        this.wake = null
+      }
+    } finally {
+      signal.removeEventListener('abort', onAbort)
+    }
+  }
+}
+
+/**
+ * In-memory fake host: fx-alpha carries history and replay scripts; fx-beta is fx-alpha's child session (lineage indent material).
+ * @returns an ApiProxy backed entirely by in-memory state — no host process, no network.
+ */
+export function createFixtureApi(): ApiProxy {
+  const sessions: SessionSummary[] = [
+    { sessionId: sid('fx-alpha'), updatedAt: Date.now(), running: true, cwd: '/tmp/fixture' },
+    { sessionId: sid('fx-beta'), updatedAt: Date.now() - 60_000, running: false, parentSessionId: sid('fx-alpha'), cwd: '/tmp/fixture' },
+    { sessionId: sid('fx-gamma'), updatedAt: Date.now() - 120_000, running: false, cwd: '/tmp/fixture' },
+  ]
+  const logs = new Map<SessionId, SessionEvent[]>([[sid('fx-alpha'), buildAlphaLog()]])
+  const nextTurn = new Map<SessionId, number>([[sid('fx-alpha'), 60]])
+  let nextSession = 1
+  let nextRpc = 1
+  const mint = (): ReturnType<typeof RpcId> => RpcId(`fx-rpc-${nextRpc++}`)
+  /** Resident pending approval (stable rpcId: every mux open replays the same id, matching host replay semantics). */
+  const pendingApprovalRpcId = mint()
+
+  const muxConns = new Set<StreamConn<MuxFrame>>()
+  const hostConns = new Set<StreamConn<HostFrame>>()
+  const emitMux = (frame: MuxFrame): void => {
+    for (const conn of muxConns) conn.push({ rpcId: mint(), payload: frame })
+  }
+  const emitHost = (frame: HostFrame): void => {
+    for (const conn of hostConns) conn.push({ rpcId: mint(), payload: frame })
+  }
+
+  /** OK response echoing the caller's rpcId (contract: responses always backfill, never mint). */
+  function ok<P, T>(request: RpcRequest<P>, value: T): Promise<RpcResponse<T>> {
+    return Promise.resolve({ rpcId: request.rpcId, result: { ok: true, value } })
+  }
+  function err<P, T>(request: RpcRequest<P>, error: Extract<RpcResult<T>, { ok: false }>['error']): Promise<RpcResponse<T>> {
+    return Promise.resolve({ rpcId: request.rpcId, result: { ok: false, error } })
+  }
+
+  const summaryOf = (id: SessionId): SessionSummary | undefined => sessions.find(s => s.sessionId === id)
+  const setRunning = (id: SessionId, running: boolean): void => {
+    const summary = summaryOf(id)
+    if (summary === undefined || summary.running === running) return
+    summary.running = running
+    emitHost({ type: 'host/session-status', sessionId: id, running })
+  }
+  const logOf = (id: SessionId): SessionEvent[] => {
+    let log = logs.get(id)
+    if (log === undefined) {
+      log = []
+      logs.set(id, log)
+    }
+    return log
+  }
+  const append = (id: SessionId, e: Record<string, unknown>): void => {
+    const log = logOf(id)
+    const event = { seq: log.length, time: Date.now(), ...e } as unknown as SessionEvent
+    log.push(event)
+    // Emission-time view derivation (mirrors the host's live path).
+    const view = viewFor(event, log)
+    /* v8 ignore next 3 -- the view-present arm needs a live tool/call emission,
+    but the fixture replay produces text-only turns; view vocabulary is
+    exercised through the history samples (turns 60-62). */
+    emitMux(view === undefined
+      ? { type: 'session/event', sessionId: id, event }
+      : { type: 'session/event', sessionId: id, event, view })
+  }
+
+  /** At most one in-flight replay per session; cancel clears it. */
+  const replays = new Map<SessionId, { timer: ReturnType<typeof setTimeout>; finish(aborted: boolean): void }>()
+
+  /** history transit delay (timing hooks below); the page snapshot is taken at request time, like a real host. */
+  let historyDelayMs = 0
+  /** One-shot history failure (timing hook: the doomed in-flight request of the S4 reconnect scenario). */
+  let failNextHistory = false
+  /** Force-enders for currently open stream generators (timing hook: simulated connection loss). */
+  const streamBreakers = new Set<() => void>()
+
+  // Timing-acceptance hooks (browser test backdoor): the in-memory fixture is ideally timed, which
+  // is exactly what masked the open-window and reconnect-gap bugs (audit S1/S3). These let
+  // browser acceptance runs create slow-history, lost-frame, and reconnect
+  // windows a real host produces naturally.
+  const timingHooks = {
+    setHistoryDelay(ms: number): void {
+      historyDelayMs = ms
+    },
+    /** Fail the NEXT history call (after its transit delay) with a transport-level throw. */
+    failNextHistory(): void {
+      failNextHistory = true
+    },
+    /** Log append + mux emit (the normal live path). */
+    appendUser(id: string, msg: string): void {
+      append(sid(id), { type: 'user/message', surfaceOp: 'append', data: { content: text(msg), source: { kind: 'user' } } })
+    },
+    /** Log append WITHOUT the mux emit: a frame lost in transit — history still serves it, the client must repull. */
+    appendSilent(id: string, msg: string): void {
+      const log = logOf(sid(id))
+      log.push({ type: 'user/message', surfaceOp: 'append', seq: log.length, time: Date.now(), data: { content: text(msg), source: { kind: 'user' } } } as unknown as SessionEvent)
+    },
+    /** End every open stream generator (client sees both streams close -> reconnect + resync path). */
+    breakStreams(): void {
+      for (const breakNow of [...streamBreakers]) breakNow()
+    },
+  }
+  ;(globalThis as Record<string, unknown>).__fxTiming = timingHooks
+
+  /** Prompt replay: chunk typewriter (80ms/frame) -> assistant/message finalize -> turn/end + running flip. */
+  const startReply = (id: SessionId, turn: number, replyText: string): void => {
+    const step = 0
+    append(id, { type: 'step/start', data: { turn, step } })
+    append(id, { type: 'assistant/chunk', data: { turn, step, chunk: { type: 'block-start', index: 0, blockType: 'text' } } })
+    /* v8 ignore next -- the ?? arm needs a null match, but replyText is never empty (prompt always prefixes 回声). */
+    const pieces = replyText.match(/.{1,6}/gu) ?? [replyText]
+    let i = 0
+    const finish = (aborted: boolean): void => {
+      replays.delete(id)
+      const done = pieces.slice(0, i).join('')
+      append(id, { type: 'assistant/chunk', data: { turn, step, chunk: { type: 'block-end', index: 0, block: { type: 'text', text: done } } } })
+      append(id, { type: 'assistant/message', surfaceOp: 'append', data: { turn, step, content: text(aborted ? `${done}（已中断）` : done), provenance: { provider: 'fixture', model: 'fx-1' } } })
+      append(id, { type: 'step/end', data: { turn, step } })
+      append(id, { type: 'turn/end', data: { turn, reason: { kind: aborted ? 'cancelled' : 'completed' } } })
+      setRunning(id, false)
+    }
+    const tick = (): void => {
+      const piece = pieces[i]
+      if (piece === undefined) {
+        finish(false)
+        return
+      }
+      i++
+      append(id, { type: 'assistant/chunk', data: { turn, step, chunk: { type: 'text-delta', index: 0, text: piece } } })
+      replays.set(id, { timer: setTimeout(tick, 80), finish })
+    }
+    replays.set(id, { timer: setTimeout(tick, 80), finish })
+  }
+
+  return {
+    sessions: {
+      list: request => ok(request, { items: [...sessions].sort((a, b) => b.updatedAt - a.updatedAt) }),
+      create: (request) => {
+        const created: SessionSummary = {
+          sessionId: sid(`fx-${nextSession++}`), updatedAt: Date.now(), running: false, cwd: '/tmp/fixture',
+        }
+        sessions.push(created)
+        emitHost({ type: 'host/session-added', sessionId: created.sessionId })
+        return ok(request, { sessionId: created.sessionId })
+      },
+      history: async (request) => {
+        const log = logs.get(request.payload.sessionId) ?? []
+        // Snapshot at request time, deliver after the transit delay (mirrors a real host under latency).
+        const page = pageOf(log, request.payload.beforeSeq, request.payload.maxMessages ?? 50)
+        const doomed = failNextHistory
+        failNextHistory = false
+        const delay = historyDelayMs
+        if (delay > 0) await new Promise(resolve => setTimeout(resolve, delay))
+        if (doomed) throw new Error('fixture: simulated history transport failure')
+        return ok(request, page)
+      },
+      prompt: (request) => {
+        const { sessionId: id, mode, content } = request.payload
+        const summary = summaryOf(id)
+        if (summary === undefined) {
+          return err(request, { code: 'session-not-found', message: `no session ${id}`, details: { sessionId: id } })
+        }
+        summary.updatedAt = Date.now()
+        const userText = content.map(b => (b.type === 'text' ? b.text : '')).join('')
+        if (mode === 'steer' && replays.has(id)) {
+          // Steering: insert a steering message into the current turn; the replay continues.
+          /* v8 ignore next -- the ?? arm needs a missing counter, but a live replay implies a prior prompt already set it. */
+          const turn = (nextTurn.get(id) ?? 1) - 1
+          append(id, { type: 'steering/message', surfaceOp: 'append', data: { turn, content, source: { kind: 'user' } } })
+          return ok(request, { accepted: true as const })
+        }
+        const turn = nextTurn.get(id) ?? 0
+        nextTurn.set(id, turn + 1)
+        setRunning(id, true)
+        append(id, { type: 'turn/start', data: { turn, trigger: { kind: 'message', source: { kind: 'user' } } } })
+        append(id, { type: 'user/message', surfaceOp: 'append', data: { content, source: { kind: 'user' } } })
+        startReply(id, turn, `回声：${userText}。这是 fixture 的流式回复，用于验证打字机增长与定稿切换。`)
+        return ok(request, { accepted: true as const })
+      },
+      cancel: (request) => {
+        const replay = replays.get(request.payload.sessionId)
+        if (replay !== undefined) {
+          clearTimeout(replay.timer)
+          replay.finish(true)
+        } else {
+          setRunning(request.payload.sessionId, false)
+        }
+        return ok(request, { accepted: true as const })
+      },
+    },
+    host: {
+      describe: request => ok(request, { version: '0.0.0-fixture', cwd: '/tmp/fixture', attachedSessions: 1 }),
+    },
+    events: {
+      async *mux(_request, signal) {
+        const conn = new FxInbox<MuxFrame>()
+        muxConns.add(conn)
+        const breakNow = (): void => { conn.breakNow() }
+        streamBreakers.add(breakNow)
+        // Open baseline: subscribed for attached (running) sessions + pending approval replay (stable rpcId).
+        for (const s of sessions) {
+          if (!s.running) continue
+          conn.push({ rpcId: mint(), payload: { type: 'session/subscribed', sessionId: s.sessionId, lastSeq: (logs.get(s.sessionId)?.length ?? 0) - 1 } })
+        }
+        conn.push({
+          rpcId: pendingApprovalRpcId,
+          payload: {
+            type: 'approval/requested', sessionId: sid('fx-alpha'),
+            approvalId: 'fx-approval-1' as MuxFrame extends never ? never : Extract<MuxFrame, { type: 'approval/requested' }>['approvalId'],
+            toolName: 'dangerous_tool', reason: 'fixture 常驻占位审批（可见不可答）',
+          },
+        })
+        try {
+          yield* conn.drain(signal)
+        } finally {
+          streamBreakers.delete(breakNow)
+          muxConns.delete(conn)
+        }
+      },
+      async *host(_request, signal) {
+        const conn = new FxInbox<HostFrame>()
+        hostConns.add(conn)
+        const breakNow = (): void => { conn.breakNow() }
+        streamBreakers.add(breakNow)
+        // Periodic material (the RPC-panel acceptance's clear-then-new-frames step depends on it): flip fx-gamma every 5s.
+        // fx-gamma only: never touch fx-alpha's running semantics (the conversation replay drives that).
+        const timer = setInterval(() => {
+          const gamma = summaryOf(sid('fx-gamma'))
+          /* v8 ignore next -- the undefined arm needs fx-gamma deleted, but the fixture never removes sessions. */
+          if (gamma !== undefined) setRunning(gamma.sessionId, !gamma.running)
+        }, 5000)
+        try {
+          yield* conn.drain(signal)
+        } finally {
+          clearInterval(timer)
+          streamBreakers.delete(breakNow)
+          hostConns.delete(conn)
+        }
+      },
+    },
+    respond(message: ClientResponse): Promise<RpcReceipt> {
+      // The v1 UI never answers (PendingCard is visible but not answerable); implemented for type completeness, always not-pending.
+      void message
+      return Promise.resolve({ accepted: false, reason: 'not-pending' })
+    },
+  }
+}
+
+/**
+ * Fixture platform subclass: there is no HTTP at all, so instead of a doFetch transport it
+ * overrides the protocol-level virtuals (callUnary/openMux/openHost/respond) to dispatch
+ * straight into the in-memory ApiProxy — while still minting rpcIds, fabricating the four
+ * named full forms, and feeding the same tap as a real carrier. Delete when the fixture moves
+ * to the isomorphic pipeline (InProcessApiClient over toFetchHandler(fixtureImpl)).
+ */
+export class FixtureApiClient extends AbstractApiClient {
+  private readonly api = createFixtureApi()
+
+  protected doFetch(): Promise<Response> {
+    throw new Error('FixtureApiClient overrides all protocol paths; doFetch must be unreachable')
+  }
+
+  protected override async callUnary<K extends keyof RpcMethodMap>(
+    method: K,
+    payload: RequestPayload<K>,
+  ): Promise<RpcResponse<ResponseValue<K>>> {
+    const request = rpcRequest(payload)
+    const full: ClientRequest = { type: 'client-request', rpcId: request.rpcId, method, payload }
+    this.onEnvelope(full)
+    const response = await this.dispatch(method, request as RpcRequest<never>) as RpcResponse<ResponseValue<K>>
+    const fullResponse: ServerResponse = { type: 'server-response', rpcId: response.rpcId, result: response.result }
+    this.onEnvelope(fullResponse)
+    return response
+  }
+
+  /** Method-key dispatch into the in-memory contract impl (a real carrier routes by URL path instead). */
+  private dispatch(method: keyof RpcMethodMap, request: RpcRequest<never>): Promise<RpcResponse<unknown>> {
+    switch (method) {
+      case 'session.list': return this.api.sessions.list(request)
+      case 'session.create': return this.api.sessions.create(request)
+      case 'session.history': return this.api.sessions.history(request)
+      case 'session.prompt': return this.api.sessions.prompt(request)
+      case 'session.cancel': return this.api.sessions.cancel(request)
+      case 'host.describe': return this.api.host.describe(request)
+    }
+  }
+
+  protected override openMux(
+    payload: { since?: Record<SessionId, number> },
+    signal: AbortSignal,
+    onOpen?: () => void,
+  ): AsyncIterable<RpcRequest<MuxFrame>> {
+    return this.tapStream(this.api.events.mux(rpcRequest(payload), signal), onOpen)
+  }
+
+  protected override openHost(
+    payload: Record<never, never>,
+    signal: AbortSignal,
+    onOpen?: () => void,
+  ): AsyncIterable<RpcRequest<HostFrame>> {
+    return this.tapStream(this.api.events.host(rpcRequest(payload), signal), onOpen)
+  }
+
+  private async *tapStream<F extends MuxFrame | HostFrame>(
+    stream: AsyncIterable<RpcRequest<F>>,
+    onOpen?: () => void,
+  ): AsyncGenerator<RpcRequest<F>> {
+    // No HTTP here: the in-memory stream is established the moment iteration starts (mirrors
+    // readSse firing onOpen after response headers, before any frame).
+    onOpen?.()
+    for await (const envelope of stream) {
+      const full: ServerRequest = { type: 'server-request', rpcId: envelope.rpcId, method: envelope.payload.type, payload: envelope.payload }
+      this.onEnvelope(full)
+      yield envelope
+    }
+  }
+
+  /**
+   * Deliver a client response to the in-memory contract impl (no HTTP POST),
+   * echoing the envelope to the observation tap like every other path.
+   * @param message - the client-response envelope answering a server request.
+   * @returns the carrier receipt from the fixture impl.
+   */
+  override async respond(message: ClientResponse): Promise<RpcReceipt> {
+    this.onEnvelope(message)
+    return this.api.respond(message)
+  }
+}

packages/client/connection/src/client/index.ts

@@ -0,0 +1,76 @@
+/**
+ * Browser half of the wire consumer layer (contract: api-contracts v3
+ * section 3; export inventory = v3 §3.2). The wire is this package's client
+ * half in its entirety — apply mounts ctx.connection: the shared api client
+ * plus the connection controller handle. Mode selection (?fixture) happens
+ * here so the rest of the client tree is mode-blind; the controller's sinks
+ * are wired by the runtime plugin (object layer), which injects this service.
+ */
+import type { Context } from 'cordis'
+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'
+
+// ---- Contract re-exports (browser-safe apiproxy channels + core types) ----
+export type {
+  ApiProxy, SessionsApi, SessionSummary, HostApi, EventsApi, MuxFrame, HostFrame,
+  ApprovalResponsePayload, QuestionResponsePayload, HistoryEntry, ToolEventView,
+  ToolCallView, ToolResultView,
+  RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
+  ClientRequest, ServerResponse, ServerRequest, ClientResponse, RpcMessage, RpcReceipt,
+  IApiClient, SessionId, SessionEvent, ContentBlock, StreamChunk,
+} from './api.ts'
+export { RpcId, AbstractApiClient, resultOf, transportError } from './api.ts'
+
+// ---- Connection loop ----
+export { ConnectionController } from './connection.ts'
+export type { ConnectionConfig, ConnectionSinks, ConnectionState }
+
+// ---- Platform client subclasses ----
+export { WebApiClient } from './web-api-client.ts'
+export { FixtureApiClient, createFixtureApi } from './fixture.ts'
+
+
+/** Required services (none — this is the wire root). */
+export const inject: string[] = []
+
+/**
+ * The ctx.connection service surface: the api client plus a one-shot
+ * controller starter (the runtime plugin supplies sinks when its object layer
+ * is ready — connection stays consumer-agnostic).
+ */
+export interface ConnectionHandle {
+  /** Shared api client (fixture or real, decided at boot from the page URL). */
+  readonly api: IApiClient
+  /**
+   * Start the connect/pump/reconnect loop with the consumer's frame sinks.
+   * One consumer owns the streams (the runtime object layer); a second call
+   * throws.
+   * @param sinks - frame/state callbacks.
+   * @param config - reconnect/backoff tunables.
+   * @returns stop handle for the loop.
+   */
+  start(sinks: ConnectionSinks, config?: ConnectionConfig): { stop(): void }
+}
+
+/**
+ * Client plugin body: pick the api by page mode and provide ctx.connection.
+ * @param ctx - client cordis context.
+ */
+export function apply(ctx: Context): void {
+  const fixture = typeof location !== 'undefined' && new URLSearchParams(location.search).has('fixture')
+  const api: IApiClient = fixture ? new FixtureApiClient() : new WebApiClient()
+  let started = false
+  const handle: ConnectionHandle = {
+    api,
+    start(sinks, config) {
+      if (started) throw new Error('connection: the stream loop is already owned by another consumer')
+      started = true
+      const controller = new ConnectionController(api, sinks, config ?? {})
+      controller.start()
+      return { stop: () => { controller.stop() } }
+    },
+  }
+  ctx.provide('connection', handle)
+}

packages/client/connection/src/client/web-api-client.ts

@@ -0,0 +1,12 @@
+// 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).
+
+import { AbstractApiClient } from './api.ts'
+
+/** Browser platform subclass: transport = global fetch over same-origin /api/*. */
+export class WebApiClient extends AbstractApiClient {
+  protected doFetch(input: URL, init?: RequestInit): Promise<Response> {
+    return globalThis.fetch(input, init)
+  }
+}

packages/client/connection/src/index.ts

@@ -0,0 +1,10 @@
+/**
+ * Connection plugin, node half. The package IS a dshClient plugin: the wire
+ * consumer layer lives in its client half in full (src/client/ — contract:
+ * api-contracts v3 section 3, inventory §3.2); consumers import the /client
+ * subpath. The empty apply exists so the plugin appears in the host Loader
+ * (lifecycle governance + dshClient discovery).
+ */
+
+/** Host plugin body — no host-side behavior for the connection plugin. */
+export function apply(_ctx: unknown): void {}

packages/client/connection/src/invariant.ts

@@ -0,0 +1,32 @@
+/**
+ * Package-owned invariant companion for `@deepseek-ai/dsh-client-connection`.
+ * @module @deepseek-ai/dsh-client-connection/invariant
+ */
+
+/* jscpd:ignore-start */
+import type { Context } from 'cordis'
+import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
+
+const PACKAGE_NAME = '@deepseek-ai/dsh-client-connection'
+
+/** Cordis companion plugin name. */
+export const name = 'client-connection-invariant'
+/** Service required before the companion can reserve package ownership. */
+export const inject = ['invariants']
+
+/**
+ * No runtime invariant: the pure wire layer emits no cordis events and owns no
+ * mutable cross-plugin relation — stream/reconnect sequencing is exercised
+ * directly by its behavior specs, and rpcId round-trip discipline is owned by
+ * the apiproxy contract layer.
+ */
+const install: InvariantInstaller = () => {}
+
+/**
+ * Register this package's invariant companion.
+ * @param ctx - Cordis context carrying the invariant service.
+ * @returns the installed registration's disposer after setup succeeds.
+ */
+export const apply = (ctx: Context): Promise<() => void> =>
+  Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
+/* jscpd:ignore-end */

packages/client/connection/tests/api-helpers.spec.ts

@@ -0,0 +1,21 @@
+/**
+ * Contract-layer helpers: transport-error folding and response unwrapping.
+ * (The assistant block classifier half of the legacy spec lives in
+ * runtime/tests — the classifier moved there.)
+ */
+
+import { describe, expect, it } from 'vitest'
+import { RpcId, resultOf, transportError } from '../src/client/api.ts'
+
+describe('transportError', () => {
+  it('folds an Error to internal keeping the message, and stringifies non-Errors', () => {
+    expect(transportError(new Error('线断了'))).toEqual({ ok: false, error: { code: 'internal', message: '线断了', details: {} } })
+    expect(transportError('raw string')).toMatchObject({ ok: false, error: { message: 'raw string' } })
+  })
+})
+
+describe('resultOf', () => {
+  it('unwraps the result slot', () => {
+    expect(resultOf({ rpcId: RpcId('r'), result: { ok: true, value: 7 } })).toEqual({ ok: true, value: 7 })
+  })
+})

packages/client/connection/tests/client-apply.spec.ts

@@ -0,0 +1,65 @@
+/**
+ * Connection plugin browser-half apply: ctx.connection handle mounting, mode
+ * selection off the page URL, and the single-consumer stream-loop ownership.
+ */
+import { Context } from 'cordis'
+import { afterEach, describe, expect, it } from 'vitest'
+import { apply, type ConnectionHandle } from '../src/client/index.ts'
+import { FixtureApiClient } from '../src/client/fixture.ts'
+import { WebApiClient } from '../src/client/web-api-client.ts'
+
+type Win = { location?: { search: string } }
+
+afterEach(() => {
+  delete (globalThis as Win).location
+})
+
+async function mount(): Promise<ConnectionHandle> {
+  const ctx = new Context()
+  await ctx.plugin({ apply, inject: [] })
+  const handle = ctx.get('connection') as ConnectionHandle | undefined
+  if (handle === undefined) throw new Error('ctx.connection not provided')
+  return handle
+}
+
+describe('connection client apply', () => {
+  it('mounts ctx.connection with the real client when no ?fixture switch is present', async () => {
+    ;(globalThis as Win).location = { search: '' }
+    const handle = await mount()
+    expect(handle.api).toBeInstanceOf(WebApiClient)
+  })
+
+  it('selects the fixture client under ?fixture (and with no location at all stays real)', async () => {
+    ;(globalThis as Win).location = { search: '?fixture' }
+    expect((await mount()).api).toBeInstanceOf(FixtureApiClient)
+    delete (globalThis as Win).location
+    expect((await mount()).api).toBeInstanceOf(WebApiClient)
+  })
+
+  it('start() hands out one loop, rejects a second consumer, and stop() aborts the streams', async () => {
+    ;(globalThis as Win).location = { search: '?fixture' }
+    const handle = await mount()
+    // config omitted: the `config ?? {}` default arm is part of the surface.
+    const loop = handle.start({})
+    expect(() => handle.start({})).toThrow(/already owned by another consumer/)
+    loop.stop() // teardown must not throw; the fixture streams abort quietly
+  })
+
+  it('WebApiClient carries requests over globalThis.fetch', async () => {
+    ;(globalThis as Win).location = { search: '' }
+    const handle = await mount()
+    const original = globalThis.fetch
+    const seen: string[] = []
+    globalThis.fetch = (input: URL | RequestInfo) => {
+      seen.push(typeof input === 'string' ? input : input instanceof URL ? input.href : input.url)
+      return Promise.resolve(new Response('{}', { status: 200 }))
+    }
+    try {
+      // Schema rejection is fine — the transport hop is the assertion.
+      await (handle.api as WebApiClient).host.describe({}).catch(() => undefined)
+    } finally {
+      globalThis.fetch = original
+    }
+    expect(seen.some(u => u.includes('/api/'))).toBe(true)
+  })
+})

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

@@ -0,0 +1,238 @@
+/**
+ * ConnectionController: stream pumping into sinks, the strict readiness
+ * handshake (describe + both streams' onOpen, timeout-guarded), generation
+ * abort on loss, backoff reconnection, state transitions, and sink-exception
+ * isolation. Real (short) timers — the timeout and backoff are configurable,
+ * so tests run them at millisecond scale.
+ */
+
+import { describe, expect, it, vi } from 'vitest'
+import type { SessionId } from '../src/client/api.ts'
+import type { ConnectionState } from '../src/client/connection.ts'
+import { ConnectionController } from '../src/client/connection.ts'
+import { FakeApiClient, deferred, ok } from './fake-api.ts'
+
+const SID = 'fk-c1' as SessionId
+const FAST = { backoffBaseMs: 10, backoffFactor: 1, backoffMaxMs: 10, streamOpenTimeoutMs: 500 }
+
+function subscribedFrame(lastSeq = 0) {
+  return { type: 'session/subscribed', sessionId: SID, lastSeq } as const
+}
+
+describe('connection lifecycle', () => {
+  it('announces connected after describe + both streams open, then pumps frames to sinks', async () => {
+    const api = new FakeApiClient()
+    const muxSeen: string[] = []
+    let connected = 0
+    const controller = new ConnectionController(api, {
+      onMuxEnvelope: envelope => muxSeen.push(envelope.payload.type),
+      onConnected: () => { connected++ },
+    }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      api.pushMux(subscribedFrame())
+      await vi.waitFor(() => { expect(muxSeen).toEqual(['session/subscribed']) })
+      expect(api.callsOf('host.describe')).toHaveLength(1)
+    } finally {
+      controller.stop()
+    }
+  })
+
+  it('reconnects with a fresh generation when a stream fails, and stop() ends the loop', async () => {
+    const api = new FakeApiClient()
+    let connected = 0
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      api.failStreams(new Error('stream torn'))
+      await vi.waitFor(() => { expect(connected).toBe(2) }) // new generation after backoff
+      expect(api.openMuxCount).toBe(1) // the dead generation's stream is gone, exactly one live
+    } finally {
+      controller.stop()
+      warnSpy.mockRestore()
+    }
+    // stop() aborts the live generation (streams tear down) and no reconnect follows.
+    await vi.waitFor(() => { expect(api.openMuxCount).toBe(0) })
+    await new Promise(resolve => setTimeout(resolve, 40))
+    expect(api.openMuxCount).toBe(0)
+  })
+
+  it('treats describe failure as generation failure and retries', async () => {
+    const api = new FakeApiClient()
+    const gate = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
+    let describeCalls = 0
+    api.onDescribe = () => {
+      describeCalls++
+      return describeCalls === 1 ? Promise.reject(new Error('host down')) : gate.promise
+    }
+    let connected = 0
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(describeCalls).toBe(2) }) // retried after backoff
+      expect(connected).toBe(0) // never announced during the failed generation
+      gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+    } finally {
+      controller.stop()
+      warnSpy.mockRestore()
+    }
+  })
+
+  it('converges stream/error frames into reconnect instead of dispatching them', async () => {
+    const api = new FakeApiClient()
+    const muxSeen: string[] = []
+    let connected = 0
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const controller = new ConnectionController(api, {
+      onMuxEnvelope: envelope => muxSeen.push(envelope.payload.type),
+      onConnected: () => { connected++ },
+    }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      api.pushMux({ type: 'stream/error', error: { code: 'internal', message: 'impl broke', details: {} } })
+      await vi.waitFor(() => { expect(connected).toBe(2) }) // treated as loss → reconnect
+      expect(muxSeen).toEqual([]) // never forwarded to the business sink
+    } finally {
+      controller.stop()
+      warnSpy.mockRestore()
+    }
+  })
+
+  it('isolates sink exceptions from the pump', async () => {
+    const api = new FakeApiClient()
+    const seen: string[] = []
+    let connected = 0
+    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined)
+    const controller = new ConnectionController(api, {
+      onMuxEnvelope: (envelope) => {
+        seen.push(envelope.payload.type)
+        throw new Error('business layer bug')
+      },
+      onConnected: () => { connected++ },
+    }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      api.pushMux(subscribedFrame(1))
+      api.pushMux(subscribedFrame(2))
+      await vi.waitFor(() => { expect(seen).toHaveLength(2) }) // second frame still pumped
+      expect(connected).toBe(1) // no reconnect triggered by the sink throw
+    } finally {
+      controller.stop()
+      errorSpy.mockRestore()
+    }
+  })
+
+  it('holds onConnected until both streams establish even after describe succeeds', async () => {
+    const api = new FakeApiClient()
+    api.holdStreamOpen = true // describe resolves immediately; stream establishment is in the case's hand
+    let connected = 0
+    const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(api.callsOf('host.describe')).toHaveLength(1) })
+      await new Promise(resolve => setTimeout(resolve, 30))
+      expect(connected).toBe(0) // describe alone must not announce
+      api.releaseStreamOpens()
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+    } finally {
+      controller.stop()
+    }
+  })
+
+  it('proceeds as connected via the timeout guard when a carrier never fires onOpen', async () => {
+    const api = new FakeApiClient()
+    api.suppressStreamOpen = true // misbehaving carrier: streams open but onOpen never fires
+    let connected = 0
+    const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, { ...FAST, streamOpenTimeoutMs: 20 })
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(connected).toBe(1) }) // handshake resolved by the guard, not wedged
+    } finally {
+      controller.stop()
+    }
+  })
+
+  it('emits deduplicated connected/reconnecting state transitions', async () => {
+    const api = new FakeApiClient()
+    const states: ConnectionState[] = []
+    let connected = 0
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const controller = new ConnectionController(api, {
+      onConnected: () => { connected++ },
+      onStateChange: state => states.push(state),
+    }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      expect(states).toEqual(['connected'])
+      api.failStreams(new Error('torn'))
+      await vi.waitFor(() => { expect(connected).toBe(2) })
+      expect(states).toEqual(['connected', 'reconnecting', 'connected'])
+    } finally {
+      controller.stop()
+      warnSpy.mockRestore()
+    }
+  })
+
+  it('deduplicates consecutive reconnecting emissions across two straight failures', async () => {
+    const api = new FakeApiClient()
+    const gate = deferred<Awaited<ReturnType<FakeApiClient['onDescribe']>>>()
+    let describeCalls = 0
+    api.onDescribe = () => {
+      describeCalls++
+      return describeCalls <= 2 ? Promise.reject(new Error('down')) : gate.promise
+    }
+    const states: ConnectionState[] = []
+    let connected = 0
+    const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    const controller = new ConnectionController(api, {
+      onConnected: () => { connected++ },
+      onStateChange: state => states.push(state),
+    }, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(describeCalls).toBe(3) })
+      gate.resolve(ok({ version: '0', cwd: '/f', attachedSessions: 0 }))
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      expect(states).toEqual(['reconnecting', 'connected']) // two failures, one reconnecting emission
+    } finally {
+      controller.stop()
+      warnSpy.mockRestore()
+    }
+  })
+
+  it('runs with no sinks at all (every callback slot optional)', async () => {
+    const api = new FakeApiClient()
+    const controller = new ConnectionController(api, {}, FAST)
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(api.callsOf('host.describe')).toHaveLength(1) })
+      api.pushMux(subscribedFrame()) // pumped with sink undefined: dropped silently
+      await new Promise(resolve => setTimeout(resolve, 20))
+    } finally {
+      controller.stop()
+    }
+  })
+
+  it('start() is idempotent (one loop, one stream set)', async () => {
+    const api = new FakeApiClient()
+    let connected = 0
+    const controller = new ConnectionController(api, { onConnected: () => { connected++ } }, FAST)
+    controller.start()
+    controller.start()
+    try {
+      await vi.waitFor(() => { expect(connected).toBe(1) })
+      expect(api.openMuxCount).toBe(1)
+      expect(api.callsOf('host.describe')).toHaveLength(1)
+    } finally {
+      controller.stop()
+    }
+  })
+})

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

@@ -0,0 +1,160 @@
+// Test-local programmable IApiClient fake (NOT the fixture: fixture is a demo
+// data source on a real clock; behavior tests need per-case responses and
+// deferred-controlled timing). Streams are hand pumps: pushMux/pushHost.
+import type {
+  HostFrame, IApiClient, MuxFrame, RpcRequest, RpcResponse, SessionId,
+} from '../src/client/api.ts'
+import { RpcId } from '../src/client/api.ts'
+
+export interface Deferred<T> {
+  promise: Promise<T>
+  resolve(value: T): void
+  reject(error: unknown): void
+}
+
+/** Test-held settlement: the case decides when an RPC lands (history-pending injections etc.). */
+export function deferred<T>(): Deferred<T> {
+  let resolve!: (value: T) => void
+  let reject!: (error: unknown) => void
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res
+    reject = rej
+  })
+  return { promise, resolve, reject }
+}
+
+let nextRpc = 0
+
+export function ok<T>(value: T): RpcResponse<T> {
+  return { rpcId: RpcId(`fake-${nextRpc++}`), result: { ok: true, value } }
+}
+
+
+type StreamItem<F> = { kind: 'frame'; envelope: RpcRequest<F> } | { kind: 'end' } | { kind: 'fail'; error: unknown }
+
+interface StreamConn<F> {
+  feed(item: StreamItem<F>): void
+}
+
+export class FakeApiClient implements IApiClient {
+  /** Chronological call record: [method, payload]. */
+  readonly calls: { method: string; payload: unknown }[] = []
+
+  // Programmable slots (defaults answer OK-empty); reassign per case.
+  onList: (payload: unknown) => Promise<RpcResponse<{ items: never[] }>> = () => Promise.resolve(ok({ items: [] }))
+  onCreate: (payload: unknown) => Promise<RpcResponse<{ sessionId: SessionId }>> = () => Promise.resolve(ok({ sessionId: 'fk-new' as SessionId }))
+  onHistory: (payload: { sessionId: SessionId; beforeSeq?: number; maxMessages?: number })
+  => Promise<RpcResponse<{ events: never[]; hasMore: boolean }>> =
+    () => Promise.resolve(ok({ events: [], hasMore: false }))
+
+  onPrompt: (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 }))
+
+  private readonly muxConns: StreamConn<MuxFrame>[] = []
+  private readonly hostConns: StreamConn<HostFrame>[] = []
+
+  // 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)),
+    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)),
+    prompt: (payload: unknown) => this.record('session.prompt', payload, this.onPrompt(payload)),
+    cancel: (payload: unknown) => this.record('session.cancel', payload, this.onCancel(payload)),
+  }
+
+  readonly host: IApiClient['host'] = {
+    describe: payload => this.record('host.describe', payload, this.onDescribe(payload)),
+  }
+
+  /** When true, streams never fire onOpen (misbehaving-carrier material for the handshake timeout guard). */
+  suppressStreamOpen = false
+
+  /** When true, onOpen callbacks are parked instead of fired; releaseStreamOpens() fires them.
+   *  Lets a case hold the readiness handshake open (describe done, streams not yet "established"). */
+  holdStreamOpen = false
+  private heldOpens: (() => void)[] = []
+
+  releaseStreamOpens(): void {
+    const held = this.heldOpens
+    this.heldOpens = []
+    for (const fire of held) fire()
+  }
+
+  readonly events: IApiClient['events'] = {
+    mux: (_payload: unknown, signal: AbortSignal, onOpen?: () => void) =>
+      this.openStream(this.muxConns, signal, onOpen),
+    host: (_payload: unknown, signal: AbortSignal, onOpen?: () => void) =>
+      this.openStream(this.hostConns, signal, onOpen),
+  }
+
+  respond(): Promise<{ accepted: false; reason: 'not-pending' }> {
+    return Promise.resolve({ accepted: false, reason: 'not-pending' })
+  }
+
+  /** Push one mux frame to every open mux stream (rpcId minted unless pinned by the case). */
+  pushMux(frame: MuxFrame, rpcId?: string): void {
+    for (const conn of [...this.muxConns]) conn.feed({ kind: 'frame', envelope: { rpcId: RpcId(rpcId ?? `push-${nextRpc++}`), payload: frame } })
+  }
+
+  pushHost(frame: HostFrame, rpcId?: string): void {
+    for (const conn of [...this.hostConns]) conn.feed({ kind: 'frame', envelope: { rpcId: RpcId(rpcId ?? `push-${nextRpc++}`), payload: frame } })
+  }
+
+  /** End (clean close) or fail (throw) every open stream — reconnect-path material. */
+  endStreams(): void {
+    for (const conn of [...this.muxConns, ...this.hostConns]) conn.feed({ kind: 'end' })
+  }
+
+  failStreams(error: unknown): void {
+    for (const conn of [...this.muxConns, ...this.hostConns]) conn.feed({ kind: 'fail', error })
+  }
+
+  get openMuxCount(): number {
+    return this.muxConns.length
+  }
+
+  callsOf(method: string): unknown[] {
+    return this.calls.filter(c => c.method === method).map(c => c.payload)
+  }
+
+  private record<T>(method: string, payload: unknown, response: Promise<T>): Promise<T> {
+    this.calls.push({ method, payload })
+    return response
+  }
+
+  private async *openStream<F>(registry: StreamConn<F>[], signal: AbortSignal, onOpen?: () => void): AsyncGenerator<RpcRequest<F>> {
+    const inbox: StreamItem<F>[] = []
+    let wake: (() => void) | null = null
+    const conn: StreamConn<F> = {
+      feed: (item) => {
+        inbox.push(item)
+        wake?.()
+      },
+    }
+    registry.push(conn)
+    if (this.holdStreamOpen && onOpen !== undefined) this.heldOpens.push(onOpen)
+    else if (!this.suppressStreamOpen) onOpen?.()
+    try {
+      while (!signal.aborted) {
+        while (inbox.length > 0) {
+          const item = inbox.shift() as StreamItem<F>
+          if (item.kind === 'end') return
+          if (item.kind === 'fail') throw item.error
+          yield item.envelope
+        }
+        await new Promise<void>((resolve) => {
+          wake = resolve
+          signal.addEventListener('abort', () => { resolve() }, { once: true })
+        })
+        wake = null
+      }
+    } finally {
+      registry.splice(registry.indexOf(conn), 1)
+    }
+  }
+}

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

@@ -0,0 +1,337 @@
+/**
+ * Fixture impl semantics: the demo data source must honor the same contract
+ * shapes as the real host (paging boundaries, rpcId echo, replay lifecycle,
+ * baseline replay, timing hooks) — this is the vitest-side drift detector for
+ * the hand-written fixture/host parallel implementations.
+ */
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { SessionId } from '../src/client/api.ts'
+import { RpcId } from '../src/client/api.ts'
+import type { HostFrame, MuxFrame, RpcMessage, RpcRequest } from '../src/client/api.ts'
+import { FixtureApiClient, createFixtureApi } from '../src/client/fixture.ts'
+
+const sid = (id: string): SessionId => id as SessionId
+const req = <P>(payload: P): RpcRequest<P> => ({ rpcId: RpcId(`t-${Math.abs(Math.sin(reqCount++)).toString(36).slice(2, 10)}`), payload })
+let reqCount = 0
+
+interface TimingHooks {
+  setHistoryDelay(ms: number): void
+  failNextHistory(): void
+  appendUser(id: string, msg: string): void
+  appendSilent(id: string, msg: string): void
+  breakStreams(): void
+}
+const timing = (): TimingHooks => (globalThis as Record<string, unknown>).__fxTiming as TimingHooks
+
+/** Collect stream frames until the predicate or a soft cap; abort ends the stream. */
+async function collect<F>(stream: AsyncIterable<RpcRequest<F>>, abort: AbortController, done: (frames: F[]) => boolean): Promise<F[]> {
+  const frames: F[] = []
+  for await (const envelope of stream) {
+    frames.push(envelope.payload)
+    if (done(frames) || frames.length > 500) {
+      abort.abort()
+      break
+    }
+  }
+  return frames
+}
+
+describe('createFixtureApi', () => {
+  it('serves the session list sorted by updatedAt desc and echoes rpcIds on every unary', async () => {
+    const api = createFixtureApi()
+    const request = req({})
+    const response = await api.sessions.list(request)
+    expect(response.rpcId).toBe(request.rpcId)
+    if (!response.result.ok) throw new Error('list failed')
+    expect(response.result.value.items.map(s => s.sessionId)).toEqual(['fx-alpha', 'fx-beta', 'fx-gamma'])
+    expect(response.result.value.items[1]?.parentSessionId).toBe('fx-alpha') // lineage material
+  })
+
+  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 }))
+    if (!tail.result.ok) throw new Error('history failed')
+    const tailPage = tail.result.value
+    expect(tailPage.hasMore).toBe(true)
+    expect(tailPage.events[0]?.event.type).toBe('turn/start') // cut lands on a turn boundary
+    const boundary = tailPage.events[0]?.event.seq ?? 0
+    expect(boundary).toBeGreaterThan(0)
+    const older = await api.sessions.history(req({ sessionId: sid('fx-alpha'), beforeSeq: boundary, maxMessages: 10 }))
+    if (!older.result.ok) throw new Error('older failed')
+    const olderTail = older.result.value.events.at(-1)?.event
+    expect((olderTail?.seq ?? -1) + 1).toBe(boundary) // pages stitch with no hole/overlap
+    // Out-of-range beforeSeq clamps instead of exploding.
+    const clamped = await api.sessions.history(req({ sessionId: sid('fx-alpha'), beforeSeq: -5, maxMessages: 10 }))
+    if (!clamped.result.ok) throw new Error('clamped failed')
+    expect(clamped.result.value.events).toEqual([])
+    // Unknown session: empty page, not an error (history of a bare id).
+    const empty = await api.sessions.history(req({ sessionId: sid('no-such'), maxMessages: 10 }))
+    if (!empty.result.ok) throw new Error('empty failed')
+    expect(empty.result.value).toEqual({ events: [], hasMore: false })
+  })
+
+  it('create adds a session and pushes host/session-added to open host streams', async () => {
+    const api = createFixtureApi()
+    const abort = new AbortController()
+    const seen: HostFrame[] = []
+    const consuming = (async () => {
+      for await (const envelope of api.events.host(req({}), abort.signal)) {
+        seen.push(envelope.payload)
+        if (seen.length >= 1) abort.abort()
+      }
+    })()
+    await new Promise(resolve => setTimeout(resolve, 10)) // let the stream register
+    const created = await api.sessions.create(req({}))
+    if (!created.result.ok) throw new Error('create failed')
+    await consuming
+    if (!created.result.ok) throw new Error('create failed')
+    const createdId = created.result.value.sessionId
+    expect(seen).toEqual([{ type: 'host/session-added', sessionId: createdId }])
+    const list = await api.sessions.list(req({}))
+    if (!list.result.ok) throw new Error('list failed')
+    expect(list.result.value.items.some(s => s.sessionId === createdId)).toBe(true)
+  })
+
+  it('prompt replays a full streamed turn and cancel mid-replay freezes with (已中断)', async () => {
+    const api = createFixtureApi()
+    const created = await api.sessions.create(req({}))
+    if (!created.result.ok) throw new Error('create failed')
+    const id = created.result.value.sessionId
+    const abort = new AbortController()
+    const frames: MuxFrame[] = []
+    const consuming = (async () => {
+      for await (const envelope of api.events.mux(req({}), abort.signal)) {
+        frames.push(envelope.payload)
+        const last = envelope.payload
+        if (last.type === 'session/event' && last.event.type === 'turn/end') {
+          abort.abort()
+        }
+      }
+    })()
+    await new Promise(resolve => setTimeout(resolve, 10))
+    // Unknown session → session-not-found with the id echoed in details.
+    const missing = await api.sessions.prompt(req({ sessionId: sid('ghost'), mode: 'queue' as const, content: [{ type: 'text' as const, text: 'x' }] }))
+    expect(missing.result).toMatchObject({ ok: false, error: { code: 'session-not-found', details: { sessionId: 'ghost' } } })
+    // Real prompt: replay starts (running flips true), cancel freezes it.
+    const accepted = await api.sessions.prompt(req({ sessionId: id, mode: 'queue' as const, content: [{ type: 'text' as const, text: '取消我' }] }))
+    expect(accepted.result).toMatchObject({ ok: true, value: { accepted: true } })
+    await new Promise(resolve => setTimeout(resolve, 120)) // a couple of typewriter ticks
+    await api.sessions.cancel(req({ sessionId: id }))
+    await consuming
+    const types = frames.filter((f): f is Extract<MuxFrame, { type: 'session/event' }> => f.type === 'session/event').map(f => f.event.type)
+    expect(types).toContain('turn/start')
+    expect(types).toContain('user/message')
+    expect(types).toContain('assistant/chunk')
+    expect(types).toContain('assistant/message')
+    expect(types.at(-1)).toBe('turn/end')
+    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.
+    const idleCancel = await api.sessions.cancel(req({ sessionId: id }))
+    expect(idleCancel.result).toMatchObject({ ok: true })
+  })
+
+  it('steer during a replay inserts a steering message and the replay continues to completion', async () => {
+    const api = createFixtureApi()
+    const created = await api.sessions.create(req({}))
+    if (!created.result.ok) throw new Error('create failed')
+    const id = created.result.value.sessionId
+    const abort = new AbortController()
+    const framesPromise = collect<MuxFrame>(api.events.mux(req({}), abort.signal), abort,
+      frames => frames.some(f => f.type === 'session/event' && f.event.type === 'turn/end'))
+    await new Promise(resolve => setTimeout(resolve, 10))
+    await api.sessions.prompt(req({ sessionId: id, mode: 'queue' as const, content: [{ type: 'text' as const, text: '短' }] }))
+    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(types.at(-1)).toBe('turn/end') // steer did not restart the turn
+  })
+
+  it('mux open replays the baseline: subscribed for running sessions + the resident approval with a stable rpcId', async () => {
+    const api = createFixtureApi()
+    const openOnce = async (): Promise<RpcRequest<MuxFrame>[]> => {
+      const abort = new AbortController()
+      const envelopes: RpcRequest<MuxFrame>[] = []
+      for await (const envelope of api.events.mux(req({}), abort.signal)) {
+        envelopes.push(envelope)
+        if (envelopes.length >= 2) abort.abort()
+      }
+      return envelopes
+    }
+    const first = await openOnce()
+    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)
+    expect(first[1]?.payload).toMatchObject({ type: 'approval/requested', toolName: 'dangerous_tool' })
+    expect(second[1]?.rpcId).toBe(first[1]?.rpcId) // stable rpcId across replays (host replay semantics)
+  })
+
+  it('steer with no replay in flight falls through to a fresh queued turn; non-text blocks stringify empty', async () => {
+    const api = createFixtureApi()
+    const abort = new AbortController()
+    const framesPromise = collect<MuxFrame>(api.events.mux(req({}), abort.signal), abort,
+      frames => frames.some(f => f.type === 'session/event' && f.event.type === 'turn/end'))
+    await new Promise(resolve => setTimeout(resolve, 10))
+    const created = await api.sessions.create(req({}))
+    if (!created.result.ok) throw new Error('create failed')
+    // steer while idle + a non-text content block (covers the '' arm of the text join).
+    await api.sessions.prompt(req({
+      sessionId: created.result.value.sessionId, mode: 'steer' as const,
+      content: [{ type: 'text' as const, text: '短' }, { type: 'image', data: 'x' } as never],
+    }))
+    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
+  })
+
+  it('gamma interval flip emits host/session-status and a running log-less session subscribes at lastSeq -1', async () => {
+    vi.useFakeTimers()
+    try {
+      const api = createFixtureApi()
+      const abort = new AbortController()
+      const hostSeen: HostFrame[] = []
+      const consuming = (async () => {
+        for await (const envelope of api.events.host(req({}), abort.signal)) hostSeen.push(envelope.payload)
+      })()
+      await vi.advanceTimersByTimeAsync(5001) // interval fires: fx-gamma flips running=true (no log exists)
+      expect(hostSeen).toContainEqual({ type: 'host/session-status', sessionId: sid('fx-gamma'), running: true })
+      // A mux stream opened now sees gamma in the baseline with lastSeq = -1 (empty log arm).
+      const mabort = new AbortController()
+      const baseline: MuxFrame[] = []
+      const muxConsuming = (async () => {
+        for await (const envelope of api.events.mux(req({}), mabort.signal)) {
+          baseline.push(envelope.payload)
+          if (baseline.length >= 3) mabort.abort()
+        }
+      })()
+      await vi.advanceTimersByTimeAsync(10)
+      mabort.abort()
+      await muxConsuming
+      expect(baseline).toContainEqual({ type: 'session/subscribed', sessionId: sid('fx-gamma'), lastSeq: -1 })
+      abort.abort()
+      await vi.advanceTimersByTimeAsync(10)
+      await consuming
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('respond is a typed stub: always not-pending', async () => {
+    const api = createFixtureApi()
+    expect(await api.respond({ type: 'client-response', rpcId: RpcId('x'), result: { ok: true, value: {} } })).toEqual({ accepted: false, reason: 'not-pending' })
+  })
+
+  it('describe answers the fixture identity', async () => {
+    const api = createFixtureApi()
+    const response = await api.host.describe(req({}))
+    expect(response.result).toMatchObject({ ok: true, value: { version: '0.0.0-fixture', attachedSessions: 1 } })
+  })
+
+  it('timing hooks: history delay + one-shot failure, silent append, and breakStreams end open generators', async () => {
+    const api = createFixtureApi()
+    const hooks = timing()
+    // One-shot transport failure after transit delay.
+    hooks.setHistoryDelay(5)
+    hooks.failNextHistory()
+    await expect(api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 5 }))).rejects.toThrow(/simulated history transport failure/)
+    hooks.setHistoryDelay(0)
+    // The failure was one-shot: the next call succeeds.
+    const ok = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 5 }))
+    expect(ok.result.ok).toBe(true)
+    // appendUser emits on the mux stream; appendSilent only lands in the log (lost frame).
+    const abort = new AbortController()
+    const seen: MuxFrame[] = []
+    const consuming = (async () => {
+      for await (const envelope of api.events.mux(req({}), abort.signal)) seen.push(envelope.payload)
+    })()
+    await new Promise(resolve => setTimeout(resolve, 10))
+    hooks.appendSilent('fx-alpha', '静默丢帧')
+    hooks.appendUser('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' && JSON.stringify(f.event.data).includes('静默丢帧'))).toBe(false)
+    // But history serves the silent event (the client's repull finds it).
+    const repull = await api.sessions.history(req({ sessionId: sid('fx-alpha'), maxMessages: 5 }))
+    if (!repull.result.ok) throw new Error('repull failed')
+    expect(JSON.stringify(repull.result.value.events)).toContain('静默丢帧')
+    // breakStreams force-ends BOTH stream kinds without the client abort.
+    const habort = new AbortController()
+    const hostConsuming = (async () => {
+      for await (const _ of api.events.host(req({}), habort.signal)) { /* drain */ }
+    })()
+    await new Promise(resolve => setTimeout(resolve, 10))
+    hooks.breakStreams()
+    await consuming // returns because the stream broke, not because we aborted
+    await hostConsuming
+    expect(abort.signal.aborted).toBe(false)
+    expect(habort.signal.aborted).toBe(false)
+  })
+})
+
+describe('FixtureApiClient (protocol-level fake carrier)', () => {
+  afterEach(() => {
+    vi.restoreAllMocks()
+  })
+
+  it('doFetch is an unreachable tripwire (all protocol paths overridden)', () => {
+    const client = new FixtureApiClient()
+    // Protected at compile time only; reach it directly to pin the tripwire message.
+    expect(() => (client as unknown as { doFetch(): Promise<Response> }).doFetch()).toThrow(/doFetch must be unreachable/)
+  })
+
+  it('mints request ids, taps all four full forms, and never touches doFetch', async () => {
+    const client = new FixtureApiClient()
+    const tapped: RpcMessage[] = []
+    client.subscribeEnvelopes(batch => tapped.push(...batch))
+    const response = await client.sessions.list({})
+    expect(response.result.ok).toBe(true)
+    await client.respond({ type: 'client-response', rpcId: RpcId('r-x'), result: { ok: true, value: {} } })
+    await vi.waitFor(() => {
+      const kinds = tapped.map(m => m.type)
+      expect(kinds).toContain('client-request')
+      expect(kinds).toContain('server-response')
+      expect(kinds).toContain('client-response')
+    })
+    const request = tapped.find(m => m.type === 'client-request')
+    const reply = tapped.find(m => m.type === 'server-response')
+    expect(request?.rpcId).toBe(reply?.rpcId) // echo discipline holds through the fake carrier
+  })
+
+  it('covers the whole unary dispatch table', async () => {
+    const client = new FixtureApiClient()
+    const created = await client.sessions.create({})
+    if (!created.result.ok) throw new Error('create failed')
+    const id = created.result.value.sessionId
+    expect((await client.sessions.history({ sessionId: id })).result.ok).toBe(true)
+    expect((await client.sessions.prompt({ sessionId: id, mode: 'queue', content: [{ type: 'text', text: '嗨' }] })).result.ok).toBe(true)
+    expect((await client.sessions.cancel({ sessionId: id })).result.ok).toBe(true)
+    expect((await client.host.describe({})).result.ok).toBe(true)
+  })
+
+  it('fires onOpen at stream-iteration start and taps server-request full forms', async () => {
+    const client = new FixtureApiClient()
+    const tapped: RpcMessage[] = []
+    client.subscribeEnvelopes(batch => tapped.push(...batch))
+    const order: string[] = []
+    const abort = new AbortController()
+    for await (const envelope of client.events.mux({}, abort.signal, () => order.push('open'))) {
+      order.push(envelope.payload.type)
+      abort.abort()
+    }
+    expect(order[0]).toBe('open')
+    expect(order[1]).toBe('session/subscribed')
+    await vi.waitFor(() => {
+      expect(tapped.some(m => m.type === 'server-request')).toBe(true)
+    })
+    // Host stream side of the pair (same tap path).
+    const habort = new AbortController()
+    const hostOrder: string[] = []
+    const hostIterator = client.events.host({}, habort.signal, () => hostOrder.push('open'))[Symbol.asyncIterator]()
+    const raced = await Promise.race([hostIterator.next(), new Promise<'idle'>(resolve => setTimeout(() => { resolve('idle') }, 50))])
+    expect(hostOrder).toEqual(['open']) // established even though the host stream stays silent
+    habort.abort()
+    if (raced === 'idle') await hostIterator.return?.(undefined)
+  })
+})

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

@@ -0,0 +1,10 @@
+/** Node half: the empty host apply (Loader governance + dshClient discovery placeholder). */
+import { describe, expect, it } from 'vitest'
+import { apply } from '../src/index.ts'
+
+describe('node half', () => {
+  it('apply is a no-op host placeholder', () => {
+    apply(undefined)
+    expect(true).toBe(true) // reaching here without throw is the contract
+  })
+})

packages/client/connection/tsconfig.json

@@ -0,0 +1,42 @@
+{
+  "extends": "../../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib/types",
+    "lib": [
+      "ES2024",
+      "DOM",
+      "DOM.Iterable"
+    ],
+    "types": []
+  },
+  "include": [
+    "src"
+  ],
+  "references": [
+    {
+      "path": "../../llm/llm"
+    },
+    {
+      "path": "../../core/session"
+    },
+    {
+      "path": "../../util/brand"
+    },
+    {
+      "path": "../../host/apiproxy"
+    },
+    {
+      "path": "../../ui/user-approval"
+    },
+    {
+      "path": "../../ui/user-interaction"
+    },
+    {
+      "path": "../../support/invariants"
+    }
+  ],
+  "exclude": [
+    "**/*.legacy.*"
+  ]
+}

packages/client/connection/tsdown.config.ts

@@ -0,0 +1,3 @@
+import { clientBundle } from '../tsdown.client.ts'
+
+export default clientBundle('@deepseek-ai/dsh-client-connection', ['lib/types/index.js', 'lib/types/invariant.js'])