fix(llm): scope adapter failures to model calls

Replace the process-wide adapter-failure WeakSet with a per-call scope bound to the exact AsyncIterable returned by LlmService.stream(). Give every call a unique wrapper so waterfall middleware can reuse an iterable without sharing provenance.

Move agent-loop recovery classification to the model-stream boundary. Only the final adapter behind that exact call can become an agent/request-error; nested llm/stream calls remain ordinary outer middleware failures while preserving the original Error.

Cover nested calls, reused middleware iterables, and end-to-end agent-loop recovery. Update the package and RFC contracts, bilingual pairing record, and generated API and catalog references.
This commit is contained in:
Tianyi Cui
2026-07-19 16:14:58 +08:00
parent bf93605f8c
commit ee6b9d081a
14 changed files with 233 additions and 56 deletions

View File

@@ -471,7 +471,7 @@ Waterfall around every streaming model call (retry, replay, routing). Bound to t
Types: [GenerateOptions](../core-data-structures/core.md) · [StreamChunk](../core-data-structures/llm-streaming.md) Types: [GenerateOptions](../core-data-structures/core.md) · [StreamChunk](../core-data-structures/llm-streaming.md)
Source: [`packages/llm/llm/src/index.ts:42`](../../packages/llm/llm/src/index.ts) Source: [`packages/llm/llm/src/index.ts:43`](../../packages/llm/llm/src/index.ts)
## `session/*` ## `session/*`

View File

@@ -461,8 +461,9 @@ async listModels(provider: string): Promise<LlmModelInfo[]>
* `options.provider`. Replay state is retained only when the same adapter * `options.provider`. Replay state is retained only when the same adapter
* instance owns its historical provider and the target provider. Final * instance owns its historical provider and the target provider. Final
* adapter selection, dispatch, and iteration failures retain their original * adapter selection, dispatch, and iteration failures retain their original
* Error identity and are tagged for narrow agent-loop request recovery; * Error identity and are tagged in a call-local scope for narrow agent-loop
* middleware failures remain untagged. * request recovery; middleware and nested-call failures remain untagged for
* the outer call.
* @param options - the full request; `options.provider` selects the adapter. * @param options - the full request; `options.provider` selects the adapter.
* @returns the chunk stream, possibly wrapped by `llm/stream` listeners. * @returns the chunk stream, possibly wrapped by `llm/stream` listeners.
*/ */
@@ -471,7 +472,7 @@ stream(options: GenerateOptions): AsyncIterable<StreamChunk>
Types: [GenerateOptions](../core-data-structures/core.md) · [StreamChunk](../core-data-structures/llm-streaming.md) Types: [GenerateOptions](../core-data-structures/core.md) · [StreamChunk](../core-data-structures/llm-streaming.md)
Source: [`packages/llm/llm/src/index.ts:96`](../../packages/llm/llm/src/index.ts) Source: [`packages/llm/llm/src/index.ts:97`](../../packages/llm/llm/src/index.ts)
## `ctx.permission` — `PermissionService` ## `ctx.permission` — `PermissionService`

View File

@@ -27,7 +27,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac
| `fs/edit-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:61`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/edit-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:61`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) |
| `fs/observed` | `emit` | [`packages/fs/fs/src/index.ts:70`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`emit`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/observed` | `emit` | [`packages/fs/fs/src/index.ts:70`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`emit`) | [`fs-policy`](../packages/fs/fs-policy) |
| `fs/write-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:53`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/write-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:53`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) |
| `llm/stream` | `waterfall` | [`packages/llm/llm/src/index.ts:42`](../packages/llm/llm/src/index.ts) | [`llm`](../packages/llm/llm) (`waterfall`) | [`invariants`](../packages/support/invariants), [`llm-replay`](../packages/support/llm-replay) | | `llm/stream` | `waterfall` | [`packages/llm/llm/src/index.ts:43`](../packages/llm/llm/src/index.ts) | [`llm`](../packages/llm/llm) (`waterfall`) | [`invariants`](../packages/support/invariants), [`llm-replay`](../packages/support/llm-replay) |
| `session/created` | `emit` | [`packages/core/session/src/index.ts:47`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence) | | `session/created` | `emit` | [`packages/core/session/src/index.ts:47`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence) |
| `session/disposed` | `emit` | [`packages/core/session/src/index.ts:57`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`session-persistence`](../packages/session-persistence/session-persistence) | | `session/disposed` | `emit` | [`packages/core/session/src/index.ts:57`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`session-persistence`](../packages/session-persistence/session-persistence) |
| `session/event` | `emit` | [`packages/core/session/src/index.ts:69`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`cli-demo`](../packages/examples/cli-demo), [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio`](../packages/ui/stdio), [`token-meter`](../packages/llm/token-meter), [`tui`](../packages/ui/tui), [`workspace-context`](../packages/context/workspace-context) | | `session/event` | `emit` | [`packages/core/session/src/index.ts:69`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`cli-demo`](../packages/examples/cli-demo), [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio`](../packages/ui/stdio), [`token-meter`](../packages/llm/token-meter), [`tui`](../packages/ui/tui), [`workspace-context`](../packages/context/workspace-context) |

View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority; # side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with: # after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write # pnpm run verify-translation-pairing --write
2026-07-10-after-call-compaction-pressure-and-overflow-recovery.md: abc374d98fed8cac039f10d930fec332515c846e 2026-07-10-after-call-compaction-pressure-and-overflow-recovery.md: d88d7aaea8ccec30b10bfeb17f1312cfe87a0ce7
2026-07-10-after-call-compaction-pressure-and-overflow-recovery.zh.md: ef13c64fc4c728e5b4fe9718ae8ef8b8ef9ecdd9 2026-07-10-after-call-compaction-pressure-and-overflow-recovery.zh.md: 42e6114304de9c8022ef8f1341035858c0c7d9ec

View File

@@ -22,7 +22,7 @@ The loop fires awaited serial `agent/post-step(agent, turn, step, signal)` after
### Request recovery is limited to the final model boundary ### Request recovery is limited to the final model boundary
`RequestError`, `RequestErrorDecision`, and the `agent/request-error` waterfall represent failures after the final adapter has been selected. Private `WeakSet` tagging preserves the original thrown error identity across dispatch, iterator construction, and iteration. Terminal in-band `error` or `aborted` finishes enter the same path. Prompt assembly, request middleware, request logging, result processing, tools, post-step listeners, and cleanup remain ordinary failures. `RequestError`, `RequestErrorDecision`, and the `agent/request-error` waterfall represent failures after the final adapter has been selected. Each returned stream handle owns a private failure set that preserves the original thrown error identity across dispatch, iterator construction, and iteration without leaking nested-call provenance into an outer call. Terminal in-band `error` or `aborted` finishes enter the same path. Prompt assembly, request middleware, request logging, result processing, tools, post-step listeners, and cleanup remain ordinary failures.
The failed step closes before recovery runs. A retry opens the next numbered step and rebuilds the request from the durable log; consecutive recovery attempts reset only after a successful provider request. Both DeepSeek adapters normalize recognized provider context-limit failures to `CONTEXT_WINDOW_EXCEEDED`. The failed step closes before recovery runs. A retry opens the next numbered step and rebuilds the request from the durable log; consecutive recovery attempts reset only after a successful provider request. Both DeepSeek adapters normalize recognized provider context-limit failures to `CONTEXT_WINDOW_EXCEEDED`.

View File

@@ -22,7 +22,7 @@ Status: implemented
### 请求恢复只覆盖最终模型边界 ### 请求恢复只覆盖最终模型边界
`RequestError`、`RequestErrorDecision` 与 `agent/request-error` waterfall 表示最终适配器已经选定之后的失败。私有 `WeakSet` 标记在分发、异步迭代器构造与迭代过程中保留原始抛出错误的身份。终止性的带内 `error` 或 `aborted` finish 进入同一路径。提示词装配、请求中间件、请求日志、结果处理、工具、post-step 监听器与清理仍属于普通失败。 `RequestError`、`RequestErrorDecision` 与 `agent/request-error` waterfall 表示最终适配器已经选定之后的失败。每个返回的流句柄都绑定一个私有失败集合;该集合在分发、异步迭代器构造与迭代过程中保留原始抛出错误的身份,同时防止把嵌套调用的错误来源误归到外层调用。终止性的带内 `error` 或 `aborted` finish 进入同一路径。提示词装配、请求中间件、请求日志、结果处理、工具、post-step 监听器与清理仍属于普通失败。
恢复运行前,失败 step 已经关闭。重试会打开下一个编号 step,并从持久日志重建请求;连续恢复尝试计数只在提供方请求成功后重置。两个 DeepSeek 适配器都把识别出的提供方上下文限制错误规范化为 `CONTEXT_WINDOW_EXCEEDED`。 恢复运行前,失败 step 已经关闭。重试会打开下一个编号 step,并从持久日志重建请求;连续恢复尝试计数只在提供方请求成功后重置。两个 DeepSeek 适配器都把识别出的提供方上下文限制错误规范化为 `CONTEXT_WINDOW_EXCEEDED`。

View File

@@ -27,7 +27,7 @@ function toError(error: unknown): RequestError {
return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error }) return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error })
} }
/** Distinguishes a terminal failure finish from failures in later step processing. */ /** Distinguishes final model-request failures from failures in later step processing. */
class TerminalModelRequestFailure extends Error { class TerminalModelRequestFailure extends Error {
constructor(readonly requestError: RequestError) { constructor(readonly requestError: RequestError) {
super(requestError.message, { cause: requestError }) super(requestError.message, { cause: requestError })
@@ -347,9 +347,7 @@ async function runTurn(
stepOutcome = await runStep( stepOutcome = await runStep(
ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal) ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
} catch (error: unknown) { } catch (error: unknown) {
if (isLlmAdapterFailure(error)) { if (error instanceof TerminalModelRequestFailure) {
stepOutcome = { requestError: error }
} else if (error instanceof TerminalModelRequestFailure) {
stepOutcome = { requestError: error.requestError } stepOutcome = { requestError: error.requestError }
} else { } else {
stepOutcome = { error: toError(error) } stepOutcome = { error: toError(error) }
@@ -624,12 +622,18 @@ async function runStep(
// --- Model call (streaming-first; raw chunks are the replay record) --- // --- Model call (streaming-first; raw chunks are the replay record) ---
const assembler = new BlockAssembler() const assembler = new BlockAssembler()
const chunkSeqs: number[] = [] const chunkSeqs: number[] = []
for await (const chunk of ctx.llm.stream(request)) { const stream = ctx.llm.stream(request)
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */ try {
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted')) for await (const chunk of stream) {
const chunkEvent = session.append('assistant/chunk', { turn, step, chunk }) /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
chunkSeqs.push(chunkEvent.seq) if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
assembler.push(chunk) const chunkEvent = session.append('assistant/chunk', { turn, step, chunk })
chunkSeqs.push(chunkEvent.seq)
assembler.push(chunk)
}
} catch (error: unknown) {
if (isLlmAdapterFailure(stream, error)) throw new TerminalModelRequestFailure(error)
throw error
} }
// Normalize failure finish chunks into the same path as thrown stream errors. // Normalize failure finish chunks into the same path as thrown stream errors.

View File

@@ -307,6 +307,42 @@ describe('agent post-step and request-error lifecycle', () => {
expect(agent.session.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'error' } } }) expect(agent.session.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'error' } } })
}) })
it('does not offer a nested model-call failure as the outer request failure', async () => {
const outer = new FailureScriptAdapter([textResponse('outer adapter must not run')])
const nested = new FailureScriptAdapter([contextError('nested overflow')])
const ctx = await harness(outer)
ctx.llm.registerAdapter(['nested'], nested)
ctx.on('llm/stream', (options, next) => {
if (options.provider !== 'mock') return next()
return (async function* () {
yield* ctx.llm.stream({
provider: 'nested',
model: 'nested',
messages: [],
...options.signal === undefined ? {} : { signal: options.signal },
})
yield* next()
})()
})
const agent = ctx.agentLoop.create(SessionId('nested-stream-not-recoverable'), { provider: 'mock', model: 'mock' })
let recoveries = 0
ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
recoveries += 1
return next()
})
send(agent)
await waitForIdle(ctx, agent)
expect(nested.requests).toHaveLength(1)
expect(outer.requests).toHaveLength(0)
expect(recoveries).toBe(0)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'error', message: 'nested overflow', code: CONTEXT_WINDOW_EXCEEDED_CODE } },
})
})
it.each(['prompt-submit', 'prompt-assembly', 'pre-step', 'request'] as const)( it.each(['prompt-submit', 'prompt-assembly', 'pre-step', 'request'] as const)(
'does not offer %s middleware failures to request recovery', 'does not offer %s middleware failures to request recovery',
async (boundary) => { async (boundary) => {

View File

@@ -13,7 +13,7 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
- `ctx.llm.listModels(provider: string): Promise<LlmModelInfo[]>` Discover the models one registered provider currently advertises. - `ctx.llm.listModels(provider: string): Promise<LlmModelInfo[]>` Discover the models one registered provider currently advertises.
- `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`. - `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 and privately tags errors from final adapter selection, synchronous dispatch, iterator construction, and iteration. `isLlmAdapterFailure(value)` exposes that provenance without classifying `llm/stream` middleware or downstream consumer failures as provider failures, and without replacing the adapter's original coded `Error`. `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; nested model calls, `llm/stream` middleware, and downstream consumer failures remain unclassified for the outer call. Classification does not replace the adapter's original coded `Error`.
Provider and model metadata is a discovery surface, not a routing whitelist. `registerAdapter()` still owns provider exclusivity, while an adapter may accept model ids absent from `listModels()`; consumers must not reject a request because its model is unlisted. Returned metadata is detached and invalid or duplicate adapter entries fail with `INVALID_ADAPTER` or `INVALID_CATALOG`. Provider and model metadata is a discovery surface, not a routing whitelist. `registerAdapter()` still owns provider exclusivity, while an adapter may accept model ids absent from `listModels()`; consumers must not reject a request because its model is unlisted. Returned metadata is detached and invalid or duplicate adapter entries fail with `INVALID_ADAPTER` or `INVALID_CATALOG`.

View File

@@ -5,30 +5,63 @@
*/ */
import { HarnessError } from './error.ts' import { HarnessError } from './error.ts'
import type { StreamChunk } from './types.ts'
/** Errors proven to originate in final adapter dispatch or iteration. */ /** Errors proven to originate in one model call's final adapter boundary. */
const adapterFailures = new WeakSet<Error>() export type AdapterFailureScope = WeakSet<Error>
/** Call-local failure scopes keyed by the exact stream handle returned to a consumer. */
const adapterFailureScopes = new WeakMap<AsyncIterable<StreamChunk>, AdapterFailureScope>()
/**
* Bind one call's adapter-failure scope to a unique returned stream handle.
* @param stream - the waterfall-selected stream for this call.
* @param failures - errors tagged by this call's final adapter boundary.
* @returns a unique stream handle that delegates iteration to `stream`.
* @internal
*/
export function bindAdapterFailureScope(
stream: AsyncIterable<StreamChunk>,
failures: AdapterFailureScope,
): AsyncIterable<StreamChunk> {
const call = {
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
return stream[Symbol.asyncIterator]()
},
}
adapterFailureScopes.set(call, failures)
return call
}
/** /**
* Preserve an adapter's Error identity while tagging its provider origin. * Preserve an adapter's Error identity while tagging its provider origin.
* @param failures - the call-local final-adapter failure scope.
* @param value - arbitrary value thrown by adapter dispatch or iteration. * @param value - arbitrary value thrown by adapter dispatch or iteration.
* @returns the original Error, or a coded Error wrapping a non-Error throw. * @returns the original Error, or a coded Error wrapping a non-Error throw.
* @internal * @internal
*/ */
export function markLlmAdapterFailure(value: unknown): Error & { code?: string } { export function markLlmAdapterFailure(
failures: AdapterFailureScope,
value: unknown,
): Error & { code?: string } {
const error = value instanceof Error const error = value instanceof Error
? value as Error & { code?: string } ? value as Error & { code?: string }
: new HarnessError(String(value), 'UNKNOWN', { cause: value }) : new HarnessError(String(value), 'UNKNOWN', { cause: value })
adapterFailures.add(error) failures.add(error)
return error return error
} }
/** /**
* Whether a failure came from final adapter dispatch, iterator construction, * Whether a failure came from final adapter dispatch, iterator construction,
* or iteration rather than from an `llm/stream` waterfall listener. * or iteration for the call represented by the exact returned stream handle.
* @param stream - the exact stream returned by the model call being classified.
* @param value - arbitrary failure caught by a model-call consumer. * @param value - arbitrary failure caught by a model-call consumer.
* @returns true only for errors tagged at the final adapter boundary. * @returns true only for errors tagged at that call's final adapter boundary.
*/ */
export function isLlmAdapterFailure(value: unknown): value is Error & { code?: string } { export function isLlmAdapterFailure(
return value instanceof Error && adapterFailures.has(value) stream: AsyncIterable<StreamChunk>,
value: unknown,
): value is Error & { code?: string } {
const failures = adapterFailureScopes.get(stream)
return value instanceof Error && failures !== undefined && failures.has(value)
} }

View File

@@ -10,7 +10,8 @@ import { Context, Service } from 'cordis'
import type { GenerateOptions, LlmModelInfo, LlmProviderInfo, Message, StreamChunk } from './types.ts' import type { GenerateOptions, LlmModelInfo, LlmProviderInfo, Message, StreamChunk } from './types.ts'
import { deepFreeze } from './call-config.ts' import { deepFreeze } from './call-config.ts'
import { HarnessError } from './error.ts' import { HarnessError } from './error.ts'
import { markLlmAdapterFailure } from './adapter-failure.ts' import { bindAdapterFailureScope, markLlmAdapterFailure } from './adapter-failure.ts'
import type { AdapterFailureScope } from './adapter-failure.ts'
export * from './attribution.ts' export * from './attribution.ts'
export * from './brand.ts' export * from './brand.ts'
@@ -206,14 +207,17 @@ export class LlmService extends Service {
* so it cannot suppress the primary provider error. A downstream close awaits * so it cannot suppress the primary provider error. A downstream close awaits
* adapter cleanup, whose failures remain ordinary untagged work. * adapter cleanup, whose failures remain ordinary untagged work.
*/ */
private async * adapterStream(options: GenerateOptions): AsyncGenerator<StreamChunk> { private async * adapterStream(
options: GenerateOptions,
failures: AdapterFailureScope,
): AsyncGenerator<StreamChunk> {
let iterator: AsyncIterator<StreamChunk> let iterator: AsyncIterator<StreamChunk>
try { try {
const adapter = this.registration(options.provider).adapter const adapter = this.registration(options.provider).adapter
const stream = adapter.stream(this.forAdapter(options, adapter)) const stream = adapter.stream(this.forAdapter(options, adapter))
iterator = stream[Symbol.asyncIterator]() iterator = stream[Symbol.asyncIterator]()
} catch (error: unknown) { } catch (error: unknown) {
throw markLlmAdapterFailure(error) throw markLlmAdapterFailure(failures, error)
} }
let completed = false let completed = false
@@ -230,7 +234,7 @@ export class LlmService extends Service {
value = item.value value = item.value
} catch (error: unknown) { } catch (error: unknown) {
iterationFailed = true iterationFailed = true
throw markLlmAdapterFailure(error) throw markLlmAdapterFailure(failures, error)
} }
// End the adapter-owned try before yielding: consumer/middleware // End the adapter-owned try before yielding: consumer/middleware
// failures resumed into this generator must remain untagged. // failures resumed into this generator must remain untagged.
@@ -251,13 +255,16 @@ export class LlmService extends Service {
* `options.provider`. Replay state is retained only when the same adapter * `options.provider`. Replay state is retained only when the same adapter
* instance owns its historical provider and the target provider. Final * instance owns its historical provider and the target provider. Final
* adapter selection, dispatch, and iteration failures retain their original * adapter selection, dispatch, and iteration failures retain their original
* Error identity and are tagged for narrow agent-loop request recovery; * Error identity and are tagged in a call-local scope for narrow agent-loop
* middleware failures remain untagged. * request recovery; middleware and nested-call failures remain untagged for
* the outer call.
* @param options - the full request; `options.provider` selects the adapter. * @param options - the full request; `options.provider` selects the adapter.
* @returns the chunk stream, possibly wrapped by `llm/stream` listeners. * @returns the chunk stream, possibly wrapped by `llm/stream` listeners.
*/ */
stream(options: GenerateOptions): AsyncIterable<StreamChunk> { stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
return this.ctx.waterfall(this, 'llm/stream', options, () => this.adapterStream(options)) const failures: AdapterFailureScope = new WeakSet<Error>()
const stream = this.ctx.waterfall(this, 'llm/stream', options, () => this.adapterStream(options, failures))
return bindAdapterFailureScope(stream, failures)
} }
} }

View File

@@ -30,6 +30,16 @@ class RecordingAdapter extends ScriptedAdapter {
} }
} }
class ThrowingAdapter extends LlmAdapter {
constructor(private readonly failure: Error) {
super()
}
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
throw this.failure
}
}
class CatalogAdapter extends ScriptedAdapter { class CatalogAdapter extends ScriptedAdapter {
constructor( constructor(
private readonly provider: LlmProviderInfo, private readonly provider: LlmProviderInfo,
@@ -83,16 +93,17 @@ describe('LlmService', () => {
it('throws NO_ADAPTER for unregistered providers', async () => { it('throws NO_ADAPTER for unregistered providers', async () => {
const ctx = new Context() const ctx = new Context()
await ctx.plugin(LlmService) await ctx.plugin(LlmService)
const stream = ctx.llm.stream({ provider: 'nope', model: 'any-model', messages: [] })
let caught: unknown let caught: unknown
try { try {
for await (const _ of ctx.llm.stream({ provider: 'nope', model: 'any-model', messages: [] })) { /* drain */ } for await (const _ of stream) { /* drain */ }
} catch (error: unknown) { } catch (error: unknown) {
caught = error caught = error
} }
expect(caught).toBeInstanceOf(LlmError) expect(caught).toBeInstanceOf(LlmError)
expect((caught as LlmError).code).toBe('NO_ADAPTER') expect((caught as LlmError).code).toBe('NO_ADAPTER')
expect((caught as LlmError).message).toContain('no adapter registered') expect((caught as LlmError).message).toContain('no adapter registered')
expect(isLlmAdapterFailure(caught)).toBe(true) expect(isLlmAdapterFailure(stream, caught)).toBe(true)
}) })
it.each(['done', 'value'] as const)('tags a throwing IteratorResult.%s getter without replacing its Error', async (field) => { it.each(['done', 'value'] as const)('tags a throwing IteratorResult.%s getter without replacing its Error', async (field) => {
@@ -122,15 +133,16 @@ describe('LlmService', () => {
await ctx.plugin(LlmService) await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter) ctx.llm.registerAdapter(['test-model'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown let caught: unknown
try { try {
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) { /* drain */ } for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) { } catch (error: unknown) {
caught = error caught = error
} }
expect(caught).toBe(original) expect(caught).toBe(original)
expect(isLlmAdapterFailure(caught)).toBe(true) expect(isLlmAdapterFailure(stream, caught)).toBe(true)
expect(cleanupLookups).toBe(0) expect(cleanupLookups).toBe(0)
}) })
@@ -146,15 +158,91 @@ describe('LlmService', () => {
await ctx.plugin(LlmService) await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter) ctx.llm.registerAdapter(['test-model'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown let caught: unknown
try { try {
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) { /* drain */ } for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) { } catch (error: unknown) {
caught = error caught = error
} }
expect(caught).toBe(original) expect(caught).toBe(original)
expect(isLlmAdapterFailure(caught)).toBe(true) expect(isLlmAdapterFailure(stream, caught)).toBe(true)
})
it('keeps a nested adapter failure scoped to the nested model call', async () => {
const original = new LlmError('nested provider failed', 'NESTED_FAILED')
const outer = new RecordingAdapter(SCRIPT)
const nested = new ThrowingAdapter(original)
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['outer'], outer)
ctx.llm.registerAdapter(['nested'], nested)
let nestedStream: AsyncIterable<StreamChunk> | undefined
ctx.on('llm/stream', (options, next) => {
if (options.provider !== 'outer') return next()
return (async function* () {
nestedStream = ctx.llm.stream({ provider: 'nested', model: 'nested', messages: [] })
yield * nestedStream
})()
})
const outerStream = ctx.llm.stream({ provider: 'outer', model: 'outer', messages: [] })
let caught: unknown
try {
for await (const _chunk of outerStream) { /* drain */ }
} catch (error: unknown) {
caught = error
}
expect(caught).toBe(original)
expect(nestedStream).toBeDefined()
expect(isLlmAdapterFailure(nestedStream!, caught)).toBe(true)
expect(isLlmAdapterFailure(outerStream, caught)).toBe(false)
expect(outer.lastOptions).toBeUndefined()
})
it('keeps call scopes distinct when middleware reuses an iterable', async () => {
const firstFailure = new LlmError('first provider failed', 'FIRST_FAILED')
const secondFailure = new LlmError('second provider failed', 'SECOND_FAILED')
const delegates: AsyncIterable<StreamChunk>[] = []
const shared: AsyncIterable<StreamChunk> = {
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
const delegate = delegates.shift()
if (delegate === undefined) throw new Error('shared stream has no call delegate')
return delegate[Symbol.asyncIterator]()
},
}
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['first'], new ThrowingAdapter(firstFailure))
ctx.llm.registerAdapter(['second'], new ThrowingAdapter(secondFailure))
ctx.on('llm/stream', (_options, next) => {
delegates.push(next())
return shared
})
const firstStream = ctx.llm.stream({ provider: 'first', model: 'first', messages: [] })
const secondStream = ctx.llm.stream({ provider: 'second', model: 'second', messages: [] })
const catchFailure = async (stream: AsyncIterable<StreamChunk>): Promise<unknown> => {
try {
for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) {
return error
}
return new Error('expected adapter to fail')
}
expect(firstStream).not.toBe(secondStream)
const firstCaught = await catchFailure(firstStream)
expect(firstCaught).toBe(firstFailure)
expect(isLlmAdapterFailure(firstStream, firstCaught)).toBe(true)
expect(isLlmAdapterFailure(secondStream, firstCaught)).toBe(false)
const secondCaught = await catchFailure(secondStream)
expect(secondCaught).toBe(secondFailure)
expect(isLlmAdapterFailure(secondStream, secondCaught)).toBe(true)
expect(isLlmAdapterFailure(firstStream, secondCaught)).toBe(false)
expect(delegates).toHaveLength(0)
}) })
it('propagates a rejected next promptly without awaiting a non-settling return', async () => { it('propagates a rejected next promptly without awaiting a non-settling return', async () => {
@@ -179,9 +267,10 @@ describe('LlmService', () => {
await ctx.plugin(LlmService) await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter) ctx.llm.registerAdapter(['test-model'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
const failure = (async (): Promise<unknown> => { const failure = (async (): Promise<unknown> => {
try { try {
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) { /* drain */ } for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) { } catch (error: unknown) {
return error return error
} }
@@ -195,7 +284,7 @@ describe('LlmService', () => {
if (timer !== undefined) clearTimeout(timer) if (timer !== undefined) clearTimeout(timer)
expect(caught).toBe(original) expect(caught).toBe(original)
expect(isLlmAdapterFailure(caught)).toBe(true) expect(isLlmAdapterFailure(stream, caught)).toBe(true)
expect(cleanupCalls).toBe(0) expect(cleanupCalls).toBe(0)
}) })
@@ -221,15 +310,16 @@ describe('LlmService', () => {
await ctx.plugin(LlmService) await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter) ctx.llm.registerAdapter(['test-model'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown let caught: unknown
try { try {
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) break for await (const _chunk of stream) break
} catch (error: unknown) { } catch (error: unknown) {
caught = error caught = error
} }
expect(caught).toBe(cleanup) expect(caught).toBe(cleanup)
expect(isLlmAdapterFailure(caught)).toBe(false) expect(isLlmAdapterFailure(stream, caught)).toBe(false)
expect(cleanupCalls).toBe(1) expect(cleanupCalls).toBe(1)
}) })
@@ -272,16 +362,17 @@ describe('LlmService', () => {
await ctx.plugin(LlmService) await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], adapter) ctx.llm.registerAdapter(['test-model'], adapter)
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown let caught: unknown
try { try {
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) { /* drain */ } for await (const _chunk of stream) { /* drain */ }
} catch (error: unknown) { } catch (error: unknown) {
caught = error caught = error
} }
expect(caught).toBeInstanceOf(HarnessError) expect(caught).toBeInstanceOf(HarnessError)
expect(caught).toMatchObject({ code: 'UNKNOWN', cause: 'plain provider failure' }) expect(caught).toMatchObject({ code: 'UNKNOWN', cause: 'plain provider failure' })
expect(isLlmAdapterFailure(caught)).toBe(true) expect(isLlmAdapterFailure(stream, caught)).toBe(true)
}) })
it('does not tag a failure thrown downstream while consuming adapter output', async () => { it('does not tag a failure thrown downstream while consuming adapter output', async () => {
@@ -290,15 +381,20 @@ describe('LlmService', () => {
await ctx.plugin(LlmService) await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], new ScriptedAdapter(SCRIPT)) ctx.llm.registerAdapter(['test-model'], new ScriptedAdapter(SCRIPT))
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
let caught: unknown let caught: unknown
try { try {
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) throw downstream for await (const _chunk of stream) throw downstream
} catch (error: unknown) { } catch (error: unknown) {
caught = error caught = error
} }
expect(caught).toBe(downstream) expect(caught).toBe(downstream)
expect(isLlmAdapterFailure(caught)).toBe(false) expect(isLlmAdapterFailure(stream, caught)).toBe(false)
expect(isLlmAdapterFailure(new ScriptedAdapter(SCRIPT).stream({
provider: 'unbound', model: 'unbound', messages: [],
}), caught)).toBe(false)
expect(isLlmAdapterFailure(stream, 'consumer failed')).toBe(false)
}) })
it('unregisters adapters when the owning fiber is disposed (HMR safety)', async () => { it('unregisters adapters when the owning fiber is disposed (HMR safety)', async () => {

View File

@@ -342,7 +342,7 @@ Waterfall around every streaming model call (retry, replay, routing). Bound to t
- `options` — the full request. A LOOP-built request arrives deep-frozen (mutation throws): its content is a pure function of the session log (the reconstructability RFC), so listeners read it, never rewrite it. A hand-built one-shot (compaction summarize) is the caller's own object and stays mutable here. - `options` — the full request. A LOOP-built request arrives deep-frozen (mutation throws): its content is a pure function of the session log (the reconstructability RFC), so listeners read it, never rewrite it. A hand-built one-shot (compaction summarize) is the caller's own object and stays mutable here.
[Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L42) [Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L43)
## session/* ## session/*

View File

@@ -6,7 +6,7 @@
The abstract `llm` service: an adapter registry plus a streaming model-call surface, interceptable via the `llm/stream` waterfall. The abstract `llm` service: an adapter registry plus a streaming model-call surface, interceptable via the `llm/stream` waterfall.
[Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L96) [Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L97)
### ctx.llm.registerAdapter(providers, adapter) ### ctx.llm.registerAdapter(providers, adapter)
@@ -21,7 +21,7 @@ Register an adapter for the given provider routes. Throws `LlmError` with code `
**Returns** the disposer that unregisters all of them. **Returns** the disposer that unregisters all of them.
[Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L111) [Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L112)
### ctx.llm.listProviders() ### ctx.llm.listProviders()
@@ -33,7 +33,7 @@ Describe provider routes with a registered adapter.
**Returns** detached provider metadata in registration order. **Returns** detached provider metadata in registration order.
[Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L142) [Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L143)
### ctx.llm.listModels(provider) ### ctx.llm.listModels(provider)
@@ -47,7 +47,7 @@ Discover models advertised by one registered provider. Catalog membership is adv
**Returns** detached model metadata in adapter-preferred order. **Returns** detached model metadata in adapter-preferred order.
[Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L152) [Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L153)
### ctx.llm.stream(options) ### ctx.llm.stream(options)
@@ -55,10 +55,10 @@ Discover models advertised by one registered provider. Catalog membership is adv
stream(options: GenerateOptions): AsyncIterable<StreamChunk> stream(options: GenerateOptions): AsyncIterable<StreamChunk>
``` ```
Stream one model call as raw chunks (token-level deltas). Throws `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 for narrow agent-loop request recovery; middleware failures remain untagged. Stream one model call as raw chunks (token-level deltas). Throws `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.
- `options` — the full request; `options.provider` selects the adapter. - `options` — the full request; `options.provider` selects the adapter.
**Returns** the chunk stream, possibly wrapped by `llm/stream` listeners. **Returns** the chunk stream, possibly wrapped by `llm/stream` listeners.
[Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L259) [Source](https://github.com/deepseek-harness/deepseek-harness/blob/master/packages/llm/llm/src/index.ts#L264)