Merge remote-tracking branch 'origin/master' into jsonl-packed-chunk-rows

Conflict: the generated event-producer-consumer doc — regenerated over merged
sources. Also strips the trailing whitespace / extra EOF blank line that
master's vitest.config.ts and client api.ts arrived with (the pre-commit
whitespace hook rejects any commit touching them otherwise).

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 */