DeepSeek Harness LLM 与流式：用适配器隔离提供方差异

  /**
   * Stream one model call as raw chunks. The only required method.
   * @param options - the fully-assembled request; implementations must honor `options.signal`.
   * @returns the chunk stream, obeying the adapter contract documented on `StreamChunk`.
   */
  abstract stream(options: GenerateOptions): AsyncIterable<StreamChunk>

  /**
   * Register an adapter for the given provider routes. Throws `LlmError` with code
   * `DUPLICATE_ADAPTER` if any provider already has an adapter (all-or-nothing).
   * Disposed with the fiber.
   * @param providers - every provider route this adapter should serve.
   * @param adapter - the adapter that streams calls for those providers.
   * @returns the disposer, carrying {@link AdapterRegistrationHandle.replace}.
   */
  registerAdapter(providers: string[], adapter: LlmAdapter): AdapterRegistrationHandle {
    // The routes this registration currently holds; `replace` rewrites it, and
    // the disposer releases whatever it holds at disposal time.
    const owned = new Set<string>()
    // The disposer has run: `owned` being empty cannot say so on its own,
    // because `replace([])` legally leaves a live registration holding none.
    let released = false
    const dispose = this.ctx.effect(function* (this: LlmRuntime) {
      if (providers.length === 0) throw new LlmError('an adapter must register at least one provider', 'INVALID_ADAPTER')
      this.commitRoutes(owned, this.prepareRoutes(providers, adapter, owned))
      yield () => {
        released = true
        for (const provider of owned) this.adapters.delete(provider)
        owned.clear()
        this.emitAdaptersUpdated()
      }
    }.bind(this), 'llm.registerAdapter()')
    // ctx.effect's disposer returns Promise<void>; our disposer API is
    // synchronous fire-and-forget — discard the (always-resolved) promise.
    const handle = (() => void dispose()) as AdapterRegistrationHandle
    handle.replace = (next: string[]): void => {
      // Registering here would leak: the effect's disposer already ran, so
      // nothing remains to release whatever this call would put in the map.
      if (released) {
        throw new LlmError('a disposed adapter registration cannot replace its routes', 'REGISTRATION_DISPOSED')
      }
      this.commitRoutes(owned, this.prepareRoutes(next, adapter, owned))
    }
    return handle
  }

/**
 * A5 LLM 与流式实证：适配器 stream 契约 + BlockAssembler 组装
 * 作者本机运行（cwd=E:/coding/deepseek-harness）：
 * node --import tsx E:/coding/rex-hugo/.tmp-research/dsh-llm-demo/adapter-stream-demo.ts
 */
import { Context } from '@deepseek-ai/cordis'
import {
  BlockAssembler, CallId, LlmAdapter, LlmRuntime,
  type GenerateOptions, type StreamChunk,
} from '@deepseek-ai/dsh-llm'

/** 最小适配器：继承 LlmAdapter，唯一必选方法是 stream() */
class MockAdapter extends LlmAdapter {
  constructor(private flavor: string) {
    super()
  }

  providerInfo(provider: string) {
    return { id: provider, name: `${provider} (${this.flavor})` }
  }

  async *stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
    // 1. 文本块：block-start -> text-delta* -> block-end
    yield { type: 'block-start', index: 0, blockType: 'text' }
    yield { type: 'text-delta', index: 0, text: `你好，我是 ${this.flavor} 模型。` }
    yield { type: 'text-delta', index: 0, text: ' 我会流式输出。' }
    yield {
      type: 'block-end', index: 0,
      block: { type: 'text', text: `你好，我是 ${this.flavor} 模型。 我会流式输出。` },
    }
    // 2. 工具调用块：argumentsDelta 可以分多个 chunk 增量到达
    yield { type: 'block-start', index: 1, blockType: 'tool-call' }
    yield { type: 'tool-call-delta', index: 1, id: CallId('c1'), name: 'read_file', argumentsDelta: '{"path":"/tmp/a.txt' }
    yield { type: 'tool-call-delta', index: 1, id: CallId('c1'), name: 'read_file', argumentsDelta: '"}' }
    yield {
      type: 'block-end', index: 1,
      block: { type: 'tool-call', id: CallId('c1'), name: 'read_file', arguments: '{"path":"/tmp/a.txt"}' },
    }
    // 3. 用量与结束原因：usage 必须在 finish 之前
    yield { type: 'usage', usage: { inputTokens: 12, outputTokens: 30 } }
    yield { type: 'finish', reason: { kind: 'tool-calls' } }
  }
}

async function main() {
  const app = new Context()
  await app.plugin(LlmRuntime)

  // 注册两个 provider，各自挂一个适配器 —— 换 provider 就是换适配器
  app.llm.registerAdapter(['mock-alpha'], new MockAdapter('Alpha'))
  app.llm.registerAdapter(['mock-beta'], new MockAdapter('Beta'))

  const signal = new AbortController().signal
  const options = {
    provider: 'mock-alpha',
    model: 'mock-model',
    messages: [],
    signal,
  } as unknown as GenerateOptions

  for (const provider of ['mock-alpha', 'mock-beta']) {
    console.log(`\n========== provider: ${provider} ==========`)

    // 1. 直接消费适配器的 stream：逐 chunk 打印
    const adapter = (app.llm as any).adapters.get(provider)?.adapter as MockAdapter
    const assembler = new BlockAssembler()
    console.log('--- 原始 chunk 流 ---')
    for await (const chunk of adapter.stream({ ...options, provider })) {
      const brief = summarizeChunk(chunk)
      console.log(`  ${brief}`)
      assembler.push(chunk)
    }

    // 2. 组装结果：chunk 流 -> 完整消息
    console.log('--- BlockAssembler 组装结果 ---')
    for (const block of assembler.blocks()) {
      if (block.type === 'text') console.log(`  text: ${block.text}`)
      if (block.type === 'tool-call') console.log(`  tool-call: ${block.name}(${block.arguments})`)
    }
    console.log(`  usage: input=${assembler.usage?.inputTokens} output=${assembler.usage?.outputTokens}`)
    console.log(`  finish: ${JSON.stringify(assembler.finish)}`)
  }

  await app.fiber.dispose()
}

function summarizeChunk(chunk: StreamChunk): string {
  switch (chunk.type) {
    case 'block-start': return `block-start  #${chunk.index} ${chunk.blockType}`
    case 'text-delta': return `text-delta   #${chunk.index} +"${chunk.text}"`
    case 'tool-call-delta': return `tool-call-delta #${chunk.index} +"${chunk.argumentsDelta}"`
    case 'block-end': return `block-end    #${chunk.index} ${chunk.block.type}`
    case 'usage': return `usage        in=${chunk.usage.inputTokens} out=${chunk.usage.outputTokens}`
    case 'finish': return `finish       ${JSON.stringify(chunk.reason)}`
    default: return chunk.type
  }
}

main().catch((err) => { console.error(err); process.exit(1) })

========== provider: mock-alpha ==========
--- 原始 chunk 流 ---
  block-start  #0 text
  text-delta   #0 +"你好，我是 Alpha 模型。"
  text-delta   #0 +" 我会流式输出。"
  block-end    #0 text
  block-start  #1 tool-call
  tool-call-delta #1 +"{"path":"/tmp/a.txt"
  tool-call-delta #1 +""}"
  block-end    #1 tool-call
  usage        in=12 out=30
  finish       {"kind":"tool-calls"}
--- BlockAssembler 组装结果 ---
  text: 你好，我是 Alpha 模型。 我会流式输出。
  tool-call: read_file({"path":"/tmp/a.txt"})
  usage: input=12 output=30
  finish: {"kind":"tool-calls"}

========== provider: mock-beta ==========
--- 原始 chunk 流 ---
  block-start  #0 text
  text-delta   #0 +"你好，我是 Beta 模型。"
  text-delta   #0 +" 我会流式输出。"
  block-end    #0 text
  block-start  #1 tool-call
  tool-call-delta #1 +"{"path":"/tmp/a.txt"
  tool-call-delta #1 +""}"
  block-end    #1 tool-call
  usage        in=12 out=30
  finish       {"kind":"tool-calls"}
--- BlockAssembler 组装结果 ---
  text: 你好，我是 Beta 模型。 我会流式输出。
  tool-call: read_file({"path":"/tmp/a.txt"})
  usage: input=12 output=30
  finish: {"kind":"tool-calls"}

/**
 * Raw streaming protocol emitted by adapters.
 * Block indexes correlate interleaved deltas, and `block-end` carries the
 * assembled block. Adapters emit usage before the terminal finish and nothing
 * afterward; tool arguments remain raw JSON strings. An adapter implementation
 * may throw, but `LlmRuntime.stream()` normalizes that failure to a terminal
 * `error` or `aborted` finish before exposing it to consumers.
 */
export type StreamChunk =
  | { type: 'block-start'; index: number; blockType: ContentBlockType }
  | { type: 'text-delta'; index: number; text: string }
  | { type: 'reasoning-delta'; index: number; text: string }
  | { type: 'tool-call-delta'; index: number; id: CallId; name?: string; argumentsDelta: string }
  | { type: 'block-end'; index: number; block: ContentBlock }
  | { type: 'usage'; usage: TokenUsage }
  | {
    type: 'finish'
    reason: FinishReason
    /** Adapter-private lossless-JSON state for replaying a successful response. */
    replayState?: unknown
  }

      const assembler = new BlockAssembler()
      const chunkSeqs: number[] = []
      const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)
      signal.throwIfAborted()
      for await (const chunk of stream) {
        signal.throwIfAborted()
        chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq)
        assembler.push(chunk)
      }
      signal.throwIfAborted()
      const finish = assembler.finish
      if (finish.kind === 'error' || finish.kind === 'aborted') {
        const action = await this.dispatch.waterfall(
          'agent/request-error', {
            turn,
            step,
            provider: request.provider,
            failure: finish.failure,
            retryPolicy: preparedCall?.retryPolicy,
            signal,
          },
          () => Promise.resolve<RequestErrorAction>(undefined),
        )
        signal.throwIfAborted()
        if (action?.kind !== 'retry') {
          throw new LlmError(finish.failure.message, finish.failure.code, finish.failure)
        }
        continue
      }

      const message = createAssistantMessage({
        content: assembler.blocks(),
        source: {
          provider: request.provider,
          model: request.model,
          ...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {},
        },
      })
      this.session.append(
        'assistant/message',
        {
          turn,
          step,
          message,
          ...assembler.usage === undefined ? {} : { usage: assembler.usage },
        },
        { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
      )

function localDelay(config: ResolvedRetryPolicy, retry: number, random: () => number): number {
  const exponent = Math.min(retry - 1, 1024)
  const exponential = Math.min(config.initialDelayMs * 2 ** exponent, config.maxDelayMs)
  const jitter = 1 - config.jitterRatio + 2 * config.jitterRatio * random()
  return Math.min(exponential * jitter, config.maxDelayMs)
}

  async function backoff(
    agent: Agent,
    turn: number,
    step: number,
    failure: LlmFailure,
    provider: string,
    policy: ResolvedRetryPolicy,
    policyKey: string,
    retry: number,
    retryId: RetryId,
    delayMs: number,
    signal: AbortSignal,
  ): Promise<RequestErrorAction> {
    const fusedSignal = AbortSignal.any([signal, lifetime.signal])
    if (fusedSignal.aborted) return
    const eventData: LlmRetryEventData = policy.mode === 'normal'
      ? {
        retryId,
        turn,
        step,
        provider,
        mode: policy.mode,
        policyKey,
        retry,
        maxRetries: policy.maxRetries,
        delayMs,
        failure,
      }
      : {
        retryId,
        turn,
        step,
        provider,
        mode: policy.mode,
        policyKey,
        retry,
        delayMs,
        failure,
      }
    agent.session.append('llm/retry', eventData)
    if (!await cancellableDelay(delayMs, fusedSignal)) return
    agent.session.append('llm/retry-started', { retryId, turn, step, retry })
    return { kind: 'retry' }
  }

    } else if (!policy.retryableCodes.includes(failure.code)) {
      return next()
    }

    const policyKey = retryPolicyKey(policy)
    const priorPolicyRetry = agent.session.events.findLast((event): event is SessionEvent<'llm/retry'> =>
      event.type === 'llm/retry'
      && event.data.turn === turn
      && event.data.step === step
      && event.data.provider === provider
      && event.data.policyKey === policyKey,
    )
    const previousRetry = priorPolicyRetry?.data.retry ?? 0
    if (policy.mode === 'normal' && previousRetry >= policy.maxRetries) return next()
    const retry = previousRetry + 1
    const retryId = priorPolicyRetry?.data.retryId ?? RetryId(randomUUID())
    let delayMs: number
    if (failure.providerRetryAfterMs !== undefined
      && Number.isFinite(failure.providerRetryAfterMs)
      && failure.providerRetryAfterMs > 0) {
      if (failure.providerRetryAfterMs > policy.maxDelayMs) {
        if (policy.mode === 'normal') return next()
        delayMs = localDelay(policy, retry, random)
      } else {
        delayMs = failure.providerRetryAfterMs
      }
    } else {
      delayMs = localDelay(policy, retry, random)
    }

    return backoff(agent, turn, step, failure, provider, policy, policyKey, retry, retryId, delayMs, signal)

  const disposeListener = ctx.on('agent/request-error', (
    payload,
    next: () => Promise<RequestErrorAction>,
  ) => {
    // A waterfall may have captured this callback before its registration was
    // removed. Lifetime cancellation must prevent that stale callback from
    // entering a downstream policy after disposal.
    if (lifetime.signal.aborted) return Promise.resolve<RequestErrorAction>(undefined)
    return track(recover(payload, next))
  })

  ctx.effect(() => async () => {
    disposeListener()
    lifetime.abort(new Error('llm-retry plugin disposed'))
    await Promise.allSettled([...active])
  }, 'llm-retry: abort and drain active recovery')

/** The usage a chunk or finalized message reports for its step, if any. */
const usageOf = (event: SessionEvent): TokenUsage | undefined =>
  event.type === 'assistant/chunk' && event.data.chunk.type === 'usage'
    ? event.data.chunk.usage
    : event.type === 'assistant/message'
      ? event.data.usage
      : undefined

/** Fixed text-density estimate used until exact tokenization is needed. */
const CHARS_PER_TOKEN = 4

/** Per-block structural overhead for JSON framing and type tags. */
const BLOCK_OVERHEAD = 4

/** Role-field framing overhead added to every priced message. */
export const ROLE_OVERHEAD = 4

/**
 * Price content blocks recursively under the fixed density heuristic.
 * @param blocks - content blocks to price without mutation.
 * @returns heuristic tokens including per-block structural overhead.
 */
export function estimateContent(blocks: readonly ContentBlock[]): number {
  let tokens = 0
  for (const block of blocks) {
    switch (block.type) {
      case 'text':
      case 'reasoning':
        tokens += Math.ceil(block.text.length / CHARS_PER_TOKEN) + BLOCK_OVERHEAD
        break
      case 'tool-call':
        tokens += Math.ceil(block.name.length / CHARS_PER_TOKEN)
          + Math.ceil(block.arguments.length / CHARS_PER_TOKEN)
          + BLOCK_OVERHEAD
        break
      case 'tool-result':
        tokens += estimateContent(block.content) + BLOCK_OVERHEAD
        break
      default:
        // ContentBlockMap is merge-extensible; unknown blocks retain a
        // conservative structural JSON price under the fixed heuristic.
        tokens += BLOCK_OVERHEAD + Math.ceil(JSON.stringify(block).length / CHARS_PER_TOKEN)
    }
  }
  return tokens
}

/**
 * Heuristically price one model-visible message.
 * @param message - message to price without mutation.
 * @returns content and role-framing tokens under the fixed heuristic.
 */
export function estimateMessage(message: Message): number {
  return estimateContent(message.content) + ROLE_OVERHEAD
}