Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(llm-deepseek): per-request connection resolution over settings + credentials

Browse Source 
The adapter takes an options thunk and a per-stream credential resolver
instead of frozen construction facts: base URL, catalog, defaults, idle
budget, and the API key re-resolve at each operation, so a settings or
credential change reaches the very next request while in-flight streams
keep the facts they started with. resolveAdapterOptions is the one explicit
resolve step (entry config fails loud at load; a live snapshot failing a
beyond-schema bound keeps the last good options). The plugin layers its
entry config under the optional llm-deepseek settings section and resolves
keys literal-first through ctx.credentials with an ambient env fallback; a
missing key now registers the route, warns, and fails each request with
actionable MISSING_CREDENTIAL instead of failing plugin load. The
registration-captured retry policy re-registers the route in place when it
changes.
This commit is contained in:
Yichen Jiang
2026-07-29 13:26:52 +08:00
parent aee06097ee
commit f05ab3f945
8 changed files with 522 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
packages/llm/llm-deepseek/src/adapter.ts
View File

@@ -1,21 +1,23 @@
/**
* `DeepSeekAdapter`: fetch + SSE against a DeepSeek (OpenAI-compatible)
* chat-completions endpoint, emitting harness StreamChunks.
* chat-completions endpoint, emitting harness StreamChunks. The adapter is
* transport-only: connection facts arrive through a thunk resolved once per
* operation and the bearer token through a per-request resolver, so the
* registering plugin owns validation, layering, and credential policy.
*
* @module dsh-llm-deepseek/adapter
*/
import { attributionHeaders, CONTEXT_WINDOW_EXCEEDED_CODE, isContextWindowExceededError, isQuotaExceededError, LlmAdapter, LlmError, ProviderRequestId, QUOTA_EXCEEDED_CODE, ReasoningEffortId, resolveRetryPolicy } from '@deepseek-ai/dsh-llm'
import { attributionHeaders, CONTEXT_WINDOW_EXCEEDED_CODE, isContextWindowExceededError, isQuotaExceededError, LlmAdapter, LlmError, ProviderRequestId, QUOTA_EXCEEDED_CODE, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import type {
GenerateOptions,
LlmModelInfo,
LlmProviderInfo,
LlmResolvedModelInfo,
ResolvedRetryPolicy,
RetryPolicyConfig,
StreamChunk,
} from '@deepseek-ai/dsh-llm'
import { idleWatchdog, MAX_TIMER_DELAY_MS, timeoutOf } from '@deepseek-ai/dsh-timeout'
import { idleWatchdog, timeoutOf } from '@deepseek-ai/dsh-timeout'
import { serializeRequest } from './serialize.ts'
import type { RequestDefaults } from './serialize.ts'
import { parseSse } from './sse.ts'
@@ -34,22 +36,37 @@ export interface DeepSeekCatalogModel {
contextWindow?: number
}
/** Constructor options for {@link DeepSeekAdapter}; the plugin's `apply` resolves them from Config + environment. */
export interface DeepSeekAdapterOptions {
/** Bearer token sent in the `authorization` header on every request. */
apiKey: string
/**
* Validated connection facts for one operation. The plugin's
* `resolveAdapterOptions` is the one explicit resolve step producing this
* shape; the adapter trusts it and re-reads it per operation, which is what
* makes a configuration change reach the next request without re-registration.
*/
export interface DeepSeekConnectionOptions {
/** Endpoint base; `/chat/completions` is appended. */
baseURL: string
/** Request defaults applied to every call (thinking mode, effort). */
defaults?: RequestDefaults
defaults: RequestDefaults
/** Positive context capacity used when the selected model has no exact value. */
defaultContextWindow?: number
/** Advisory models exposed to discovery consumers; requests remain unrestricted. */
models?: readonly DeepSeekCatalogModel[]
models: readonly DeepSeekCatalogModel[]
/** Maximum provider idle time while one stream read is outstanding. */
streamIdleTimeoutMs?: number
/** Provider-owned model-request retry policy; omission uses normal defaults. */
retryPolicy?: RetryPolicyConfig
streamIdleTimeoutMs: number
/** Provider-owned model-request retry policy, already resolved. */
retryPolicy: ResolvedRetryPolicy
}
/** Constructor options for {@link DeepSeekAdapter}: the two resolution seams the plugin owns. */
export interface DeepSeekAdapterOptions {
/** Current validated connection facts; called once per operation. */
options: () => DeepSeekConnectionOptions
/**
* Resolve the bearer token for one request; called once per stream call and
* frozen for that call. Throws `LlmError` `MISSING_CREDENTIAL` when no key
* is available anywhere.
*/
resolveApiKey: () => Promise<string>
}
/** Default maximum idle interval while an adapter stream read is outstanding. */
@@ -118,29 +135,8 @@ export function httpErrorCode(status: number, error?: WireError['error']): strin
* map to `ABORTED`; the configured per-read idle watchdog maps to `TIMEOUT`.
*/
export class DeepSeekAdapter extends LlmAdapter {
private readonly streamIdleTimeoutMs: number
private readonly retryPolicy: ResolvedRetryPolicy
constructor(private readonly options: DeepSeekAdapterOptions) {
constructor(private readonly config: DeepSeekAdapterOptions) {
super()
if (options.defaults?.thinking === 'disabled'
&& options.defaults.reasoningEffort !== undefined
&& options.defaults.reasoningEffort !== 'off') {
throw new Error('llm-deepseek: only reasoningEffort "off" can be configured when thinking is disabled')
}
if (options.defaultContextWindow !== undefined
&& (!Number.isInteger(options.defaultContextWindow) || options.defaultContextWindow <= 0)) {
throw new Error('llm-deepseek: defaultContextWindow must be a positive integer')
}
this.streamIdleTimeoutMs = options.streamIdleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT_MS
if (!Number.isFinite(this.streamIdleTimeoutMs)
|| this.streamIdleTimeoutMs <= 0
|| this.streamIdleTimeoutMs > MAX_TIMER_DELAY_MS) {
throw new Error(
`llm-deepseek: streamIdleTimeoutMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`,
)
}
this.retryPolicy = resolveRetryPolicy(options.retryPolicy, 'llm-deepseek: retryPolicy')
}
override providerInfo(provider: string): LlmProviderInfo {
@@ -148,11 +144,11 @@ export class DeepSeekAdapter extends LlmAdapter {
}
override providerRetryPolicy(_provider: string): ResolvedRetryPolicy {
return this.retryPolicy
return this.config.options().retryPolicy
}
override listModels(provider: string): Promise<readonly LlmModelInfo[]> {
return Promise.resolve((this.options.models ?? []).map(model => modelInfo(provider, model)))
return Promise.resolve(this.config.options().models.map(model => modelInfo(provider, model)))
}
override resolveModel(
@@ -160,15 +156,16 @@ export class DeepSeekAdapter extends LlmAdapter {
model: string,
_signal?: AbortSignal,
): Promise<LlmResolvedModelInfo> {
const configured = this.options.models?.find(entry => entry.id === model)
const connection = this.config.options()
const configured = connection.models.find(entry => entry.id === model)
const contextWindow = configured?.contextWindow
?? this.options.defaultContextWindow
?? connection.defaultContextWindow
return Promise.resolve({
...configured === undefined
? { provider, id: model, name: model }
: modelInfo(provider, configured),
...contextWindow === undefined ? {} : { context: { contextWindow } },
...this.options.defaults?.thinking === 'disabled'
...connection.defaults.thinking === 'disabled'
? {
reasoning: {
efforts: OFF_ONLY_REASONING_EFFORTS,
@@ -178,9 +175,9 @@ export class DeepSeekAdapter extends LlmAdapter {
: {
reasoning: {
efforts: REASONING_EFFORTS,
defaultEffort: this.options.defaults?.reasoningEffort === 'off'
defaultEffort: connection.defaults.reasoningEffort === 'off'
? OFF_REASONING_EFFORT
: this.options.defaults?.reasoningEffort === 'max'
: connection.defaults.reasoningEffort === 'max'
? MAX_REASONING_EFFORT
: HIGH_REASONING_EFFORT,
},
@@ -189,12 +186,17 @@ export class DeepSeekAdapter extends LlmAdapter {
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
// One resolution per stream call: connection facts and the credential
// freeze here and hold for this whole request, so an in-flight stream
// never observes a configuration change and the next call re-resolves.
const connection = this.config.options()
const apiKey = await this.config.resolveApiKey()
const consumer = new AbortController()
const upstream = options.signal === undefined
? consumer.signal
: AbortSignal.any([options.signal, consumer.signal])
using watchdog = idleWatchdog(upstream, this.streamIdleTimeoutMs, STREAM_IDLE_TIMEOUT_CODE)
const iterator = this.request(options, watchdog.signal)[Symbol.asyncIterator]()
using watchdog = idleWatchdog(upstream, connection.streamIdleTimeoutMs, STREAM_IDLE_TIMEOUT_CODE)
const iterator = this.request(options, watchdog.signal, connection, apiKey)[Symbol.asyncIterator]()
let exhausted = false
try {
while (true) {
@@ -208,7 +210,7 @@ export class DeepSeekAdapter extends LlmAdapter {
} catch (error: unknown) {
if (timeoutOf(watchdog.signal, STREAM_IDLE_TIMEOUT_CODE) !== undefined) {
throw new LlmError(
`DeepSeek stream idle timeout after ${this.streamIdleTimeoutMs}ms`,
`DeepSeek stream idle timeout after ${connection.streamIdleTimeoutMs}ms`,
'TIMEOUT',
{ cause: error },
)
@@ -217,7 +219,7 @@ export class DeepSeekAdapter extends LlmAdapter {
throw new LlmError('DeepSeek request aborted by caller', 'ABORTED', { cause: error })
}
if (error instanceof LlmError) throw error
throw new LlmError(`DeepSeek API stream from ${this.options.baseURL} failed`, 'TRANSPORT', { cause: error })
throw new LlmError(`DeepSeek API stream from ${connection.baseURL} failed`, 'TRANSPORT', { cause: error })
} finally {
consumer.abort('DeepSeek stream consumer stopped')
if (!exhausted && iterator.return !== undefined) {
@@ -230,13 +232,18 @@ export class DeepSeekAdapter extends LlmAdapter {
}
}
private async * request(options: GenerateOptions, signal: AbortSignal): AsyncIterable<StreamChunk> {
const body = serializeRequest(options, this.options.defaults ?? {})
private async * request(
options: GenerateOptions,
signal: AbortSignal,
connection: DeepSeekConnectionOptions,
apiKey: string,
): AsyncIterable<StreamChunk> {
const body = serializeRequest(options, connection.defaults)
// Prepared outside the try so the TRANSPORT label below covers exactly the
// transport boundary, never a serialization failure.
const payload = JSON.stringify(body)
const headers = {
'authorization': `Bearer ${this.options.apiKey}`,
'authorization': `Bearer ${apiKey}`,
'content-type': 'application/json',
'accept': 'text/event-stream',
...attributionHeaders(),
@@ -252,7 +259,7 @@ export class DeepSeekAdapter extends LlmAdapter {
// outweighs its additional runtime dependencies.
let response: Response
try {
response = await fetch(`${this.options.baseURL}/chat/completions`, {
response = await fetch(`${connection.baseURL}/chat/completions`, {
method: 'POST',
headers,
body: payload,
@@ -266,7 +273,7 @@ export class DeepSeekAdapter extends LlmAdapter {
// lives on `cause`. Wrapping with the endpoint and chaining the cause
// lets `errorChain` render the full diagnosis at every reporting seam.
throw new LlmError(
`DeepSeek API request to ${this.options.baseURL} failed`,
`DeepSeek API request to ${connection.baseURL} failed`,
'TRANSPORT',
{ cause: error },
)