fix(llm): bind reasoning resolution to adapter lifecycle

2026-07-25 22:59:45 +08:00
parent 478376acaf
commit baea5018e5
30 changed files with 429 additions and 68 deletions
--- a/packages/llm/llm-deepseek/src/adapter.ts
+++ b/packages/llm/llm-deepseek/src/adapter.ts
@@ -147,6 +147,7 @@ export class DeepSeekAdapter extends LlmAdapter {
  override resolveModelReasoning(
    _provider: string,
    _model: string,
+    _signal?: AbortSignal,
  ): Promise<LlmModelReasoningInfo | undefined> {
    if (this.options.defaults?.thinking === 'disabled') return Promise.resolve(undefined)
    return Promise.resolve({
--- a/packages/llm/llm-pi-ai/src/adapter.ts
+++ b/packages/llm/llm-pi-ai/src/adapter.ts
@@ -142,6 +142,7 @@ export class PiAiAdapter extends LlmAdapter {
  override resolveModelReasoning(
    provider: string,
    model: string,
+    _signal?: AbortSignal,
  ): Promise<LlmModelReasoningInfo | undefined> {
    const profile = this.profiles.get(provider)
    if (profile === undefined) {
--- a/packages/llm/llm/README.md
+++ b/packages/llm/llm/README.md
@@ -12,8 +12,9 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
 - `ctx.llm.listProviders(): LlmProviderInfo[]` Describe registered provider routes in registration order.
 - `ctx.llm.listModels(provider: string): Promise<LlmModelInfo[]>` Discover the models one registered provider currently advertises.
 - `ctx.llm.resolveModelContext(provider: string, model: string): Promise<LlmModelContext | undefined>` Resolve authoritative context capacity for one exact route from its owning adapter.
- `ctx.llm.resolveModelReasoning(provider: string, model: string): Promise<LlmModelReasoningInfo | undefined>` Resolve ordered adapter-owned reasoning efforts and an optional deployment default for one exact route.
- `ctx.llm.resolveCallConfig(config: LlmCallConfig): Promise<LlmCallConfig>` Validate an explicit effort and materialize an adapter-configured default without clamping.
+- `ctx.llm.resolveModelReasoning(provider: string, model: string, signal?: AbortSignal): Promise<LlmModelReasoningInfo | undefined>` Resolve ordered adapter-owned reasoning efforts and an optional deployment default for one exact route, with optional cancellation for asynchronous adapters.
+- `ctx.llm.resolveCallConfig(config: LlmCallConfig, signal?: AbortSignal): Promise<LlmCallConfig>` Validate an explicit effort and materialize an adapter-configured default without clamping.
+- `ctx.llm.prepareCall(config: LlmCallConfig, signal?: AbortSignal): Promise<PreparedLlmCall>` Resolve a config and capture its current adapter registration as one cancellable, one-shot call.
 - `ctx.llm.stream(options: GenerateOptions): AsyncIterable<StreamChunk>` Stream one model call as raw chunks (token-level deltas). Consumers assemble the chunks into blocks/messages with `BlockAssembler`.

 `LlmService` preserves errors from final adapter selection, synchronous dispatch, iterator construction, and iteration, and binds their provenance to the exact stream handle returned for that model call. `isLlmAdapterFailure(stream, value)` reports only errors from that call's final adapter boundary; `llmFailureOf(stream, value)` returns the adjacent immutable `LlmFailure`. Nested model calls, `llm/stream` middleware, and downstream consumer failures remain unclassified for the outer call. Classification never replaces or mutates the adapter's original coded `Error`.
@@ -22,7 +23,7 @@ Provider and model metadata is a discovery surface, not a routing whitelist. `re

 Context capacity is a separate correctness query, not a catalog decoration or global LLM setting. `resolveModelContext()` asks the adapter that owns the exact provider/model route; an adapter can describe an unlisted dynamic model, and `undefined` means only that capacity is unavailable. Invalid returned capacity fails with `INVALID_MODEL_CONTEXT`.

-Reasoning effort is also an exact-route capability, but its identifiers are opaque adapter-owned strings rather than a core enum. `resolveModelReasoning()` validates and detaches the ordered display metadata; `undefined` means the model has no selectable effort. `resolveCallConfig()` accepts only an exact advertised identifier, materializes `defaultEffort` when present, and otherwise preserves the provider default. Invalid capability metadata fails with `INVALID_MODEL_REASONING`; an unsupported explicit or configured effort fails with `UNSUPPORTED_REASONING_EFFORT` before provider I/O.
+Reasoning effort is also an exact-route capability, but its identifiers are opaque adapter-owned strings rather than a core enum. `resolveModelReasoning()` validates and detaches the ordered display metadata; `undefined` means the model has no selectable effort. `resolveCallConfig()` accepts only an exact advertised identifier, materializes `defaultEffort` when present, and otherwise preserves the provider default. Asynchronous resolvers receive the caller's signal and must settle promptly after cancellation. `prepareCall()` additionally retains the exact adapter registration through header logging and terminal dispatch, so HMR cannot combine one adapter's capability result with another adapter's request; reusing its one-shot handle or changing its call-config fields fails with `INVALID_PREPARED_CALL`. Invalid capability metadata fails with `INVALID_MODEL_REASONING`; an unsupported explicit or configured effort fails with `UNSUPPORTED_REASONING_EFFORT` before provider I/O.

 ### Events

@@ -32,7 +33,7 @@ Reasoning effort is also an exact-route capability, but its identifiers are opaq

 ### Extension points

- Subclass `LlmAdapter` and call `ctx.llm.registerAdapter(providers, adapter)` to add one or more provider routes. `GenerateOptions.provider` selects the adapter; `GenerateOptions.model` is adapter-owned and may be resolved dynamically. Override `providerInfo()` and asynchronous `listModels()` to expose selector metadata, `resolveModelContext()` when exact capacity is known, and `resolveModelReasoning()` when a model exposes selectable efforts; the defaults use the route id as its name, advertise no models, and return neither capacity nor reasoning metadata.
+- Subclass `LlmAdapter` and call `ctx.llm.registerAdapter(providers, adapter)` to add one or more provider routes. `GenerateOptions.provider` selects the adapter; `GenerateOptions.model` is adapter-owned and may be resolved dynamically. Override `providerInfo()` and asynchronous `listModels()` to expose selector metadata, `resolveModelContext()` when exact capacity is known, and `resolveModelReasoning()` when a model exposes selectable efforts; an asynchronous reasoning resolver must honor its optional cancellation signal. The defaults use the route id as its name, advertise no models, and return neither capacity nor reasoning metadata.
 - Wrap `llm/stream` via `ctx.on()` waterfall listeners for caching, logging, or routing. A wrapper that retries after emitting a chunk has no durable attempt boundary; shipped agent retry policy therefore uses `agent/request-error` instead.

 ### Content-block vocabulary (`types.ts`)
@@ -43,7 +44,7 @@ Streaming is a raw chunk protocol (`block-start`, `text-delta`, `reasoning-delta

 ### Call configuration (`call-config.ts`)

-`LlmCallConfig` is the provider, model, optional adapter-owned reasoning effort, and sampling scalars of one conversation's requests (`provider`, `model`, `reasoningEffort`, `temperature`, `maxTokens`, `stop` — each mapping 1:1 onto the same-named `GenerateOptions` field). It is per-conversation state recorded in the session log as part of the request header (see the dsh-session `request/header` events), never a silently-adjustable per-call knob: the `agent/request` waterfall proposes a replacement, `resolveCallConfig()` validates and defaults it, and the loop logs the effective value before dispatch. `callConfigEquals(a, b)` is the field-wise real-change detector; `deepFreeze(value)` is the ownership helper the loop applies to every built request before dispatch (`llm/stream` listeners and adapters read, never rewrite). `markAgentLoopRequest()` gives that exact object process-local loop provenance, and `isAgentLoopRequest()` lets observers distinguish it from independently logged auxiliary calls that may also be frozen and session-associated. `GenerateOptions.purpose` classifies logged auxiliary compaction and session-title calls so adapters can apply purpose-specific transport policy without changing ordinary conversation requests.
+`LlmCallConfig` is the provider, model, optional adapter-owned reasoning effort, and sampling scalars of one conversation's requests (`provider`, `model`, `reasoningEffort`, `temperature`, `maxTokens`, `stop` — each mapping 1:1 onto the same-named `GenerateOptions` field). It is per-conversation state recorded in the session log as part of the request header (see the dsh-session `request/header` events), never a silently-adjustable per-call knob: the `agent/request` waterfall proposes a replacement, `prepareCall()` validates and defaults it under the turn signal, and the loop logs the effective value before using the prepared call's registration-bound stream. `callConfigEquals(a, b)` is the field-wise real-change detector; `deepFreeze(value)` is the ownership helper the loop applies to every built request before dispatch (`llm/stream` listeners and adapters read, never rewrite). `markAgentLoopRequest()` gives that exact object process-local loop provenance, and `isAgentLoopRequest()` lets observers distinguish it from independently logged auxiliary calls that may also be frozen and session-associated. `GenerateOptions.purpose` classifies logged auxiliary compaction and session-title calls so adapters can apply purpose-specific transport policy without changing ordinary conversation requests.

 ### App attribution (`attribution.ts`)

--- a/packages/llm/llm/src/index.ts
+++ b/packages/llm/llm/src/index.ts
@@ -105,6 +105,20 @@ export class LlmError extends HarnessError {
  }
 }

+/** One model call whose config and adapter registration were resolved together. */
+export interface PreparedLlmCall {
+  /** Detached, deep-frozen config with any adapter-owned default materialized. */
+  readonly config: LlmCallConfig
+  /**
+   * Dispatch this call once through the registration captured during
+   * preparation. The request's call-config fields must match {@link config};
+   * reuse or mismatch fails with `INVALID_PREPARED_CALL`.
+   * @param options - fully assembled request carrying the prepared config.
+   * @returns the chunk stream, including the `llm/stream` waterfall.
+   */
+  stream(options: GenerateOptions): AsyncIterable<StreamChunk>
+}
+
 /**
 * Provider-wire adapter for the harness message and stream vocabulary. Register implementations
 * with `ctx.llm.registerAdapter(providers, adapter)`. Every provider HTTP request must include
@@ -151,11 +165,14 @@ export abstract class LlmAdapter {
   * the model has no selectable reasoning-effort capability.
   * @param _provider - one provider route owned by this adapter.
   * @param _model - exact model id passed to {@link GenerateOptions.model}.
+   * @param _signal - cancellation for this exact-model lookup; implementations
+   *   must settle promptly after it aborts.
   * @returns adapter-owned effort metadata, or `undefined` when unsupported.
   */
  resolveModelReasoning(
    _provider: string,
    _model: string,
+    _signal?: AbortSignal,
  ): Promise<LlmModelReasoningInfo | undefined> {
    return Promise.resolve(undefined)
  }
@@ -173,7 +190,7 @@ export abstract class LlmAdapter {
 * surface, interceptable via the `llm/stream` waterfall.
 */
 export class LlmService extends Service {
-  private adapters = new Map<string, { adapter: LlmAdapter; provider: LlmProviderInfo }>()
+  private adapters = new Map<string, AdapterRegistration>()

  constructor(ctx: Context) {
    super(ctx, 'llm')
@@ -191,7 +208,7 @@ export class LlmService extends Service {
    const dispose = this.ctx.effect(function* (this: LlmService) {
      if (providers.length === 0) throw new LlmError('an adapter must register at least one provider', 'INVALID_ADAPTER')
      const unique = new Set<string>()
-      const registrations: { adapter: LlmAdapter; provider: LlmProviderInfo }[] = []
+      const registrations: AdapterRegistration[] = []
      for (const provider of providers) {
        if (provider.length === 0) throw new LlmError('adapter provider names must be non-empty', 'INVALID_ADAPTER')
        if (unique.has(provider) || this.adapters.has(provider)) {
@@ -284,13 +301,24 @@ export class LlmService extends Service {
   * effort selector is unsupported for that model.
   * @param provider - registered provider route to inspect.
   * @param model - exact model id passed to the adapter.
+   * @param signal - optional cancellation for adapter-owned asynchronous lookup.
   * @returns detached reasoning metadata, or `undefined` when unsupported.
   */
  async resolveModelReasoning(
    provider: string,
    model: string,
+    signal?: AbortSignal,
  ): Promise<LlmModelReasoningInfo | undefined> {
-    const reasoning = await this.registration(provider).adapter.resolveModelReasoning(provider, model)
+    return this.resolveModelReasoningFor(this.registration(provider), model, signal)
+  }
+
+  private async resolveModelReasoningFor(
+    registration: AdapterRegistration,
+    model: string,
+    signal?: AbortSignal,
+  ): Promise<LlmModelReasoningInfo | undefined> {
+    const provider = registration.provider.id
+    const reasoning = await registration.adapter.resolveModelReasoning(provider, model, signal)
    if (reasoning === undefined) return undefined
    if (reasoning.efforts.length === 0) {
      throw new LlmError(
@@ -335,12 +363,23 @@ export class LlmService extends Service {
  /**
   * Validate a conversation call config against its exact model capability and
   * materialize an adapter-configured default. Unsupported explicit efforts
-   * reject before provider I/O; no clamping or aliasing is performed.
+   * reject before provider I/O; no clamping or aliasing is performed. This
+   * standalone query does not bind a later dispatch; use {@link prepareCall}
+   * when logging and streaming must share one adapter registration.
   * @param config - provider/model route and optional request controls.
+   * @param signal - optional cancellation for adapter-owned capability lookup.
   * @returns a detached config only when a default must be materialized.
   */
-  async resolveCallConfig(config: LlmCallConfig): Promise<LlmCallConfig> {
-    const reasoning = await this.resolveModelReasoning(config.provider, config.model)
+  async resolveCallConfig(config: LlmCallConfig, signal?: AbortSignal): Promise<LlmCallConfig> {
+    return this.resolveCallConfigFor(this.registration(config.provider), config, signal)
+  }
+
+  private async resolveCallConfigFor(
+    registration: AdapterRegistration,
+    config: LlmCallConfig,
+    signal?: AbortSignal,
+  ): Promise<LlmCallConfig> {
+    const reasoning = await this.resolveModelReasoningFor(registration, config.model, signal)
    const requested = config.reasoningEffort
    if (reasoning === undefined) {
      if (requested !== undefined) {
@@ -362,7 +401,33 @@ export class LlmService extends Service {
    return requested === effective ? config : { ...config, reasoningEffort: effective }
  }

-  private registration(provider: string): { adapter: LlmAdapter; provider: LlmProviderInfo } {
+  /**
+   * Resolve one call under its current adapter registration. The returned
+   * one-shot handle keeps that registration across header logging and dispatch,
+   * so HMR cannot combine one adapter's capability result with another adapter.
+   * @param config - provider/model route and optional request controls.
+   * @param signal - optional cancellation for adapter-owned capability lookup.
+   * @returns a prepared config and its registration-bound stream entry point.
+   */
+  async prepareCall(config: LlmCallConfig, signal?: AbortSignal): Promise<PreparedLlmCall> {
+    const registration = this.registration(config.provider)
+    const resolvedConfig = deepFreeze(structuredClone(
+      await this.resolveCallConfigFor(registration, config, signal),
+    ))
+    let dispatched = false
+    return Object.freeze({
+      config: resolvedConfig,
+      stream: (options: GenerateOptions): AsyncIterable<StreamChunk> => {
+        if (dispatched) {
+          throw new LlmError('a prepared LLM call can only be dispatched once', 'INVALID_PREPARED_CALL')
+        }
+        dispatched = true
+        return this.streamWithRegistration(options, { registration, config: resolvedConfig })
+      },
+    })
+  }
+
+  private registration(provider: string): AdapterRegistration {
    const registration = this.adapters.get(provider)
    if (!registration) throw new LlmError(`no adapter registered for provider "${provider}"`, 'NO_ADAPTER')
    return registration
@@ -395,16 +460,26 @@ export class LlmService extends Service {
  private async * adapterStream(
    options: GenerateOptions,
    failures: AdapterFailureScope,
+    prepared?: { registration: AdapterRegistration; config: LlmCallConfig },
  ): AsyncGenerator<StreamChunk> {
    let iterator: AsyncIterator<StreamChunk>
    try {
-      const resolvedConfig = await this.resolveCallConfig(options)
-      const resolvedOptions = callConfigEquals(options, resolvedConfig)
+      const registration = prepared?.registration ?? this.registration(options.provider)
+      const resolvedConfig = prepared === undefined
+        ? await this.resolveCallConfigFor(registration, options, options.signal)
+        : prepared.config
+      if (prepared !== undefined && !callConfigEquals(options, resolvedConfig)) {
+        throw new LlmError(
+          'prepared LLM call config changed before adapter dispatch',
+          'INVALID_PREPARED_CALL',
+        )
+      }
+      const resolvedOptions = prepared !== undefined || callConfigEquals(options, resolvedConfig)
        ? options
        : Object.isFrozen(options)
          ? deepFreeze({ ...options, ...resolvedConfig })
          : { ...options, ...resolvedConfig }
-      const adapter = this.registration(resolvedOptions.provider).adapter
+      const adapter = registration.adapter
      const stream = adapter.stream(this.forAdapter(resolvedOptions, adapter))
      iterator = stream[Symbol.asyncIterator]()
    } catch (error: unknown) {
@@ -445,18 +520,36 @@ export class LlmService extends Service {
   * `LlmError` with code `NO_ADAPTER` if no adapter is registered for
   * `options.provider`. Replay state is retained only when the same adapter
   * instance owns its historical provider and the target provider. Final
-   * adapter selection, dispatch, and iteration failures retain their original
-   * Error identity and are tagged in a call-local scope for narrow agent-loop
-   * request recovery; middleware and nested-call failures remain untagged for
-   * the outer call.
+   * adapter selection remains fixed through asynchronous reasoning resolution
+   * and dispatch. Selection, dispatch, and iteration failures retain their
+   * original Error identity and are tagged in a call-local scope for narrow
+   * agent-loop request recovery; middleware and nested-call failures remain
+   * untagged for the outer call.
   * @param options - the full request; `options.provider` selects the adapter.
   * @returns the chunk stream, possibly wrapped by `llm/stream` listeners.
   */
  stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
+    return this.streamWithRegistration(options)
+  }
+
+  private streamWithRegistration(
+    options: GenerateOptions,
+    prepared?: { registration: AdapterRegistration; config: LlmCallConfig },
+  ): AsyncIterable<StreamChunk> {
    const failures: AdapterFailureScope = new WeakMap<Error, LlmFailure>()
-    const stream = this.ctx.waterfall(this, 'llm/stream', options, () => this.adapterStream(options, failures))
+    const stream = this.ctx.waterfall(
+      this,
+      'llm/stream',
+      options,
+      () => this.adapterStream(options, failures, prepared),
+    )
    return bindAdapterFailureScope(stream, failures)
  }
 }

+interface AdapterRegistration {
+  readonly adapter: LlmAdapter
+  readonly provider: LlmProviderInfo
+}
+
 export default LlmService
--- a/packages/llm/llm/tests/service.spec.ts
+++ b/packages/llm/llm/tests/service.spec.ts
@@ -799,6 +799,115 @@ describe('LlmService', () => {
    expect(Object.isFrozen(adapter.lastOptions)).toBe(true)
  })

+  it('pins one adapter registration across asynchronous reasoning resolution and dispatch', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    const started = Promise.withResolvers<undefined>()
+    const reasoning = Promise.withResolvers<LlmModelReasoningInfo>()
+    const first = new class extends RecordingAdapter {
+      override resolveModelReasoning(
+        _provider: string,
+        _model: string,
+        _signal?: AbortSignal,
+      ): Promise<LlmModelReasoningInfo> {
+        started.resolve(undefined)
+        return reasoning.promise
+      }
+    }(SCRIPT)
+    const disposeFirst = ctx.llm.registerAdapter(['route'], first)
+    const draining = (async () => {
+      for await (const _chunk of ctx.llm.stream({
+        provider: 'route',
+        model: 'model',
+        messages: [],
+      })) { /* drain */ }
+    })()
+
+    await started.promise
+    disposeFirst()
+    const second = new RecordingAdapter(SCRIPT)
+    ctx.llm.registerAdapter(['route'], second)
+    reasoning.resolve({
+      efforts: [{ id: ReasoningEffortId('high'), name: 'High' }],
+      defaultEffort: ReasoningEffortId('high'),
+    })
+    await draining
+
+    expect(first.lastOptions?.reasoningEffort).toBe(ReasoningEffortId('high'))
+    expect(second.lastOptions).toBeUndefined()
+  })
+
+  it('prepares a one-shot registration-bound call and rejects config drift', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    const adapter = new CatalogAdapter(
+      { id: 'route', name: 'Route' },
+      [],
+      {},
+      {
+        model: {
+          efforts: [{ id: ReasoningEffortId('high'), name: 'High' }],
+          defaultEffort: ReasoningEffortId('high'),
+        },
+      },
+    )
+    ctx.llm.registerAdapter(['route'], adapter)
+    const prepared = await ctx.llm.prepareCall({ provider: 'route', model: 'model' })
+    expect(Object.isFrozen(prepared.config)).toBe(true)
+    const stream = prepared.stream({
+      ...prepared.config,
+      model: 'other',
+      messages: [],
+    })
+
+    await expect((async () => {
+      for await (const _chunk of stream) { /* drain */ }
+    })()).rejects.toMatchObject({ code: 'INVALID_PREPARED_CALL' })
+    expect(() => prepared.stream({
+      ...prepared.config,
+      messages: [],
+    })).toThrow(expect.objectContaining({ code: 'INVALID_PREPARED_CALL' }))
+  })
+
+  it('passes cancellation through reasoning capability resolution', async () => {
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    const started = Promise.withResolvers<undefined>()
+    const adapter = new class extends ScriptedAdapter {
+      override resolveModelReasoning(
+        _provider: string,
+        _model: string,
+        signal?: AbortSignal,
+      ): Promise<LlmModelReasoningInfo | undefined> {
+        started.resolve(undefined)
+        return new Promise((_resolve, reject) => {
+          if (signal === undefined) {
+            reject(new Error('missing reasoning signal'))
+            return
+          }
+          if (signal.aborted) {
+            reject(signal.reason instanceof Error ? signal.reason : new Error('reasoning aborted'))
+            return
+          }
+          signal.addEventListener('abort', () => {
+            reject(signal.reason instanceof Error ? signal.reason : new Error('reasoning aborted'))
+          }, { once: true })
+        })
+      }
+    }(SCRIPT)
+    ctx.llm.registerAdapter(['route'], adapter)
+    const controller = new AbortController()
+    const resolving = ctx.llm.resolveCallConfig(
+      { provider: 'route', model: 'model' },
+      controller.signal,
+    )
+
+    await started.promise
+    const reason = new Error('cancel reasoning')
+    controller.abort(reason)
+    await expect(resolving).rejects.toBe(reason)
+  })
+
  it.each([0, -1, 1.5, Number.NaN])(
    'rejects invalid adapter model context %s',
    async (contextWindow) => {