diff --git a/packages/llm/llm-deepseek/package.json b/packages/llm/llm-deepseek/package.json index d233ef7764..2c39e2d920 100644 --- a/packages/llm/llm-deepseek/package.json +++ b/packages/llm/llm-deepseek/package.json @@ -27,8 +27,10 @@ ], "license": "BSD-3-Clause", "peerDependencies": { + "@deepseek-ai/dsh-credentials": "^0.0.1", "@deepseek-ai/dsh-invariants": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", + "@deepseek-ai/dsh-settings": "^0.0.1", "@deepseek-ai/dsh-timeout": "^0.0.1", "cordis": "^4.0.0-rc.7" }, @@ -37,8 +39,10 @@ "schemastery": "^3.18.0" }, "devDependencies": { + "@deepseek-ai/dsh-credentials": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", + "@deepseek-ai/dsh-settings": "workspace:^", "@deepseek-ai/dsh-timeout": "workspace:^", "cordis": "^4.0.0-rc.7" } diff --git a/packages/llm/llm-deepseek/src/adapter.ts b/packages/llm/llm-deepseek/src/adapter.ts index ff5ce9bf72..0163dd3cab 100644 --- a/packages/llm/llm-deepseek/src/adapter.ts +++ b/packages/llm/llm-deepseek/src/adapter.ts @@ -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 } /** 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 { - 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 { - 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 { + // 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 { - const body = serializeRequest(options, this.options.defaults ?? {}) + private async * request( + options: GenerateOptions, + signal: AbortSignal, + connection: DeepSeekConnectionOptions, + apiKey: string, + ): AsyncIterable { + 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 }, ) diff --git a/packages/llm/llm-deepseek/src/index.ts b/packages/llm/llm-deepseek/src/index.ts index 00db46d642..054f1984b0 100644 --- a/packages/llm/llm-deepseek/src/index.ts +++ b/packages/llm/llm-deepseek/src/index.ts @@ -1,41 +1,56 @@ /** - * Register a {@link DeepSeekAdapter} for the `deepseek` provider route on `ctx.llm`. Configuration uses - * Cordis schemastery; pass secrets from environment variables through `cordis.yml` with `!!js`, - * as shown in the package README, rather than reading ad hoc files. + * Register a {@link DeepSeekAdapter} for the `deepseek` provider route on + * `ctx.llm`, with connection facts resolved per request instead of frozen at + * load: the plugin layers its `cordis.yml` entry config under the optional + * `llm-deepseek` user-settings section (`ctx.settings`) and resolves the API + * key through the optional credential seam (`ctx.credentials`), so a changed + * base URL, catalog, or key reaches the very next request without restarting + * anything, while an in-flight stream keeps the facts it started with. The + * one registration-captured fact — the retry policy — re-registers the route + * in place when it changes. * @module @deepseek-ai/dsh-llm-deepseek */ import type { Context } from 'cordis' import z from 'schemastery' -import { RetryPolicySchema } from '@deepseek-ai/dsh-llm' +import { LlmError, resolveRetryPolicy, RetryPolicySchema } from '@deepseek-ai/dsh-llm' import type { RetryPolicyConfig } from '@deepseek-ai/dsh-llm' +import { credentialRef } from '@deepseek-ai/dsh-credentials' +import type { CredentialRef } from '@deepseek-ai/dsh-credentials' +import { deepEqualJson, settingsNamespace } from '@deepseek-ai/dsh-settings' import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' import { DEFAULT_STREAM_IDLE_TIMEOUT_MS, DeepSeekAdapter } from './adapter.ts' -import type { DeepSeekCatalogModel } from './adapter.ts' +import type { DeepSeekCatalogModel, DeepSeekConnectionOptions } from './adapter.ts' export { DeepSeekAdapter } from './adapter.ts' -export type { DeepSeekAdapterOptions, DeepSeekCatalogModel } from './adapter.ts' +export type { DeepSeekAdapterOptions, DeepSeekCatalogModel, DeepSeekConnectionOptions } from './adapter.ts' export type { RequestDefaults } from './serialize.ts' export type * from './types.ts' export const name = 'llm-deepseek' export const inject = ['llm'] +const NS = settingsNamespace('llm-deepseek') +const DEFAULT_API_KEY_ENV = 'DEEPSEEK_API_KEY' + const DEFAULT_MODELS: DeepSeekCatalogModel[] = [ { id: 'deepseek-v4-flash', name: 'DeepSeek-V4-Flash', contextWindow: 256_000 }, { id: 'deepseek-v4-pro', name: 'DeepSeek-V4-Pro', contextWindow: 256_000 }, ] /** - * Plugin config, validated by the same-named schemastery schema. Every field - * is optional in yml: credentials/endpoint fall back to the environment (a - * missing API key fails plugin load, not the first call), omitted thinking - * mode uses the provider default, and omitted reasoning effort resolves to - * `high`. + * Plugin config, validated by the same-named schemastery schema and doubling + * as the `llm-deepseek` settings-section shape. Every field is optional in + * yml: a missing API key resolves through {@link Config.apiKeyEnv} at each + * request (a request without any key fails with `MISSING_CREDENTIAL`, not at + * plugin load), omitted thinking mode uses the provider default, and omitted + * reasoning effort resolves to `high`. */ export interface Config { - /** API key; falls back to $DEEPSEEK_API_KEY. Required one way or the other. */ + /** Literal API key; prefer {@link apiKeyEnv} so no secret enters configuration files. */ apiKey?: string + /** Credential reference (environment-variable name) resolved per request; defaults to `DEEPSEEK_API_KEY`. */ + apiKeyEnv?: string /** Endpoint base; falls back to $DEEPSEEK_BASE_URL, then the public API. */ baseURL?: string /** Deployment thinking policy; `disabled` limits every conversation request to `off`. */ @@ -60,7 +75,8 @@ const catalogModel: z = z.object({ }) export const Config: z = z.object({ - apiKey: z.string(), + apiKey: z.string().role('secret'), + apiKeyEnv: z.string().default(DEFAULT_API_KEY_ENV), baseURL: z.string(), thinking: z.union(['enabled', 'disabled']), reasoningEffort: z.union(['off', 'high', 'max']), @@ -73,6 +89,12 @@ export const Config: z = z.object({ /** Public API default; the internal endpoint comes from $DEEPSEEK_BASE_URL. */ export const PUBLIC_BASE_URL = 'https://api.deepseek.com' +/** Connection facts plus the plugin-consumed credential reference. */ +export interface ResolvedDeepSeekOptions extends DeepSeekConnectionOptions { + /** Reference resolved per request when no literal key is configured. */ + apiKeyEnv: CredentialRef +} + /** Resolve, validate, and detach the advisory model catalog. */ function resolveModels(models: readonly DeepSeekCatalogModel[] | undefined): DeepSeekCatalogModel[] { const seen = new Set() @@ -98,20 +120,35 @@ function resolveModels(models: readonly DeepSeekCatalogModel[] | undefined): Dee }) } -export function apply(ctx: Context, config: Config): void { +/** + * The one explicit resolve step from raw config to validated connection + * facts. Programmatic construction may bypass Schemastery normalization, so + * every default and bound is re-judged here — for the composition entry at + * load (fail loud) and for each settings snapshot at its first use. + * @param config - raw plugin config or resolved settings snapshot. + * @returns validated connection facts plus the credential reference. + */ +export function resolveAdapterOptions(config: Config): ResolvedDeepSeekOptions { if (config.thinking === 'disabled' && config.reasoningEffort !== undefined && config.reasoningEffort !== 'off') { throw new Error('llm-deepseek: only reasoningEffort "off" can be configured when thinking is disabled') } - const apiKey = config.apiKey ?? process.env.DEEPSEEK_API_KEY - if (apiKey === undefined || apiKey.length === 0) { - throw new Error('llm-deepseek: an API key is required (Config.apiKey or $DEEPSEEK_API_KEY)') + if (config.defaultContextWindow !== undefined + && (!Number.isInteger(config.defaultContextWindow) || config.defaultContextWindow <= 0)) { + throw new Error('llm-deepseek: defaultContextWindow must be a positive integer') } - const baseURL = config.baseURL ?? process.env.DEEPSEEK_BASE_URL ?? PUBLIC_BASE_URL - ctx.llm.registerAdapter(['deepseek'], new DeepSeekAdapter({ - apiKey, - baseURL, + const streamIdleTimeoutMs = config.streamIdleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT_MS + if (!Number.isFinite(streamIdleTimeoutMs) + || streamIdleTimeoutMs <= 0 + || streamIdleTimeoutMs > MAX_TIMER_DELAY_MS) { + throw new Error( + `llm-deepseek: streamIdleTimeoutMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`, + ) + } + return { + apiKeyEnv: credentialRef(config.apiKeyEnv ?? DEFAULT_API_KEY_ENV), + baseURL: config.baseURL ?? process.env.DEEPSEEK_BASE_URL ?? PUBLIC_BASE_URL, defaults: { thinking: config.thinking, reasoningEffort: config.reasoningEffort, @@ -120,7 +157,92 @@ export function apply(ctx: Context, config: Config): void { ? {} : { defaultContextWindow: config.defaultContextWindow }, models: resolveModels(config.models), - streamIdleTimeoutMs: config.streamIdleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT_MS, - ...config.retryPolicy === undefined ? {} : { retryPolicy: config.retryPolicy }, - })) + streamIdleTimeoutMs, + retryPolicy: resolveRetryPolicy(config.retryPolicy, 'llm-deepseek: retryPolicy'), + } +} + +export function apply(ctx: Context, config: Config): void { + let current: () => Config = () => config + let lastRaw: Config | undefined + let lastGood: ResolvedDeepSeekOptions | undefined + const options = (): ResolvedDeepSeekOptions => { + const raw = current() + if (raw === lastRaw && lastGood !== undefined) return lastGood + try { + const next = resolveAdapterOptions(raw) + lastRaw = raw + lastGood = next + return next + } catch (error) { + // Static composition resolves before anything registers, so this branch + // only sees a live settings snapshot failing a beyond-schema bound: + // keep serving the last good facts and say so once per bad snapshot. + if (lastGood === undefined) throw error + lastRaw = raw + ctx.logger.error('llm-deepseek: keeping the last good configuration after an invalid settings section') + ctx.logger.error(error) + return lastGood + } + } + options() + + const resolveApiKey = async (): Promise => { + const raw = current() + if (raw.apiKey !== undefined && raw.apiKey.length > 0) return raw.apiKey + const ref = options().apiKeyEnv + const credentials = ctx.get('credentials') + if (credentials !== undefined) { + const hit = await credentials.resolve(ref) + if (hit !== undefined) return hit.value + } else { + // Without the seam, keep the historical ambient fallback so a plain + // cordis.yml composition works from the environment alone. + const ambient = process.env[ref] + if (ambient !== undefined && ambient.length > 0) return ambient + } + throw new LlmError( + 'llm-deepseek: no API key for provider route "deepseek"; set the llm-deepseek "apiKey" setting,' + + ` store ${ref} with the credentials service, or export ${ref}`, + 'MISSING_CREDENTIAL', + ) + } + + const adapter = new DeepSeekAdapter({ options, resolveApiKey }) + // Route effects bind to this apply fiber via the stable `ctx` reference, + // even when a swap runs inside the scoped settings callback below. + let disposeRoute = ctx.llm.registerAdapter(['deepseek'], adapter) + let registeredPolicy = options().retryPolicy + const ensureRegistrationFacts = (): void => { + const policy = options().retryPolicy + if (deepEqualJson(policy, registeredPolicy)) return + // The registry captures the retry policy at registration, so it is the one + // fact per-request resolution cannot refresh: swap the registration in one + // synchronous section (same adapter instance, no NO_ADAPTER window). + disposeRoute() + disposeRoute = ctx.llm.registerAdapter(['deepseek'], adapter) + registeredPolicy = policy + } + + void resolveApiKey().then(() => undefined, () => { + // Expected on a first boot with dynamic sources: the route stays + // registered (the catalog is browsable) and each request fails with the + // actionable MISSING_CREDENTIAL message until a key arrives. + ctx.logger.warn('llm-deepseek: no API key resolved yet for route "deepseek"; requests will fail until one is configured') + }) + + ctx.inject(['settings'], (sctx) => { + const scope = sctx.settings.register(NS, Config, { base: config }) + current = () => scope.get() + sctx.effect(() => () => { + // Settings detached (provider disposed or reloading): fall back to the + // composition entry so the plugin keeps working exactly as configured. + current = () => config + ensureRegistrationFacts() + }) + ensureRegistrationFacts() + scope.watch(() => { + ensureRegistrationFacts() + }) + }) } diff --git a/packages/llm/llm-deepseek/tests/adapter.spec.ts b/packages/llm/llm-deepseek/tests/adapter.spec.ts index 1b64c57982..935235d825 100644 --- a/packages/llm/llm-deepseek/tests/adapter.spec.ts +++ b/packages/llm/llm-deepseek/tests/adapter.spec.ts @@ -1,5 +1,3 @@ -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, { createUserMessage, @@ -14,90 +12,18 @@ import LlmService, { createUserMessage, import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout' import { SessionId } from '@deepseek-ai/dsh-session' import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek' -import { DeepSeekAdapter } from '@deepseek-ai/dsh-llm-deepseek' +import { DeepSeekAdapter, resolveAdapterOptions } from '@deepseek-ai/dsh-llm-deepseek' import { httpErrorCode } from '../src/adapter.ts' import { assemble } from './assemble.ts' - -/** One scripted behavior for the next request the mock server receives. */ -type Behavior = - | { kind: 'sse'; events: string[]; delayMs?: number } - | { kind: 'http-error'; status: number; body: string; contentType?: string; headers?: Record } - | { kind: 'close-early'; events: string[] } - -interface MockServer { - url: string - /** Bodies of received requests, in order. */ - requests: unknown[] - /** Header bags of received requests, in order (parallel to `requests`). */ - headers: IncomingMessage['headers'][] - script: Behavior[] - close(): Promise -} - -const servers: Server[] = [] +import { closeMockServers, mockServer, textEvents } from './mock-server.ts' +import type { Behavior } from './mock-server.ts' afterEach(async () => { - await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve)))) + await closeMockServers() vi.unstubAllEnvs() vi.useRealTimers() }) -/** Local chat-completions stand-in: replays scripted behaviors per request. */ -async function mockServer(script: Behavior[]): Promise { - const requests: unknown[] = [] - const headers: IncomingMessage['headers'][] = [] - const server = createServer((request: IncomingMessage, response: ServerResponse) => { - let body = '' - request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') }) - request.on('end', () => { - requests.push(JSON.parse(body)) - headers.push(request.headers) - const behavior = script.shift() - if (!behavior) { - response.writeHead(500).end('mock script exhausted') - return - } - if (behavior.kind === 'http-error') { - response.writeHead(behavior.status, { - 'content-type': behavior.contentType ?? 'application/json', - ...behavior.headers, - }) - response.end(behavior.body) - return - } - response.writeHead(200, { 'content-type': 'text/event-stream' }) - const write = (index: number): void => { - if (index >= behavior.events.length) { - if (behavior.kind === 'sse') response.end() - else response.destroy() // close-early: drop the socket mid-stream - return - } - response.write(`data: ${behavior.events[index]}\n\n`) - setTimeout(() => { write(index + 1) }, behavior.kind === 'sse' ? behavior.delayMs ?? 0 : 5) - } - write(0) - }) - }) - servers.push(server) - await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) - const address = server.address() - if (address === null || typeof address === 'string') throw new Error('no port') - return { - url: `http://127.0.0.1:${address.port}`, - requests, - headers, - script, - close: () => new Promise(resolve => server.close(() => { resolve() })), - } -} - -const textEvents = [ - '{"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}', - '{"choices":[{"delta":{"content":"hello"}}]}', - '{"choices":[{"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}', - '[DONE]', -] - async function harness(baseURL: string, config: object = {}) { const ctx = new Context() await ctx.plugin(LlmService) @@ -105,6 +31,15 @@ async function harness(baseURL: string, config: object = {}) { return ctx } +/** Direct adapter over the plugin's real resolve step, with a static key. */ +function adapterOf(config: Partial & { apiKey?: string } = {}): DeepSeekAdapter { + const { apiKey, ...rest } = config + return new DeepSeekAdapter({ + options: () => resolveAdapterOptions(rest), + resolveApiKey: () => Promise.resolve(apiKey ?? 'k'), + }) +} + describe('DeepSeekAdapter against a mock server', () => { it('streams a text generation end to end through the assembler', async () => { const server = await mockServer([{ kind: 'sse', events: textEvents }]) @@ -275,11 +210,7 @@ describe('DeepSeekAdapter against a mock server', () => { 'rejects direct adapter effort %s before I/O when thinking is disabled', async (effort) => { const server = await mockServer([]) - const adapter = new DeepSeekAdapter({ - apiKey: 'test-key', - baseURL: server.url, - defaults: { thinking: 'disabled' }, - }) + const adapter = adapterOf({ apiKey: 'test-key', baseURL: server.url, thinking: 'disabled' }) const stream = adapter.stream({ provider: 'deepseek', @@ -483,7 +414,7 @@ describe('DeepSeekAdapter against a mock server', () => { }) it('throws EMPTY_RESPONSE when the response has no body', async () => { - const adapter = new DeepSeekAdapter({ apiKey: 'k', baseURL: 'http://127.0.0.1:1' }) + const adapter = adapterOf({ baseURL: 'http://127.0.0.1:1' }) const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue( new Response(null, { status: 200 }), ) @@ -538,7 +469,7 @@ describe('DeepSeekAdapter against a mock server', () => { it('maps connection failures to TRANSPORT without losing the cause', async () => { const cause = new TypeError('connection refused') const fetchSpy = vi.spyOn(globalThis, 'fetch').mockRejectedValue(cause) - const adapter = new DeepSeekAdapter({ apiKey: 'k', baseURL: 'https://example.invalid' }) + const adapter = adapterOf({ baseURL: 'https://example.invalid' }) try { const drain = async (): Promise => { for await (const _chunk of adapter.stream({ provider: 'deepseek', model: 'm', messages: [] })) { /* drain */ } @@ -555,7 +486,7 @@ describe('DeepSeekAdapter against a mock server', () => { failed.reject('offline') return failed.promise }) - const adapter = new DeepSeekAdapter({ apiKey: 'k', baseURL: 'https://example.invalid' }) + const adapter = adapterOf({ baseURL: 'https://example.invalid' }) try { const drain = async (): Promise => { for await (const _chunk of adapter.stream({ provider: 'deepseek', model: 'm', messages: [] })) { /* drain */ } @@ -585,11 +516,7 @@ describe('DeepSeekAdapter against a mock server', () => { }) return Promise.resolve(new Response(body, { status: 200 })) }) - const adapter = new DeepSeekAdapter({ - apiKey: 'k', - baseURL: 'https://example.invalid', - streamIdleTimeoutMs: 100, - }) + const adapter = adapterOf({ baseURL: 'https://example.invalid', streamIdleTimeoutMs: 100 }) try { const drain = (async () => { for await (const _chunk of adapter.stream({ provider: 'deepseek', model: 'm', messages: [] })) { /* drain */ } @@ -733,22 +660,15 @@ describe('plugin registration and config', () => { ) it.each(['high', 'max'] as const)( - 'rejects disabled-thinking effort %s at the direct constructor boundary', + 'rejects disabled-thinking effort %s at the resolver boundary', (reasoningEffort) => { - expect(() => new DeepSeekAdapter({ - apiKey: 'k', - baseURL: 'http://127.0.0.1:1', - defaults: { thinking: 'disabled', reasoningEffort }, - })).toThrow(/only reasoningEffort "off"/) + expect(() => resolveAdapterOptions({ thinking: 'disabled', reasoningEffort })) + .toThrow(/only reasoningEffort "off"/) }, ) - it('accepts disabled thinking with off at the direct constructor boundary', async () => { - const adapter = new DeepSeekAdapter({ - apiKey: 'k', - baseURL: 'http://127.0.0.1:1', - defaults: { thinking: 'disabled', reasoningEffort: 'off' }, - }) + it('accepts disabled thinking with off at the resolver boundary', async () => { + const adapter = adapterOf({ thinking: 'disabled', reasoningEffort: 'off' }) await expect(adapter.resolveModel('deepseek', 'pass-through')).resolves.toMatchObject({ reasoning: { efforts: [{ id: ReasoningEffortId('off'), name: 'Off' }], @@ -863,11 +783,8 @@ describe('plugin registration and config', () => { it.each([0, 1.5])( 'rejects invalid adapter-wide default context capacity %s', async (defaultContextWindow) => { - expect(() => new DeepSeekAdapter({ - apiKey: 'k', - baseURL: 'http://127.0.0.1:1', - defaultContextWindow, - })).toThrow(/defaultContextWindow must be a positive integer/) + expect(() => resolveAdapterOptions({ defaultContextWindow })) + .toThrow(/defaultContextWindow must be a positive integer/) const ctx = new Context() await ctx.plugin(LlmService) @@ -889,13 +806,19 @@ describe('plugin registration and config', () => { expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }]) }) - it('throws a clear error when no API key is available', async () => { + it('loads keyless, keeps the catalog browsable, and fails the request actionably', async () => { vi.stubEnv('DEEPSEEK_API_KEY', '') const ctx = new Context() await ctx.plugin(LlmService) - await expect(ctx.plugin(LlmDeepSeek, {})) - .rejects.toThrow(/an API key is required/) - expect(ctx.llm.listProviders()).toEqual([]) + await ctx.plugin(LlmDeepSeek, { baseURL: 'http://127.0.0.1:1' }) + // First-boot onboarding: the route registers so models stay discoverable; + // only the request itself needs a key. + expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }]) + await expect(ctx.llm.listModels('deepseek')).resolves.toHaveLength(2) + await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })) + .rejects.toMatchObject({ code: 'MISSING_CREDENTIAL' }) + await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })) + .rejects.toThrow(/store DEEPSEEK_API_KEY with the credentials service, or export DEEPSEEK_API_KEY/) }) it('prefers explicit config over env for key and base URL', async () => { @@ -927,23 +850,32 @@ describe('plugin registration and config', () => { expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }]) }) - it('adapter is constructible directly for embedding', async () => { - const adapter = new DeepSeekAdapter({ apiKey: 'k', baseURL: 'http://127.0.0.1:1' }) + it('adapter is constructible directly for embedding over the shared resolver', async () => { + const adapter = adapterOf() expect(adapter).toBeInstanceOf(DeepSeekAdapter) - await expect(adapter.listModels('deepseek')).resolves.toEqual([]) + // Direct embedding shares the plugin's one resolve step, so it advertises + // the same default catalog instead of a divergent empty one. + await expect(adapter.listModels('deepseek')).resolves.toHaveLength(2) + }) + + it('resolves connection facts and the credential exactly once per stream call', async () => { + const server = await mockServer([{ kind: 'sse', events: textEvents }]) + const options = vi.fn(() => resolveAdapterOptions({ baseURL: server.url })) + const resolveApiKey = vi.fn(() => Promise.resolve('per-request-key')) + const adapter = new DeepSeekAdapter({ options, resolveApiKey }) + + for await (const _chunk of adapter.stream({ provider: 'deepseek', model: 'm', messages: [] })) { /* drain */ } + + expect(options).toHaveBeenCalledTimes(1) + expect(resolveApiKey).toHaveBeenCalledTimes(1) + expect(server.headers[0]?.authorization).toBe('Bearer per-request-key') }) it('rejects invalid idle watchdog bounds for direct and plugin composition', async () => { - expect(() => new DeepSeekAdapter({ - apiKey: 'k', - baseURL: 'http://127.0.0.1:1', - streamIdleTimeoutMs: Number.POSITIVE_INFINITY, - })).toThrow(/streamIdleTimeoutMs.*positive finite/) - expect(() => new DeepSeekAdapter({ - apiKey: 'k', - baseURL: 'http://127.0.0.1:1', - streamIdleTimeoutMs: MAX_TIMER_DELAY_MS + 1, - })).toThrow(/streamIdleTimeoutMs.*no greater/) + expect(() => resolveAdapterOptions({ streamIdleTimeoutMs: Number.POSITIVE_INFINITY })) + .toThrow(/streamIdleTimeoutMs.*positive finite/) + expect(() => resolveAdapterOptions({ streamIdleTimeoutMs: MAX_TIMER_DELAY_MS + 1 })) + .toThrow(/streamIdleTimeoutMs.*no greater/) const ctx = new Context() await ctx.plugin(LlmService) diff --git a/packages/llm/llm-deepseek/tests/dynamic-config.spec.ts b/packages/llm/llm-deepseek/tests/dynamic-config.spec.ts new file mode 100644 index 0000000000..79a8afb671 --- /dev/null +++ b/packages/llm/llm-deepseek/tests/dynamic-config.spec.ts @@ -0,0 +1,163 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import LlmService from '@deepseek-ai/dsh-llm' +import { credentialRef } from '@deepseek-ai/dsh-credentials' +import { CredentialsLocal } from '@deepseek-ai/dsh-credentials-local' +import { settingsNamespace } from '@deepseek-ai/dsh-settings' +import { SettingsLocal } from '@deepseek-ai/dsh-settings-local' +import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek' +import { assemble } from './assemble.ts' +import { closeMockServers, mockServer, textEvents } from './mock-server.ts' + +const NS = settingsNamespace('llm-deepseek') +const KEY_REF = credentialRef('DEEPSEEK_API_KEY') + +const cleanups: Array<() => Promise> = [] + +afterEach(async () => { + while (cleanups.length > 0) await cleanups.pop()!() + await closeMockServers() + vi.unstubAllEnvs() +}) + +async function home(): Promise { + const dir = await mkdtemp(join(tmpdir(), 'dsh-llm-dynamic-')) + cleanups.push(() => rm(dir, { recursive: true, force: true })) + return dir +} + +interface Harness { + ctx: Context + settingsFiber: { dispose(): Promise } +} + +/** + * Real dynamic composition: llm + settings-local + credentials-local + + * llm-deepseek over one temp harness home. `watch: false` keeps every change + * flowing through the in-process write path, which is deterministic; external + * file watching is the providers' own covered concern. + */ +async function boot(dir: string, config: object): Promise { + const ctx = new Context() + cleanups.push(async () => { + await ctx.fiber.dispose() + }) + await ctx.plugin(LlmService) + const settingsFiber = ctx.plugin(SettingsLocal, { path: join(dir, 'settings.yaml'), watch: false }) + await settingsFiber + await ctx.plugin(CredentialsLocal, { path: join(dir, '.env'), watch: false }) + await ctx.plugin(LlmDeepSeek, config) + return { ctx, settingsFiber } +} + +function prompt(ctx: Context) { + return assemble(ctx, { model: 'deepseek-v4-flash', messages: [] }) +} + +describe('request-level dynamic configuration', () => { + it('routes the next request with the freshly resolved base URL and credential', async () => { + vi.stubEnv('DEEPSEEK_API_KEY', '') + const dir = await home() + await writeFile(join(dir, '.env'), 'DEEPSEEK_API_KEY=first-key\n') + const serverA = await mockServer([{ kind: 'sse', events: textEvents }]) + const serverB = await mockServer([{ kind: 'sse', events: textEvents }]) + const { ctx } = await boot(dir, { baseURL: serverA.url }) + + await prompt(ctx) + expect(serverA.headers[0]?.authorization).toBe('Bearer first-key') + + await ctx.settings.update(NS, { baseURL: serverB.url }) + await ctx.credentials.set(KEY_REF, 'second-key') + + await prompt(ctx) + // No restart, no re-registration: the next request resolved both facts. + expect(serverA.requests).toHaveLength(1) + expect(serverB.headers[0]?.authorization).toBe('Bearer second-key') + }) + + it('prefers a literal settings apiKey over the credential layers', async () => { + vi.stubEnv('DEEPSEEK_API_KEY', '') + const dir = await home() + await writeFile(join(dir, '.env'), 'DEEPSEEK_API_KEY=file-key\n') + const server = await mockServer([{ kind: 'sse', events: textEvents }]) + const { ctx } = await boot(dir, { baseURL: server.url }) + + await ctx.settings.update(NS, { apiKey: 'literal-key' }) + await prompt(ctx) + expect(server.headers[0]?.authorization).toBe('Bearer literal-key') + }) + + it('starts keyless and serves the next request once the key arrives', async () => { + vi.stubEnv('DEEPSEEK_API_KEY', '') + const dir = await home() + const server = await mockServer([{ kind: 'sse', events: textEvents }]) + const { ctx } = await boot(dir, { baseURL: server.url }) + + await expect(prompt(ctx)).rejects.toMatchObject({ code: 'MISSING_CREDENTIAL' }) + await ctx.credentials.set(KEY_REF, 'sk-arrived') + await prompt(ctx) + expect(server.headers[0]?.authorization).toBe('Bearer sk-arrived') + }) + + it('advertises a live settings catalog without re-registration', async () => { + const dir = await home() + const { ctx } = await boot(dir, { apiKey: 'k', baseURL: 'http://127.0.0.1:1' }) + + await expect(ctx.llm.listModels('deepseek')).resolves.toHaveLength(2) + await ctx.settings.update(NS, { models: [{ id: 'settings-model', name: 'From Settings' }] }) + await expect(ctx.llm.listModels('deepseek')).resolves.toEqual([ + { provider: 'deepseek', id: 'settings-model', name: 'From Settings' }, + ]) + }) + + it('re-registers the route in place when the captured retry policy changes', async () => { + const dir = await home() + const { ctx } = await boot(dir, { apiKey: 'k', baseURL: 'http://127.0.0.1:1' }) + + await ctx.settings.update(NS, { + retryPolicy: { mode: 'always', backoff: { initialDelayMs: 25, maxDelayMs: 100, jitterRatio: 0.2 } }, + }) + expect(ctx.llm.providerRetryPolicy('deepseek')).toEqual({ + mode: 'always', + initialDelayMs: 25, + maxDelayMs: 100, + jitterRatio: 0.2, + }) + expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }]) + }) + + it('keeps the last good options when a settings snapshot fails beyond-schema validation', async () => { + const dir = await home() + const { ctx } = await boot(dir, { apiKey: 'k', baseURL: 'http://127.0.0.1:1' }) + + // Schema-valid but resolver-invalid: duplicate catalog ids pass the array + // schema and fail the explicit resolve step. + await ctx.settings.update(NS, { models: [{ id: 'dup' }, { id: 'dup' }] }) + await expect(ctx.llm.listModels('deepseek')).resolves.toHaveLength(2) + await ctx.settings.update(NS, { models: [{ id: 'recovered' }] }) + await expect(ctx.llm.listModels('deepseek')).resolves.toEqual([ + { provider: 'deepseek', id: 'recovered', name: 'recovered' }, + ]) + }) + + it('falls back to the composition entry when settings detach', async () => { + vi.stubEnv('DEEPSEEK_API_KEY', '') + const dir = await home() + await writeFile(join(dir, '.env'), 'DEEPSEEK_API_KEY=steady-key\n') + const serverA = await mockServer([{ kind: 'sse', events: textEvents }]) + const serverB = await mockServer([{ kind: 'sse', events: textEvents }]) + const { ctx, settingsFiber } = await boot(dir, { baseURL: serverA.url }) + + await ctx.settings.update(NS, { baseURL: serverB.url }) + await prompt(ctx) + expect(serverB.requests).toHaveLength(1) + + await settingsFiber.dispose() + await prompt(ctx) + expect(serverA.requests).toHaveLength(1) + expect(serverA.headers[0]?.authorization).toBe('Bearer steady-key') + }) +}) diff --git a/packages/llm/llm-deepseek/tests/mock-server.ts b/packages/llm/llm-deepseek/tests/mock-server.ts new file mode 100644 index 0000000000..cdb499e143 --- /dev/null +++ b/packages/llm/llm-deepseek/tests/mock-server.ts @@ -0,0 +1,82 @@ +import { createServer } from 'node:http' +import type { IncomingMessage, Server, ServerResponse } from 'node:http' + +/** One scripted behavior for the next request the mock server receives. */ +export type Behavior = + | { kind: 'sse'; events: string[]; delayMs?: number } + | { kind: 'http-error'; status: number; body: string; contentType?: string; headers?: Record } + | { kind: 'close-early'; events: string[] } + +export interface MockServer { + url: string + /** Bodies of received requests, in order. */ + requests: unknown[] + /** Header bags of received requests, in order (parallel to `requests`). */ + headers: IncomingMessage['headers'][] + script: Behavior[] + close(): Promise +} + +const servers: Server[] = [] + +/** Close every server opened since the last call; run from each spec's afterEach. */ +export async function closeMockServers(): Promise { + await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve)))) +} + +/** A minimal complete text generation, reused by request-shape assertions. */ +export const textEvents = [ + '{"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}', + '{"choices":[{"delta":{"content":"hello"}}]}', + '{"choices":[{"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}', + '[DONE]', +] + +/** Local chat-completions stand-in: replays scripted behaviors per request. */ +export async function mockServer(script: Behavior[]): Promise { + const requests: unknown[] = [] + const headers: IncomingMessage['headers'][] = [] + const server = createServer((request: IncomingMessage, response: ServerResponse) => { + let body = '' + request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') }) + request.on('end', () => { + requests.push(JSON.parse(body)) + headers.push(request.headers) + const behavior = script.shift() + if (!behavior) { + response.writeHead(500).end('mock script exhausted') + return + } + if (behavior.kind === 'http-error') { + response.writeHead(behavior.status, { + 'content-type': behavior.contentType ?? 'application/json', + ...behavior.headers, + }) + response.end(behavior.body) + return + } + response.writeHead(200, { 'content-type': 'text/event-stream' }) + const write = (index: number): void => { + if (index >= behavior.events.length) { + if (behavior.kind === 'sse') response.end() + else response.destroy() // close-early: drop the socket mid-stream + return + } + response.write(`data: ${behavior.events[index]}\n\n`) + setTimeout(() => { write(index + 1) }, behavior.kind === 'sse' ? behavior.delayMs ?? 0 : 5) + } + write(0) + }) + }) + servers.push(server) + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const address = server.address() + if (address === null || typeof address === 'string') throw new Error('no port') + return { + url: `http://127.0.0.1:${address.port}`, + requests, + headers, + script, + close: () => new Promise(resolve => server.close(() => { resolve() })), + } +} diff --git a/packages/llm/llm-deepseek/tsconfig.json b/packages/llm/llm-deepseek/tsconfig.json index 45c2af21a5..ee8a81e73b 100644 --- a/packages/llm/llm-deepseek/tsconfig.json +++ b/packages/llm/llm-deepseek/tsconfig.json @@ -20,6 +20,12 @@ { "path": "../../llm/llm" }, + { + "path": "../../credentials/credentials" + }, + { + "path": "../../settings/settings" + }, { "path": "../../support/invariants" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 52e7b070b7..c7f3ae3f7d 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2949,12 +2949,18 @@ importers: specifier: ^3.18.0 version: 3.18.0 devDependencies: + '@deepseek-ai/dsh-credentials': + specifier: workspace:^ + version: link:../../credentials/credentials '@deepseek-ai/dsh-invariants': specifier: workspace:^ version: link:../../support/invariants '@deepseek-ai/dsh-llm': specifier: workspace:^ version: link:../llm + '@deepseek-ai/dsh-settings': + specifier: workspace:^ + version: link:../../settings/settings '@deepseek-ai/dsh-timeout': specifier: workspace:^ version: link:../../util/timeout