Merge PR #500 head into all-gates recovery

# Conflicts:
#	package.json
#	scripts/run-gates.ts

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,339 @@
+/**
+ * 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
+  })
+})