LLM 适配层：twin adapters 与流式词汇

17  llm/         LLM capability: Service Definition/Consumer + DeepSeek providers

73     step/start74     append entered messages as user/message75     derive model history from the log76     agent/request -> llm/stream -> assistant/chunk* -> assistant/message77     tool/call* -> tools/pre-execute -> tools/execute -> tools/post-execute -> tool/result*78     step/end79     tools owe another request, or next-step input arrived -> claim -> next step80  -> agent/turn-stopping81turn/end

238  /**239   * Bind exact model metadata and the eventual request dispatch to one adapter generation.240   * Dynamic adapters override this so settings changes between preparation and241   * dispatch cannot combine one generation's capabilities with another's endpoint.242   * @param provider - registered provider route.243   * @param model - exact model id.244   * @param signal - cancellation for model resolution.245   * @returns model metadata and a one-generation stream entry point.246   */247  async prepareCall(provider: string, model: string, signal?: AbortSignal): Promise<PreparedAdapterCall> {248    return {249      model: await this.resolveModel(provider, model, signal),250      stream: options => this.stream(options),251    }252  }253254  /**255   * Stream one model call as raw chunks. The only required method.256   * @param options - the fully-assembled request; implementations must honor `options.signal`.257   * @returns the chunk stream, obeying the adapter contract documented on `StreamChunk`.258   */259  abstract stream(options: GenerateOptions): AsyncIterable<StreamChunk>

365  registerAdapter(providers: string[], adapter: LlmAdapter): AdapterRegistrationHandle {366    // The routes this registration currently holds; `replace` rewrites it, and367    // the disposer releases whatever it holds at disposal time.368    const owned = new Set<string>()369    // The disposer has run: `owned` being empty cannot say so on its own,370    // because `replace([])` legally leaves a live registration holding none.371    let released = false372    const dispose = this.ctx.effect(function* (this: LlmRuntime) {373      if (providers.length === 0) throw new LlmError('an adapter must register at least one provider', 'INVALID_ADAPTER')374      this.commitRoutes(owned, this.prepareRoutes(providers, adapter, owned))375      yield () => {376        released = true377        for (const provider of owned) this.adapters.delete(provider)378        owned.clear()379        this.emitAdaptersUpdated()380      }381    }.bind(this), 'llm.registerAdapter()')⋯    // ... 中间两行是效果注册的收尾，见正文 ...384    const handle = (() => void dispose()) as AdapterRegistrationHandle385    handle.replace = (next: string[]): void => {386      // Registering here would leak: the effect's disposer already ran, so387      // nothing remains to release whatever this call would put in the map.388      if (released) {389        throw new LlmError('a disposed adapter registration cannot replace its routes', 'REGISTRATION_DISPOSED')390      }391      this.commitRoutes(owned, this.prepareRoutes(next, adapter, owned))392    }393    return handle394  }

985  stream(options: GenerateOptions): AsyncIterable<StreamChunk> {986    return this.streamWithRegistration(options)987  }988989  private streamWithRegistration(990    options: GenerateOptions,991    prepared?: PreparedDispatch,992  ): AsyncIterable<StreamChunk> {993    return this.ctx.waterfall(994      this,995      'llm/stream',996      options,997      () => this.adapterStream(options, prepared),998    )999  }

1002/** Convert one adapter throw into the stream protocol's terminal outcome. */1003function adapterFailureChunk(error: unknown, signal?: AbortSignal): StreamChunk {1004  const failure = normalizeLlmFailure(error)1005  return {1006    type: 'finish',1007    reason: signal?.aborted || failure.code === 'ABORTED'1008      ? { kind: 'aborted', failure }1009      : { kind: 'error', failure },1010  }1011}

304/**305 * Raw streaming protocol emitted by adapters.306 * Block indexes correlate interleaved deltas, and `block-end` carries the307 * assembled block. Adapters emit usage before the terminal finish and nothing308 * afterward; tool arguments remain raw JSON strings. An adapter implementation309 * may throw, but `LlmRuntime.stream()` normalizes that failure to a terminal310 * `error` or `aborted` finish before exposing it to consumers.311 */312export type StreamChunk =313  | { type: 'block-start'; index: number; blockType: ContentBlockType }314  | { type: 'text-delta'; index: number; text: string }315  | { type: 'reasoning-delta'; index: number; text: string }316  | { type: 'tool-call-delta'; index: number; id: CallId; name?: string; argumentsDelta: string }317  | { type: 'block-end'; index: number; block: ContentBlock }318  | { type: 'usage'; usage: TokenUsage }319  | {320    type: 'finish'321    reason: FinishReason322    /** Replay metadata for a successful response; see {@link ReplayEnvelope}. */323    replayState?: ReplayEnvelope324  }

95/**96 * Merge-extensible content blocks keyed by `type`. New core blocks must land97 * with adapter, UI, and compaction support.98 */99export interface ContentBlockMap {100  'text': TextBlock101  'reasoning': ReasoningBlock102  'image': ImageBlock103  'tool-call': ToolCallBlock104  'tool-result': ToolResultBlock105}

31export function mapFinishReason(reason: string): FinishReason {32  switch (reason) {33    case 'stop': return { kind: 'stop' }34    case 'tool_calls': return { kind: 'tool-calls' }35    case 'length': return { kind: 'max-tokens' }36    default:37      // content_filter, insufficient_system_resource, future additions.38      return {39        kind: 'error',40        failure: { message: `model stopped: ${reason}`, code: reason.toUpperCase() },41      }42  }43}

86export async function* translate(payloads: AsyncIterable<string>): AsyncGenerator<StreamChunk> {87  let nextIndex = 088  let textBlock: OpenBlock | undefined89  let reasoningBlock: OpenBlock | undefined90  const toolBlocks = new Map<number, OpenBlock>()91  const order: OpenBlock[] = []92  let pendingFinish: FinishReason | undefined93  let pendingUsage: TokenUsage | undefined9495  function open(kind: OpenBlock['kind']): OpenBlock {96    const block: OpenBlock = { index: nextIndex++, kind, text: '' }97    order.push(block)98    return block99  }100101  for await (const payload of payloads) {102    if (payload === DONE) {103      for (const block of order) {104        yield { type: 'block-end', index: block.index, block: closeBlock(block) }105      }106      if (pendingUsage) yield { type: 'usage', usage: pendingUsage }107      const reason = pendingFinish ?? { kind: 'stop' as const }108      yield {109        type: 'finish',110        reason: reason.kind === 'stop' && order.length === 0111          ? {112            kind: 'error',113            failure: { message: 'model returned a completed response with no content', code: EMPTY_RESPONSE_CODE },114          }115          : reason,116      }117      return118    }

130      // Reasoning first: thinking mode interleaves it before text. The131      // empty-string first chunk must not open a block.132      const reasoning = delta?.reasoning_content133      if (typeof reasoning === 'string' && reasoning.length > 0) {134        if (!reasoningBlock) {135          reasoningBlock = open('reasoning')136          yield { type: 'block-start', index: reasoningBlock.index, blockType: 'reasoning' }137        }138        reasoningBlock.text += reasoning139        yield { type: 'reasoning-delta', index: reasoningBlock.index, text: reasoning }140      }141142      const content = delta?.content143      if (typeof content === 'string' && content.length > 0) {144        if (!textBlock) {145          textBlock = open('text')146          yield { type: 'block-start', index: textBlock.index, blockType: 'text' }147        }148        textBlock.text += content149        yield { type: 'text-delta', index: textBlock.index, text: content }150      }

127export async function* toStreamChunks(128  events: AsyncIterable<AssistantMessageEvent>,129  contextWindow?: number,130): AsyncGenerator<StreamChunk> {131  // pi-ai contentIndex ↔ our block index map 1:1 (both count blocks from 0132  // in stream order), but we track ids per index for tool calls.133  const toolIds = new Map<number, { id: string; name: string }>()134135  for await (const event of events) {136    switch (event.type) {137      case 'start':138        break139      case 'text_start':140        yield { type: 'block-start', index: event.contentIndex, blockType: 'text' }141        break142      case 'text_delta':143        yield { type: 'text-delta', index: event.contentIndex, text: event.delta }144        break145      case 'text_end':146        yield { type: 'block-end', index: event.contentIndex, block: { type: 'text', text: event.content } }147        break148      case 'thinking_start':149        yield { type: 'block-start', index: event.contentIndex, blockType: 'reasoning' }150        break151      case 'thinking_delta':152        yield { type: 'reasoning-delta', index: event.contentIndex, text: event.delta }153        break154      case 'thinking_end':155        yield { type: 'block-end', index: event.contentIndex, block: { type: 'reasoning', text: event.content } }156        break

191      case 'done':192        yield { type: 'usage', usage: mapUsage(event.message.usage) }193        yield {194          type: 'finish',195          reason: mapStopReason(event.message, contextWindow),196          replayState: toPiReplayState(event.message),197        }198        return

339    while (true) {340      const { request, preparedCall } = await this.buildRequest(341        turn, step, assembly.tools, system, this.session.deriveMessages(), signal,342      )343      const assembler = new BlockAssembler()344      const chunkSeqs: number[] = []345      try {346        const stream = preparedCall?.stream(request) ?? this.loopCtx.llm.stream(request)347        signal.throwIfAborted()348        for await (const chunk of stream) {349          signal.throwIfAborted()350          chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq)351          assembler.push(chunk)352        }353        signal.throwIfAborted()

392      const message = createAssistantMessage({393        content: assembler.blocks(),394        source: {395          provider: request.provider,396          model: request.model,⋯          ...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {},·        },·      })·      this.session.append(·        'assistant/message',·        {·          turn,·          step,·          message,⋯          ...assembler.usage === undefined ? {} : { usage: assembler.usage },407        },408        { surfaceOp: 'append', sourceEventSeqs: chunkSeqs },409      )

265  /** Raw stream chunk — token-level replay fidelity. */266  'assistant/chunk': { turn: number; step: number; chunk: StreamChunk }267  /**268   * Assembled assistant message for one step (derived history uses this).269   * Carries the step's `usage` when the adapter reported token accounting, so270   * the model output and its accounting travel together (there is no separate271   * usage record). `usage` is absent when the adapter reported none. A turn272   * cancelled mid-stream finalizes its delivered text/reasoning prefix as this273   * event with `interrupted: true`; undispatched tool calls are absent. The274   * marker distinguishes that prefix without re-deriving interruption from turn275   * boundaries. An aborted turn with no such event streamed no visible content.276   */277  'assistant/message': { turn: number; step: number; message: AssistantMessage; usage?: TokenUsage; interrupted?: true }

289## `BlockAssembler`290291`BlockAssembler` ([`packages/llm/llm/src/assembler.ts`](../../packages/llm/llm/src/assembler.ts)) is the single shared implementation that folds a `StreamChunk` stream back into `ContentBlock`s, usage, finish reason, and replay state. The loop logs the raw chunks while feeding the same chunks through an assembler, then stores the assembled assistant content with the provider and model that produced it. A consumer that needs the assembled result without re-implementing the fold uses this.292293One keep/drop decision covers content and metadata together: a `max-tokens` finish drops every tool call because a truncated call is unsafe to execute, and the same decision prunes the replay envelope's per-block entry at each dropped position. `blocks()` and `replayState` therefore cannot disagree, whatever assembly removes.

48  push(chunk: StreamChunk): void {49    switch (chunk.type) {50      case 'block-start': {51        if (!this.partials.has(chunk.index)) {52          this.order.push(chunk.index)53          this.partials.set(chunk.index, {54            blockType: chunk.blockType,55            text: '',56            toolCallArguments: '',57          })58        }59        return60      }61      case 'text-delta':62      case 'reasoning-delta': {63        const partial = this.ensure(chunk.index, chunk.type === 'text-delta' ? 'text' : 'reasoning')64        if (partial.block) return // closed by block-end; ignore stragglers65        partial.text += chunk.text66        return67      }68      case 'tool-call-delta': {69        const partial = this.ensure(chunk.index, 'tool-call')70        if (partial.block) return // closed by block-end; ignore stragglers71        partial.toolCallId = chunk.id72        if (chunk.name) partial.toolCallName = chunk.name73        partial.toolCallArguments += chunk.argumentsDelta74        return75      }76      case 'block-end': {77        const partial = this.ensure(chunk.index, chunk.block.type)78        // First close wins; ignoring re-close stragglers keeps streamed output79        // and the final assembled block in agreement.80        if (partial.block) return81        partial.block = chunk.block82        return83      }84      case 'usage': {85        this._usage = chunk.usage86        return87      }88      case 'finish': {89        this._finish = chunk.reason90        this._replayState = chunk.replayState91        return92      }93      default: return assertNever(chunk, 'BlockAssembler.push')94    }95  }

134  private assembled(): { blocks: ContentBlock[]; replay: ReplayEnvelope | undefined } {135    const all = this.order.map(index => this.assemble(this.mustGet(index), index))136    const kept = this.finish.kind === 'max-tokens'137      ? all.map(block => block.type !== 'tool-call')138      : undefined139    const blocks = kept === undefined ? all : all.filter((_, position) => kept[position])140    const envelope = this._replayState141    if (envelope?.blocks === undefined) return { blocks, replay: envelope }142    if (envelope.blocks.length !== all.length) return { blocks, replay: undefined }143    return {144      blocks,145      replay: kept === undefined || blocks.length === all.length146        ? envelope147        : { response: envelope.response, blocks: envelope.blocks.filter((_, position) => kept[position]) },148    }149  }

168  interruptedBlocks(): ContentBlock[] {169    return this.order170      .map((index) => {171        const partial = this.mustGet(index)172        const type = partial.block?.type ?? partial.blockType173        if (type !== 'text' && type !== 'reasoning') return undefined174        return this.assemble(partial, index)175      })176      .filter((block): block is ContentBlock =>177        (block?.type === 'text' || block?.type === 'reasoning') && block.text.trim() !== '')178  }

128/** One immutable message representation shared by delivery, durable history, and model requests. */129export interface Message {130  /** Stable identity preserved across every representation boundary. */131  readonly id: MessageId132  /** Provider-neutral conversation role. */133  readonly role: 'system' | 'user' | 'assistant'134  /** Exact model-facing blocks. */135  readonly content: ContentBlock[]136  /** Required source fields supplied by the producer. */137  readonly source: MessageSource138}

169export function freezeMessage<T extends Message>(message: T): T {170  return deepFreeze(structuredClone(message))171}172173/**174 * Create one identified message and freeze it before publication.175 * @param input - complete role, content, and source for a new message.176 * @returns an immutable message with a fresh stable identity.177 */178export function createMessage<T extends NewMessage>(179  input: T & { readonly id?: never },180): T & Pick<Message, 'id'> {181  return freezeMessage({⋯    ...input,183    id: MessageId(crypto.randomUUID()),184  })185}

13Ship **two** adapters against the one contract from the start, deliberately built on different internals:1415- `dsh-llm-deepseek` — direct `fetch` + in-repo translation against the DeepSeek API; SSE framing is delegated to `eventsource-parser` ([the archived SSE-parser swap](../../archived/simplification/2026-07-26-eventsource-parser-for-deepseek-sse.md)). The twin identity is owning the fetch/translate internals rather than delegating to a full provider SDK, not hand-rolling transport plumbing.16- `dsh-llm-pi-ai` — the same endpoint through the `@earendil-works/pi-ai` library (its own event vocabulary).1718The rule they enforce: **anything the StreamChunk vocabulary cannot express for BOTH implementations is a core-vocabulary bug**, caught immediately rather than at the next provider. The pair pinned down conventions now documented on `StreamChunk` in `dsh-llm/src/types.ts`: usage emitted before finish, nothing after finish, tool-call `arguments` as raw JSON strings end-to-end, and the two sanctioned error paths (throw from `stream()` *or* end with `finish {kind:'error'|'aborted'}`) that a consumer must handle on both sides — a divergence the library-backed adapter surfaced that a single direct-fetch adapter would have hidden.

231- **`usage` before `finish`, nothing after `finish`.** Defer both to the provider's end-of-stream marker so a trailing usage-only chunk can't violate the ordering.232- **Tool-call `arguments` stay raw JSON strings end-to-end.** Partial fragments stream via `argumentsDelta`; a provider that hands back parsed objects re-stringifies at `block-end`.233- **Two sanctioned error paths, one `LlmFailure` type.** A failure may either THROW from `stream()` (transport/protocol errors) **or** end the stream with `finish {kind:'error'|'aborted', failure}` (provider in-band errors, for adapters that can't throw mid-stream). `LlmError.failure` carries the same `LlmFailure`. After the call selects its adapter, the stream preserves the exact thrown `Error` object and associates immutable facts plus the serving registration's immutable retry policy with that call; the agent loop closes the failed step and offers the error, facts, immutable prior-retried facts, serving policy, and turn signal to `agent/request-error`. A handling listener returns `{ kind: 'retry' }` after its awaited repair; absent recovery the structured failure becomes the turn error, and no normal assistant message or tool side effect is committed for that attempt.234- **One adapter call is one provider attempt.** Adapters disable library retries. Agent-level recovery opens another durable numbered turn; direct `ctx.llm.stream()` callers remain single-attempt.235- **Provider stalls are bounded at the transport.** Both shipping remote adapters expose positive finite `streamIdleTimeoutMs` with a five-minute default. The watchdog arms only while iterator `next()` is outstanding, uses one stable signal for the whole request, maps its own expiry to `TIMEOUT`, and keeps an earlier caller abort as `ABORTED`.236- **Context overflow has one canonical code.** Both DeepSeek adapters classify explicit provider detail through `isContextWindowExceededError()` and surface `CONTEXT_WINDOW_EXCEEDED`, whether the failure arrives as a thrown HTTP `LlmError` or an in-band finish error. Consumers route on the code, never provider text.237- **An empty completion is a retryable error, not a silent success.** Both adapters map a terminal `stop` finish that carried no content blocks to `finish {kind:'error'}` with the canonical `EMPTY_RESPONSE` code, and `dsh-llm-retry` retries it by default; see [empty model responses are retryable](../../.agents/notes/implemented/bug-fix/2026-07-24-empty-model-response-is-retryable.md).238- **Every provider HTTP request carries the app-attribution header.** Adapters send `attributionHeaders()` (below) - the `User-Agent` baseline - and prove it with a wire-level test.239- **Replay state is adapter-owned; its split is shared.** A successful `finish` may carry a `ReplayEnvelope`: opaque response-level metadata plus optional per-block entries aligned with the emitted block sequence. The alignment is the harness's vocabulary — when assembly drops a block it drops the entry at the same position, so stored metadata always describes stored content. The loop stores the pruned envelope with the assembled assistant message. On a later request, `LlmRuntime` passes the state only when the historical provider and target provider are currently registered to the exact same adapter instance. That adapter validates the state and owns any cross-model or cross-provider conversion; other adapters receive the provider-neutral content plus provider/model fields without the private state. Durable content stays authoritative: a stored state the reading adapter cannot use degrades that one message to provider-neutral conversion with a diagnostic instead of failing the request.

19- Fixed Bedrock streaming behind proxies that strip the response Content-Type header, which silently doubled billed API calls by re-running every turn non-streaming