Merge branch 'master' into worktree/pi-ai-manual-e2e
This commit is contained in:
Yichen Jiang
@@ -42,7 +42,7 @@ Every request carries the shared attribution header from dsh-llm's `attributionH
|
||||
|
||||
## Errors
|
||||
|
||||
Non-2xx responses throw `LlmError` with stable codes: `AUTH` (401/403), `RATE_LIMIT` (429), `INVALID_REQUEST` (400), `SERVER` (5xx), `HTTP_<status>` otherwise. Protocol violations throw `STREAM_CLOSED` (no `[DONE]`) or `MALFORMED_RESPONSE` (bad JSON payload). Unknown wire `finish_reason`s (e.g. `content_filter`, `insufficient_system_resource`) become `finish {kind: 'error', code: <REASON>}` chunks.
|
||||
Non-2xx responses throw `LlmError` with stable codes: `AUTH` (401/403), `RATE_LIMIT` (429), `CONTEXT_WINDOW_EXCEEDED` (a 400 whose provider code, type, or message identifies context overflow), `INVALID_REQUEST` (other 400s), `SERVER` (5xx), `HTTP_<status>` otherwise. Protocol violations throw `STREAM_CLOSED` (no `[DONE]`) or `MALFORMED_RESPONSE` (bad JSON payload). Unknown wire `finish_reason`s (e.g. `content_filter`, `insufficient_system_resource`) become `finish {kind: 'error', code: <REASON>}` chunks.
|
||||
|
||||
## Testing
|
||||
|
||||
@@ -52,15 +52,31 @@ Unit suites run against a local `node:http` mock SSE server (no network). Real-A
|
||||
|
||||
### DeepSeek request
|
||||
|
||||
**What the model sees**: The selected DeepSeek model receives the harness system prompt, message history, tool schemas, stop sequences, and call config without adapter-authored prompt prose. On a prior assistant turn with tool calls, its reasoning content is passed back as required; reasoning from tool-call-free turns is omitted.
|
||||
#### What the model sees
|
||||
|
||||
**Token effect**: Provider tokenization governs exact input. Conditional reasoning passback increases tool-round-trip context, while dropping other reasoning avoids paying those tokens again; cache-read usage is reported when available.
|
||||
The selected DeepSeek model receives the harness system prompt, message history, tool schemas, stop sequences, and call config without adapter-authored prompt prose. On a prior assistant turn with tool calls, its reasoning content is passed back as required; reasoning from tool-call-free turns is omitted.
|
||||
|
||||
#### Token effect
|
||||
|
||||
Provider tokenization governs exact input. Conditional reasoning passback increases tool-round-trip context, while dropping other reasoning avoids paying those tokens again; cache-read usage is reported when available.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
An unchanged assembled prefix is eligible for DeepSeek cache reuse, which this adapter reports in usage. A model-route change or any upstream prompt, schema, prefix, or history change may prevent reuse from the first changed token; reasoning passback appends during tool round trips.
|
||||
|
||||
### DeepSeek response
|
||||
|
||||
**What the model sees**: Reasoning, text, and raw-string tool arguments are translated into harness chunks for the loop to log and assemble.
|
||||
#### What the model sees
|
||||
|
||||
**Token effect**: Generated tokens follow provider thinking and effort settings plus the request's `maxTokens`; only loop-retained blocks affect later input.
|
||||
Reasoning, text, and raw-string tool arguments are translated into harness chunks for the loop to log and assemble.
|
||||
|
||||
#### Token effect
|
||||
|
||||
Generated tokens follow provider thinking and effort settings plus the request's `maxTokens`; only loop-retained blocks affect later input.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
Loop-retained response blocks append to the next request and preserve its earlier reusable prefix; dropped blocks have no later cache effect. Changing the provider or model selects a different cache domain.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
* @module dsh-llm-deepseek/adapter
|
||||
*/
|
||||
|
||||
import { attributionHeaders, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import { attributionHeaders, CONTEXT_WINDOW_EXCEEDED_CODE, isContextWindowExceededError, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, LlmModelInfo, LlmProviderInfo, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { serializeRequest } from './serialize.ts'
|
||||
import type { RequestDefaults } from './serialize.ts'
|
||||
@@ -38,12 +38,17 @@ export interface DeepSeekAdapterOptions {
|
||||
/**
|
||||
* Map an HTTP status to a stable LlmError code.
|
||||
* @param status - status of a non-2xx provider response.
|
||||
* @returns `AUTH` (401/403), `RATE_LIMIT` (429), `INVALID_REQUEST` (400), `SERVER` (5xx), or `HTTP_<status>` for anything else.
|
||||
* @param error - parsed provider error body, when available.
|
||||
* @returns the normalized harness error code.
|
||||
*/
|
||||
export function httpErrorCode(status: number): string {
|
||||
export function httpErrorCode(status: number, error?: WireError['error']): string {
|
||||
if (status === 401 || status === 403) return 'AUTH'
|
||||
if (status === 429) return 'RATE_LIMIT'
|
||||
if (status === 400) return 'INVALID_REQUEST'
|
||||
if (status === 400) {
|
||||
const detail = [error?.code, error?.type, error?.message].filter(Boolean).join(' ')
|
||||
if (isContextWindowExceededError(detail)) return CONTEXT_WINDOW_EXCEEDED_CODE
|
||||
return 'INVALID_REQUEST'
|
||||
}
|
||||
if (status >= 500) return 'SERVER'
|
||||
return `HTTP_${status}`
|
||||
}
|
||||
@@ -92,16 +97,17 @@ export class DeepSeekAdapter extends LlmAdapter {
|
||||
})
|
||||
|
||||
if (!response.ok) {
|
||||
const code = httpErrorCode(response.status)
|
||||
let message = `DeepSeek API error (HTTP ${response.status})`
|
||||
let providerError: WireError['error']
|
||||
try {
|
||||
const parsed = await response.json() as WireError
|
||||
if (parsed.error?.message) message = parsed.error.message
|
||||
providerError = parsed.error
|
||||
if (providerError?.message) message = providerError.message
|
||||
} catch {
|
||||
// Only swallow error-body parsing: the stable code and status-line message
|
||||
// are already captured, so malformed gateway JSON must not mask the failure.
|
||||
// Only swallow error-body parsing: the HTTP status still identifies the
|
||||
// failure, so malformed gateway JSON must not mask it.
|
||||
}
|
||||
throw new LlmError(message, code)
|
||||
throw new LlmError(message, httpErrorCode(response.status, providerError))
|
||||
}
|
||||
if (!response.body) {
|
||||
throw new LlmError('DeepSeek API returned no response body', 'EMPTY_RESPONSE')
|
||||
|
||||
@@ -2,7 +2,7 @@ import { createServer } from 'node:http'
|
||||
import type { IncomingMessage, Server, ServerResponse } from 'node:http'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { LlmError, userAgent } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { CONTEXT_WINDOW_EXCEEDED_CODE, LlmError, userAgent } from '@deepseek-ai/dsh-llm'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import { DeepSeekAdapter } from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import { httpErrorCode } from '../src/adapter.ts'
|
||||
@@ -170,6 +170,32 @@ describe('DeepSeekAdapter against a mock server', () => {
|
||||
).resolves.toBe(code)
|
||||
})
|
||||
|
||||
it('classifies a thrown HTTP context-window rejection with the canonical code', async () => {
|
||||
const server = await mockServer([{
|
||||
kind: 'http-error',
|
||||
status: 400,
|
||||
body: JSON.stringify({
|
||||
error: {
|
||||
message: 'This model maximum context length is 128000 tokens; your input exceeds that limit.',
|
||||
type: 'invalid_request_error',
|
||||
code: 'context_length_exceeded',
|
||||
},
|
||||
}),
|
||||
}])
|
||||
const ctx = await harness(server.url)
|
||||
const code = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
|
||||
.catch((error: unknown) => (error as LlmError).code)
|
||||
expect(code).toBe(CONTEXT_WINDOW_EXCEEDED_CODE)
|
||||
})
|
||||
|
||||
it('classifies only context-capacity HTTP 400 details as context overflow', () => {
|
||||
expect(httpErrorCode(400, { message: 'request too large for model context' }))
|
||||
.toBe(CONTEXT_WINDOW_EXCEEDED_CODE)
|
||||
expect(httpErrorCode(400, { message: 'invalid input: temperature exceeds maximum allowed value' }))
|
||||
.toBe('INVALID_REQUEST')
|
||||
expect(httpErrorCode(413, { code: 'context_length_exceeded' })).toBe('HTTP_413')
|
||||
})
|
||||
|
||||
it('keeps the status-line message for JSON error bodies without a message', async () => {
|
||||
const server = await mockServer([{ kind: 'http-error', status: 500, body: '{"error":{"type":"x"}}' }])
|
||||
const ctx = await harness(server.url)
|
||||
|
||||
@@ -43,7 +43,7 @@ If a listener rewrites assembled assistant content, the loop drops replay state
|
||||
## Vocabulary differences
|
||||
|
||||
- pi-ai tool-call arguments are parsed objects; the harness stores raw JSON strings. The adapter parses input and re-stringifies output.
|
||||
- pi-ai reports failures as in-stream error events; these map to `finish {kind:'error'|'aborted'}` chunks.
|
||||
- pi-ai reports failures as in-stream error events; these map to `finish {kind:'error'|'aborted'}` chunks. Provider-specific error text and usage signals evaluated against the resolved model's context window normalize overflow to `CONTEXT_WINDOW_EXCEEDED`.
|
||||
- pi-ai folds reasoning tokens into output usage; there is no separate reasoning count to map.
|
||||
- `GenerateOptions.stop` is rejected with `UNSUPPORTED_OPTION` because pi-ai's common streaming surface cannot guarantee it across providers.
|
||||
|
||||
@@ -63,15 +63,31 @@ Unit tests use pi-ai catalog models redirected to local mock servers and cover p
|
||||
|
||||
### Provider request through pi-ai
|
||||
|
||||
**What the model sees**: The selected catalog model receives `GenerateOptions.system`, history, tools, and sampling fields supported by pi-ai's common streaming API. This package adds no prompt prose. Provider-native replay metadata is restored only when the adapter validates it for the historical content.
|
||||
#### What the model sees
|
||||
|
||||
**Token effect**: Provider tokenization governs exact input. Conversion adds no model-visible text; replay metadata may let a native API reuse provider-side state.
|
||||
The selected catalog model receives `GenerateOptions.system`, history, tools, and sampling fields supported by pi-ai's common streaming API. This package adds no prompt prose. Provider-native replay metadata is restored only when the adapter validates it for the historical content.
|
||||
|
||||
#### Token effect
|
||||
|
||||
Provider tokenization governs exact input. Conversion adds no model-visible text; replay metadata may let a native API reuse provider-side state.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
Conversion preserves logical request order without adding text, while the selected provider's serialization and replay state determine reuse. Changing adapter instance, provider, model, or any upstream request token may prevent reuse from the first difference.
|
||||
|
||||
### Provider response
|
||||
|
||||
**What the model sees**: pi-ai events become harness reasoning, text, tool-call, usage, and finish chunks. Parsed tool arguments cross the harness boundary as raw JSON strings.
|
||||
#### What the model sees
|
||||
|
||||
**Token effect**: Generated content affects later inputs only after the loop records it. pi-ai folds reasoning tokens into output usage when the provider does not report them separately.
|
||||
pi-ai events become harness reasoning, text, tool-call, usage, and finish chunks. Parsed tool arguments cross the harness boundary as raw JSON strings.
|
||||
|
||||
#### Token effect
|
||||
|
||||
Generated content affects later inputs only after the loop records it. pi-ai folds reasoning tokens into output usage when the provider does not report them separately.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
Recorded response content appends to the next request and does not invalidate its earlier reusable prefix. Unrecorded transport metadata and usage accounting do not affect cache identity.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
|
||||
@@ -115,7 +115,7 @@ export class PiAiAdapter extends LlmAdapter {
|
||||
// Harness-owned and therefore win collisions.
|
||||
headers: requestHeaders(profile.headers),
|
||||
})
|
||||
yield* toStreamChunks(events)
|
||||
yield* toStreamChunks(events, model.contextWindow)
|
||||
} finally {
|
||||
options.signal?.removeEventListener('abort', onCallerAbort)
|
||||
controller.abort('consumer stopped streaming')
|
||||
|
||||
@@ -8,8 +8,9 @@
|
||||
* @module dsh-llm-pi-ai/stream
|
||||
*/
|
||||
|
||||
import { CallId, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, isContextWindowExceededError, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import type { FinishReason, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm'
|
||||
import { isContextOverflow } from '@earendil-works/pi-ai'
|
||||
import type { AssistantMessage, AssistantMessageEvent, Usage as PiUsage } from '@earendil-works/pi-ai'
|
||||
import { toPiReplayState } from './replay.ts'
|
||||
|
||||
@@ -38,9 +39,24 @@ function classifyPiAiError(message: string): string {
|
||||
/**
|
||||
* Map a terminal pi-ai event to the harness finish reason.
|
||||
* @param message - the assistant message carried by the `done` or `error` event.
|
||||
* @returns the harness reason; `error` yields `{kind: 'error'}` with a code classified from the error text.
|
||||
* @param contextWindow - resolved catalog capacity for usage-based overflow detection.
|
||||
* @returns the mapped harness reason. Recognized error text, `stop` usage above
|
||||
* `contextWindow`, and zero-output `length` usage that fills the window map
|
||||
* to `CONTEXT_WINDOW_EXCEEDED`.
|
||||
*/
|
||||
export function mapStopReason(message: AssistantMessage): FinishReason {
|
||||
export function mapStopReason(message: AssistantMessage, contextWindow?: number): FinishReason {
|
||||
const piAiOverflow = isContextOverflow(message, contextWindow)
|
||||
const harnessOverflow = message.stopReason === 'error'
|
||||
&& message.errorMessage !== undefined
|
||||
&& isContextWindowExceededError(message.errorMessage)
|
||||
if (piAiOverflow || harnessOverflow) {
|
||||
return {
|
||||
kind: 'error',
|
||||
message: message.errorMessage ?? `pi-ai detected context overflow for model "${message.model}"`,
|
||||
code: CONTEXT_WINDOW_EXCEEDED_CODE,
|
||||
}
|
||||
}
|
||||
|
||||
switch (message.stopReason) {
|
||||
case 'stop': return { kind: 'stop' }
|
||||
case 'length': return { kind: 'max-tokens' }
|
||||
@@ -58,10 +74,14 @@ export function mapStopReason(message: AssistantMessage): FinishReason {
|
||||
* mid-stream — failures arrive as `error` events, which become error/aborted
|
||||
* `finish` chunks (the harness protocol's other error-delivery style).
|
||||
* @param events - one assistant turn's pi-ai event stream.
|
||||
* @param contextWindow - resolved catalog capacity for usage-based overflow detection.
|
||||
* @returns the harness chunks, ending with `usage` then `finish`; throws
|
||||
* `LlmError` (`STREAM_CLOSED`) if the source ends without a terminal event.
|
||||
*/
|
||||
export async function* toStreamChunks(events: AsyncIterable<AssistantMessageEvent>): AsyncGenerator<StreamChunk> {
|
||||
export async function* toStreamChunks(
|
||||
events: AsyncIterable<AssistantMessageEvent>,
|
||||
contextWindow?: number,
|
||||
): AsyncGenerator<StreamChunk> {
|
||||
// pi-ai contentIndex ↔ our block index map 1:1 (both count blocks from 0
|
||||
// in stream order), but we track ids per index for tool calls.
|
||||
const toolIds = new Map<number, { id: string; name: string }>()
|
||||
@@ -124,13 +144,17 @@ export async function* toStreamChunks(events: AsyncIterable<AssistantMessageEven
|
||||
break
|
||||
case 'done':
|
||||
yield { type: 'usage', usage: mapUsage(event.message.usage) }
|
||||
yield { type: 'finish', reason: mapStopReason(event.message), replayState: toPiReplayState(event.message) }
|
||||
yield {
|
||||
type: 'finish',
|
||||
reason: mapStopReason(event.message, contextWindow),
|
||||
replayState: toPiReplayState(event.message),
|
||||
}
|
||||
return
|
||||
case 'error':
|
||||
// In-stream error delivery (pi-ai's style) → error finish chunk
|
||||
// (the harness's other sanctioned error path besides throwing).
|
||||
yield { type: 'usage', usage: mapUsage(event.error.usage) }
|
||||
yield { type: 'finish', reason: mapStopReason(event.error) }
|
||||
yield { type: 'finish', reason: mapStopReason(event.error, contextWindow) }
|
||||
return
|
||||
// no default: AssistantMessageEvent is pi-ai's closed union; a new
|
||||
// event type should fail compilation here via tsc's exhaustiveness
|
||||
|
||||
@@ -2,9 +2,10 @@ import { createServer } from 'node:http'
|
||||
import type { IncomingMessage, Server, ServerResponse } from 'node:http'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { LlmError, userAgent } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { CONTEXT_WINDOW_EXCEEDED_CODE, LlmError, userAgent } from '@deepseek-ai/dsh-llm'
|
||||
import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai'
|
||||
import { PiAiAdapter } from '@deepseek-ai/dsh-llm-pi-ai'
|
||||
import { getModels } from '@earendil-works/pi-ai'
|
||||
import { resolveProfiles } from '../src/config.ts'
|
||||
import { assemble } from './assemble.ts'
|
||||
|
||||
@@ -198,6 +199,29 @@ describe('PiAiAdapter provider routing', () => {
|
||||
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
|
||||
expect(result.finish).toMatchObject({ kind: 'error', code })
|
||||
})
|
||||
|
||||
it('uses the resolved catalog context window for usage-based overflow detection', async () => {
|
||||
const model = getModels('deepseek').find(candidate => candidate.id === 'deepseek-v4-flash')
|
||||
if (model === undefined) throw new Error('deepseek-v4-flash missing from pi-ai test catalog')
|
||||
const events = [
|
||||
'{"choices":[{"delta":{"role":"assistant","content":""},"index":0,"finish_reason":null}]}',
|
||||
JSON.stringify({
|
||||
choices: [{ delta: {}, index: 0, finish_reason: 'stop' }],
|
||||
usage: { prompt_tokens: model.contextWindow + 1, completion_tokens: 0 },
|
||||
}),
|
||||
'[DONE]',
|
||||
]
|
||||
const server = await mockServer([{ events }])
|
||||
const ctx = await harness(server.url)
|
||||
|
||||
const result = await assemble(ctx, { model: model.id, messages: [] })
|
||||
|
||||
expect(result.finish).toEqual({
|
||||
kind: 'error',
|
||||
message: `pi-ai detected context overflow for model "${model.id}"`,
|
||||
code: CONTEXT_WINDOW_EXCEEDED_CODE,
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('provider profile lifecycle', () => {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { CallId, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId, CONTEXT_WINDOW_EXCEEDED_CODE, LlmError } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import type { AssistantMessage, AssistantMessageEvent, Usage } from '@earendil-works/pi-ai'
|
||||
import { toPiContext } from '../src/context.ts'
|
||||
@@ -523,6 +523,46 @@ describe('mapStopReason / mapUsage', () => {
|
||||
.toMatchObject({ kind: 'error', code: 'RATE_LIMIT' })
|
||||
expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 500: backend down' })))
|
||||
.toMatchObject({ kind: 'error', code: 'SERVER' })
|
||||
expect(mapStopReason(assistant({
|
||||
stopReason: 'error',
|
||||
errorMessage: 'HTTP 400: input exceeds the model context window limit',
|
||||
}))).toMatchObject({ kind: 'error', code: CONTEXT_WINDOW_EXCEEDED_CODE })
|
||||
expect(mapStopReason(assistant({
|
||||
stopReason: 'error',
|
||||
errorMessage: 'HTTP 400: request too large for model context',
|
||||
}))).toMatchObject({ kind: 'error', code: CONTEXT_WINDOW_EXCEEDED_CODE })
|
||||
expect(mapStopReason(assistant({
|
||||
stopReason: 'error',
|
||||
errorMessage: 'HTTP 400: invalid input: temperature exceeds maximum allowed value',
|
||||
}))).toMatchObject({ kind: 'error', code: 'INVALID_REQUEST' })
|
||||
})
|
||||
|
||||
it('uses pi-ai provider-specific overflow classification without losing rate-limit exclusions', () => {
|
||||
expect(mapStopReason(assistant({
|
||||
stopReason: 'error',
|
||||
errorMessage: 'prompt is too long: 213462 tokens > 200000 maximum',
|
||||
}))).toMatchObject({ kind: 'error', code: CONTEXT_WINDOW_EXCEEDED_CODE })
|
||||
expect(mapStopReason(assistant({
|
||||
stopReason: 'error',
|
||||
errorMessage: 'ThrottlingException: Too many tokens, rate limit reached',
|
||||
}))).toMatchObject({ kind: 'error', code: 'RATE_LIMIT' })
|
||||
})
|
||||
|
||||
it('uses the resolved context window for silent and length-stop overflows', () => {
|
||||
const silent = assistant({ stopReason: 'stop', usage: usage(101, 0) })
|
||||
expect(mapStopReason(silent)).toEqual({ kind: 'stop' })
|
||||
expect(mapStopReason(silent, 100)).toEqual({
|
||||
kind: 'error',
|
||||
message: 'pi-ai detected context overflow for model "deepseek-v4-flash"',
|
||||
code: CONTEXT_WINDOW_EXCEEDED_CODE,
|
||||
})
|
||||
|
||||
const truncated = assistant({ stopReason: 'length', usage: usage(80, 0, 19) })
|
||||
expect(mapStopReason(truncated)).toEqual({ kind: 'max-tokens' })
|
||||
expect(mapStopReason(truncated, 100)).toMatchObject({
|
||||
kind: 'error',
|
||||
code: CONTEXT_WINDOW_EXCEEDED_CODE,
|
||||
})
|
||||
})
|
||||
|
||||
it('maps cache fields only when nonzero', () => {
|
||||
|
||||
@@ -13,6 +13,8 @@ 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.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; 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`.
|
||||
|
||||
### Events
|
||||
@@ -46,6 +48,7 @@ Every product adapter sends application identity on provider HTTP requests. `att
|
||||
- `BlockAssembler` — incrementally assembles raw chunks into complete content blocks and an assistant message. The agent loop feeds it raw chunks (logging them for replay) while reading the assembled blocks/message for history.
|
||||
- `HarnessError` — base class for the harness error taxonomy: a stable `code` string (distinct from the human `message`) plus `cause` chaining. Lives here, in the leaf package every other imports, so a single base is shared without a new dependency edge. Per-package errors (`LlmError`, `ToolArgsError`, `InvariantError`, …) extend it. `isHarnessError(value)` narrows at seams.
|
||||
- `LlmError` — extends `HarnessError`; its stable `code` string (`NO_ADAPTER`, `DUPLICATE_ADAPTER`, and adapter codes like `AUTH`/`RATE_LIMIT`) is the programmatic failure contract.
|
||||
- `CONTEXT_WINDOW_EXCEEDED_CODE` — the provider-neutral code both DeepSeek adapters use when a request exceeds the model context window, regardless of thrown-HTTP versus in-band finish delivery. `isContextWindowExceededError(detail)` is their shared conservative classifier for OpenAI-compatible provider detail.
|
||||
|
||||
### Real adapters
|
||||
|
||||
@@ -55,9 +58,13 @@ Two adapters implement `LlmAdapter` on different internals: [`@deepseek-ai/dsh-l
|
||||
|
||||
None, as this adapter registry forwards an already assembled request without adding or changing any model-bound text, schema, or message.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
Pass-through; the registry preserves the assembled request prefix, while the selected adapter and provider own cache reuse and routing boundaries.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **No retry/caching/rate-limit layer ships** — `llm/stream` is the intended wrap seam and has no production listener, so provider 429/5xx failures surface immediately.
|
||||
- **No default retry/caching/rate-limit policy ships in this service** — `llm/stream` remains the call-wrapper seam; the agent loop separately offers proven model-request failures to `agent/request-error`, whose default preserves the original failure.
|
||||
- **`GenerateOptions` sampling is `temperature`/`maxTokens`/`stop` only** — no `tool_choice`, `top_p`, or penalty fields; the vocabulary grows when a producer lands ([dropped inert knobs](../../../docs/rfc/implemented/simplification/2026-07-04-drop-inert-request-knobs.md)).
|
||||
- **Producer-gated variants stay out until produced** — `prefill`, per-tool `strict`, block `cache` hints, and the `agent` message-source variant were pruned as producerless ([RFC](../../../docs/rfc/implemented/simplification/2026-07-04-prune-producerless-vocabulary-variants.md)).
|
||||
- **`BlockAssembler` handles core block kinds only** — a plugin-added block type whose stream is never closed by `block-end` makes `blocks()` throw.
|
||||
|
||||
67
packages/llm/llm/src/adapter-failure.ts
Normal file
67
packages/llm/llm/src/adapter-failure.ts
Normal file
@@ -0,0 +1,67 @@
|
||||
/**
|
||||
* Private provider-failure tagging shared by `LlmService` and its consumers.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-llm/adapter-failure
|
||||
*/
|
||||
|
||||
import { HarnessError } from './error.ts'
|
||||
import type { StreamChunk } from './types.ts'
|
||||
|
||||
/** Errors proven to originate in one model call's final adapter boundary. */
|
||||
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.
|
||||
* @param failures - the call-local final-adapter failure scope.
|
||||
* @param value - arbitrary value thrown by adapter dispatch or iteration.
|
||||
* @returns the original Error, or a coded Error wrapping a non-Error throw.
|
||||
* @internal
|
||||
*/
|
||||
export function markLlmAdapterFailure(
|
||||
failures: AdapterFailureScope,
|
||||
value: unknown,
|
||||
): Error & { code?: string } {
|
||||
const error = value instanceof Error
|
||||
? value as Error & { code?: string }
|
||||
: new HarnessError(String(value), 'UNKNOWN', { cause: value })
|
||||
failures.add(error)
|
||||
return error
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a failure came from final adapter dispatch, iterator construction,
|
||||
* 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.
|
||||
* @returns true only for errors tagged at that call's final adapter boundary.
|
||||
*/
|
||||
export function isLlmAdapterFailure(
|
||||
stream: AsyncIterable<StreamChunk>,
|
||||
value: unknown,
|
||||
): value is Error & { code?: string } {
|
||||
const failures = adapterFailureScopes.get(stream)
|
||||
return value instanceof Error && failures !== undefined && failures.has(value)
|
||||
}
|
||||
@@ -21,6 +21,47 @@ export class HarnessError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/** Canonical provider-neutral code for a model request rejected because its context window was exceeded. */
|
||||
export const CONTEXT_WINDOW_EXCEEDED_CODE = 'CONTEXT_WINDOW_EXCEEDED'
|
||||
|
||||
/** Structured codes and plain phrases that explicitly name a context bound being exceeded. */
|
||||
const STRUCTURED_CONTEXT_OVERFLOW = new RegExp(
|
||||
String.raw`(?:^|[^a-z0-9])context[\s_-](?:length|window)[\s_-]`
|
||||
+ String.raw`(?:exceed(?:ed|s)?|overflow(?:ed)?|limit[\s_-]exceeded)(?:$|[^a-z0-9])`,
|
||||
'i',
|
||||
)
|
||||
|
||||
/** Request-size wording that ties "too large" directly to model context capacity. */
|
||||
const TOO_LARGE_FOR_CONTEXT = new RegExp(
|
||||
String.raw`\b(?:request|prompt|input|messages?)\s+(?:is\s+|are\s+)?`
|
||||
+ String.raw`too\s+(?:large|long)\s+for\s+(?:(?:this|the)\s+)?`
|
||||
+ String.raw`(?:model(?:'s)?\s+)?context(?:\s+window)?\b`,
|
||||
'i',
|
||||
)
|
||||
|
||||
/** "Exceeds" wording is safe only when its object is explicitly the model context. */
|
||||
const EXCEEDS_MODEL_CONTEXT = new RegExp(
|
||||
String.raw`\b(?:input|prompt|request|messages?)\b.{0,40}`
|
||||
+ String.raw`\b(?:exceed(?:s|ed)?|overflows?|is\s+larger\s+than)\b.{0,40}`
|
||||
+ String.raw`\b(?:the\s+)?(?:model(?:'s)?\s+)?context(?:\s+(?:length|window))?\b`,
|
||||
'i',
|
||||
)
|
||||
|
||||
/**
|
||||
* Recognize the context-overflow wording used by OpenAI-compatible providers
|
||||
* and library adapters. Adapters pass all available provider code, type, and
|
||||
* message text so both thrown and in-band delivery styles share one classifier.
|
||||
* @param detail - provider error code/type/message text joined into one string.
|
||||
* @returns true when the detail identifies a request exceeding the model context window.
|
||||
*/
|
||||
export function isContextWindowExceededError(detail: string): boolean {
|
||||
return STRUCTURED_CONTEXT_OVERFLOW.test(detail)
|
||||
|| /\b(?:maximum|max)(?:\s+(?:allowed|supported))?\s+context\s+(?:length|window)\b/i.test(detail)
|
||||
|| TOO_LARGE_FOR_CONTEXT.test(detail)
|
||||
|| /\b(?:input|prompt|request)\s+(?:is\s+)?too\s+(?:long|large)\s+for\s+(?:this|the)\s+model\b/i.test(detail)
|
||||
|| EXCEEDS_MODEL_CONTEXT.test(detail)
|
||||
}
|
||||
|
||||
/**
|
||||
* Narrow an arbitrary thrown value to a HarnessError (for `instanceof` at seams).
|
||||
* @param value - the caught value (`unknown` in catch clauses).
|
||||
|
||||
@@ -8,8 +8,10 @@
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import type { GenerateOptions, LlmModelInfo, LlmProviderInfo, Message, StreamChunk } from './types.ts'
|
||||
import { HarnessError } from './error.ts'
|
||||
import { deepFreeze } from './call-config.ts'
|
||||
import { HarnessError } from './error.ts'
|
||||
import { bindAdapterFailureScope, markLlmAdapterFailure } from './adapter-failure.ts'
|
||||
import type { AdapterFailureScope } from './adapter-failure.ts'
|
||||
|
||||
export * from './attribution.ts'
|
||||
export * from './brand.ts'
|
||||
@@ -19,6 +21,7 @@ export * from './types.ts'
|
||||
export { BlockAssembler } from './assembler.ts'
|
||||
export { callConfigEquals, deepFreeze } from './call-config.ts'
|
||||
export type { LlmCallConfig } from './call-config.ts'
|
||||
export { isLlmAdapterFailure } from './adapter-failure.ts'
|
||||
|
||||
declare module 'cordis' {
|
||||
interface Context {
|
||||
@@ -196,20 +199,72 @@ export class LlmService extends Service {
|
||||
return Object.isFrozen(options) ? deepFreeze(filtered) : filtered
|
||||
}
|
||||
|
||||
/**
|
||||
* Final adapter boundary. It tags only failures from adapter selection,
|
||||
* synchronous dispatch, iterator construction, or iteration while preserving
|
||||
* the original Error object. Middleware outside this generator remains
|
||||
* distinguishable as plugin work. An iteration failure skips adapter cleanup
|
||||
* so it cannot suppress the primary provider error. A downstream close awaits
|
||||
* adapter cleanup, whose failures remain ordinary untagged work.
|
||||
*/
|
||||
private async * adapterStream(
|
||||
options: GenerateOptions,
|
||||
failures: AdapterFailureScope,
|
||||
): AsyncGenerator<StreamChunk> {
|
||||
let iterator: AsyncIterator<StreamChunk>
|
||||
try {
|
||||
const adapter = this.registration(options.provider).adapter
|
||||
const stream = adapter.stream(this.forAdapter(options, adapter))
|
||||
iterator = stream[Symbol.asyncIterator]()
|
||||
} catch (error: unknown) {
|
||||
throw markLlmAdapterFailure(failures, error)
|
||||
}
|
||||
|
||||
let completed = false
|
||||
let iterationFailed = false
|
||||
try {
|
||||
while (true) {
|
||||
let value: StreamChunk
|
||||
try {
|
||||
const item = await iterator.next()
|
||||
if (item.done) {
|
||||
completed = true
|
||||
return
|
||||
}
|
||||
value = item.value
|
||||
} catch (error: unknown) {
|
||||
iterationFailed = true
|
||||
throw markLlmAdapterFailure(failures, error)
|
||||
}
|
||||
// End the adapter-owned try before yielding: consumer/middleware
|
||||
// failures resumed into this generator must remain untagged.
|
||||
yield value
|
||||
}
|
||||
} finally {
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- the iteration catch sets its latch before entering finally.
|
||||
if (!completed && !iterationFailed) {
|
||||
const close = iterator.return?.bind(iterator)
|
||||
if (close) await close()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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. Dispatches
|
||||
* through the `llm/stream` waterfall.
|
||||
* 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.
|
||||
* @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.ctx.waterfall(this, 'llm/stream', options, () => {
|
||||
const adapter = this.registration(options.provider).adapter
|
||||
return adapter.stream(this.forAdapter(options, adapter))
|
||||
})
|
||||
const failures: AdapterFailureScope = new WeakSet<Error>()
|
||||
const stream = this.ctx.waterfall(this, 'llm/stream', options, () => this.adapterStream(options, failures))
|
||||
return bindAdapterFailureScope(stream, failures)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,14 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { GenerateOptions, LlmAdapter, LlmError, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, {
|
||||
GenerateOptions,
|
||||
HarnessError,
|
||||
isContextWindowExceededError,
|
||||
isLlmAdapterFailure,
|
||||
LlmAdapter,
|
||||
LlmError,
|
||||
StreamChunk,
|
||||
} from '@deepseek-ai/dsh-llm'
|
||||
import type { LlmModelInfo, LlmProviderInfo } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
class ScriptedAdapter extends LlmAdapter {
|
||||
@@ -22,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 {
|
||||
constructor(
|
||||
private readonly provider: LlmProviderInfo,
|
||||
@@ -46,6 +64,22 @@ const SCRIPT: StreamChunk[] = [
|
||||
]
|
||||
|
||||
describe('LlmService', () => {
|
||||
it('recognizes structured and model-capacity context-window overflow details', () => {
|
||||
expect(isContextWindowExceededError('context_length_exceeded maximum context length')).toBe(true)
|
||||
expect(isContextWindowExceededError('context-window-overflowed')).toBe(true)
|
||||
expect(isContextWindowExceededError('This model maximum context length is 128000 tokens')).toBe(true)
|
||||
expect(isContextWindowExceededError('input is too long for this model')).toBe(true)
|
||||
expect(isContextWindowExceededError('request too large for model context')).toBe(true)
|
||||
expect(isContextWindowExceededError('input exceeds the model context window limit')).toBe(true)
|
||||
})
|
||||
|
||||
it('does not mistake unrelated input validation for context-window overflow', () => {
|
||||
expect(isContextWindowExceededError('invalid request: malformed tool arguments')).toBe(false)
|
||||
expect(isContextWindowExceededError('invalid input: temperature exceeds maximum allowed value')).toBe(false)
|
||||
expect(isContextWindowExceededError('input exceeds maximum allowed value')).toBe(false)
|
||||
expect(isContextWindowExceededError('context window size must be positive')).toBe(false)
|
||||
})
|
||||
|
||||
it('routes stream() to the registered adapter', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
@@ -59,9 +93,308 @@ describe('LlmService', () => {
|
||||
it('throws NO_ADAPTER for unregistered providers', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await expect((async () => {
|
||||
for await (const _ of ctx.llm.stream({ provider: 'nope', model: 'any-model', messages: [] })) { /* drain */ }
|
||||
})()).rejects.toThrow('no adapter registered')
|
||||
const stream = ctx.llm.stream({ provider: 'nope', model: 'any-model', messages: [] })
|
||||
let caught: unknown
|
||||
try {
|
||||
for await (const _ of stream) { /* drain */ }
|
||||
} catch (error: unknown) {
|
||||
caught = error
|
||||
}
|
||||
expect(caught).toBeInstanceOf(LlmError)
|
||||
expect((caught as LlmError).code).toBe('NO_ADAPTER')
|
||||
expect((caught as LlmError).message).toContain('no adapter registered')
|
||||
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
|
||||
})
|
||||
|
||||
it.each(['done', 'value'] as const)('tags a throwing IteratorResult.%s getter without replacing its Error', async (field) => {
|
||||
const original = new LlmError(`${field} getter failed`, 'RESULT_GETTER_FAILED')
|
||||
const result = field === 'done' ? {} : { done: false }
|
||||
Object.defineProperty(result, field, { get: () => { throw original } })
|
||||
let cleanupLookups = 0
|
||||
const iterator: AsyncIterator<StreamChunk> = {
|
||||
next: () => Promise.resolve(result as unknown as IteratorResult<StreamChunk>),
|
||||
}
|
||||
Object.defineProperty(iterator, 'return', {
|
||||
get: () => {
|
||||
cleanupLookups += 1
|
||||
throw new Error('return getter must not run after iteration fails')
|
||||
},
|
||||
})
|
||||
const adapter = new class extends LlmAdapter {
|
||||
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
return {
|
||||
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
|
||||
return iterator
|
||||
},
|
||||
}
|
||||
}
|
||||
}()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
ctx.llm.registerAdapter(['test-model'], adapter)
|
||||
|
||||
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
|
||||
let caught: unknown
|
||||
try {
|
||||
for await (const _chunk of stream) { /* drain */ }
|
||||
} catch (error: unknown) {
|
||||
caught = error
|
||||
}
|
||||
|
||||
expect(caught).toBe(original)
|
||||
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
|
||||
expect(cleanupLookups).toBe(0)
|
||||
})
|
||||
|
||||
it.each(['dispatch', 'iterator'] as const)('tags synchronous adapter %s failures without replacing their Error', async (boundary) => {
|
||||
const original = new LlmError(`${boundary} failed`, 'BOUNDARY_FAILED')
|
||||
const adapter = new class extends LlmAdapter {
|
||||
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
if (boundary === 'dispatch') throw original
|
||||
return { [Symbol.asyncIterator]: () => { throw original } }
|
||||
}
|
||||
}()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
ctx.llm.registerAdapter(['test-model'], adapter)
|
||||
|
||||
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
|
||||
let caught: unknown
|
||||
try {
|
||||
for await (const _chunk of stream) { /* drain */ }
|
||||
} catch (error: unknown) {
|
||||
caught = error
|
||||
}
|
||||
|
||||
expect(caught).toBe(original)
|
||||
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 () => {
|
||||
const original = new LlmError('provider failed', 'PROVIDER_FAILED')
|
||||
let cleanupCalls = 0
|
||||
const adapter = new class extends LlmAdapter {
|
||||
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
return {
|
||||
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
|
||||
return {
|
||||
next: () => Promise.reject(original),
|
||||
return: () => {
|
||||
cleanupCalls += 1
|
||||
return new Promise<IteratorResult<StreamChunk>>(() => {})
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
ctx.llm.registerAdapter(['test-model'], adapter)
|
||||
|
||||
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
|
||||
const failure = (async (): Promise<unknown> => {
|
||||
try {
|
||||
for await (const _chunk of stream) { /* drain */ }
|
||||
} catch (error: unknown) {
|
||||
return error
|
||||
}
|
||||
return new Error('expected adapter iteration to fail')
|
||||
})()
|
||||
let timer: ReturnType<typeof setTimeout> | undefined
|
||||
const timeout = new Promise<Error>((resolve) => {
|
||||
timer = setTimeout(() => { resolve(new Error('adapter failure did not settle promptly')) }, 100)
|
||||
})
|
||||
const caught = await Promise.race([failure, timeout])
|
||||
if (timer !== undefined) clearTimeout(timer)
|
||||
|
||||
expect(caught).toBe(original)
|
||||
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
|
||||
expect(cleanupCalls).toBe(0)
|
||||
})
|
||||
|
||||
it('awaits one adapter return on downstream close and leaves its rejection unclassified', async () => {
|
||||
const cleanup = new Error('cleanup failed')
|
||||
let cleanupCalls = 0
|
||||
const adapter = new class extends LlmAdapter {
|
||||
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
return {
|
||||
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
|
||||
return {
|
||||
next: () => Promise.resolve({ done: false, value: SCRIPT[0]! }),
|
||||
return: () => {
|
||||
cleanupCalls += 1
|
||||
return Promise.reject(cleanup)
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
ctx.llm.registerAdapter(['test-model'], adapter)
|
||||
|
||||
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
|
||||
let caught: unknown
|
||||
try {
|
||||
for await (const _chunk of stream) break
|
||||
} catch (error: unknown) {
|
||||
caught = error
|
||||
}
|
||||
|
||||
expect(caught).toBe(cleanup)
|
||||
expect(isLlmAdapterFailure(stream, caught)).toBe(false)
|
||||
expect(cleanupCalls).toBe(1)
|
||||
})
|
||||
|
||||
it('allows downstream close when the adapter iterator has no return method', async () => {
|
||||
const adapter = new class extends LlmAdapter {
|
||||
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
return {
|
||||
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
|
||||
return { next: () => Promise.resolve({ done: false, value: SCRIPT[0]! }) }
|
||||
},
|
||||
}
|
||||
}
|
||||
}()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
ctx.llm.registerAdapter(['test-model'], adapter)
|
||||
|
||||
let chunks = 0
|
||||
for await (const _chunk of ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })) {
|
||||
chunks += 1
|
||||
break
|
||||
}
|
||||
|
||||
expect(chunks).toBe(1)
|
||||
})
|
||||
|
||||
it('normalizes and tags non-Error adapter failures once', async () => {
|
||||
const adapter = new class extends LlmAdapter {
|
||||
stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
return {
|
||||
[Symbol.asyncIterator](): AsyncIterator<StreamChunk> {
|
||||
// Third-party adapters can reject with arbitrary values.
|
||||
// eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
|
||||
return { next: () => Promise.reject('plain provider failure') }
|
||||
},
|
||||
}
|
||||
}
|
||||
}()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
ctx.llm.registerAdapter(['test-model'], adapter)
|
||||
|
||||
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
|
||||
let caught: unknown
|
||||
try {
|
||||
for await (const _chunk of stream) { /* drain */ }
|
||||
} catch (error: unknown) {
|
||||
caught = error
|
||||
}
|
||||
|
||||
expect(caught).toBeInstanceOf(HarnessError)
|
||||
expect(caught).toMatchObject({ code: 'UNKNOWN', cause: 'plain provider failure' })
|
||||
expect(isLlmAdapterFailure(stream, caught)).toBe(true)
|
||||
})
|
||||
|
||||
it('does not tag a failure thrown downstream while consuming adapter output', async () => {
|
||||
const downstream = new Error('consumer failed')
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
ctx.llm.registerAdapter(['test-model'], new ScriptedAdapter(SCRIPT))
|
||||
|
||||
const stream = ctx.llm.stream({ provider: 'test-model', model: 'test-model', messages: [] })
|
||||
let caught: unknown
|
||||
try {
|
||||
for await (const _chunk of stream) throw downstream
|
||||
} catch (error: unknown) {
|
||||
caught = error
|
||||
}
|
||||
|
||||
expect(caught).toBe(downstream)
|
||||
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 () => {
|
||||
|
||||
@@ -42,6 +42,10 @@ Both plugins have usable defaults. A deployment with a different capacity config
|
||||
|
||||
Indirectly, through consumers such as `dsh-compact-basic`; the service itself adds no prompt, message, schema, tool, or model call.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
No direct invalidation; the named consumer owns any request-prefix changes.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **The fixed heuristic is approximate** — content without reusable provider usage is priced by character count plus structural overhead, not an exact provider tokenizer or request serializer.
|
||||
|
||||
Reference in New Issue
Block a user