Files
deepseek-harness/packages/llm/llm-pi-ai/src/adapter.ts
T
creatixchu 0d1250f743 fix: address ds-review-bot v7 findings on the merged image-input head
- gate model selection on steering-placement image carriers from enqueue
  until their steering/message event publishes; release the gate when an
  admission ends idle without publication (both behaviorally asserted)
- reject session.updateQueue edits carrying non-text blocks at the RPC
  boundary (queue edits cannot bypass image admission)
- extend the durable-directory walk past a first-created DSH_HOME to the
  deepest pre-existing ancestor
- strip Windows-style separators from attachment display names on POSIX
- verify attachment reads with a header-only probe (digest already proves
  the bytes decoded fully at admission); document the read path
- make SessionInputShell.addImages refusal observable and keep workspace
  transfers/composer intake from leaking refused drafts
- own ONE recursive image walk (dsh-llm contentHasImage) across apiproxy,
  pi-ai, compact-basic, and the DeepSeek text-only assertion
- drop the redundant canonical-base64 regex and the no-op role read
- move AttachmentId/AttachmentError out of types.ts (brand.ts/error.ts);
  document why AttachmentError does not extend HarnessError
- document the hard attachments inject in both consumer READMEs
2026-07-30 14:34:08 +08:00

250 lines
10 KiB
TypeScript

/**
* Generic pi-ai-backed implementation of the Harness LLM seam.
*
* @module dsh-llm-pi-ai/adapter
*/
import { streamSimple } from '@earendil-works/pi-ai/compat'
import { getBuiltinModels } from '@earendil-works/pi-ai/providers/all'
import type { BuiltinProvider } from '@earendil-works/pi-ai/providers/all'
import { getSupportedThinkingLevels } from '@earendil-works/pi-ai'
import type {
Api,
Model,
ModelThinkingLevel,
SimpleStreamOptions,
ThinkingLevel,
} from '@earendil-works/pi-ai'
import type { AttachmentStore } from '@deepseek-ai/dsh-attachment'
import {
attributionHeaders,
LlmAdapter,
LlmError,
ReasoningEffortId,
} from '@deepseek-ai/dsh-llm'
import type {
GenerateOptions,
LlmModelInfo,
LlmResolvedModelInfo,
ReasoningEffortId as ReasoningEffortIdType,
ResolvedRetryPolicy,
StreamChunk,
} from '@deepseek-ai/dsh-llm'
import { idleWatchdog, timeoutOf } from '@deepseek-ai/dsh-timeout'
import { resolveProfiles } from './config.ts'
import type { PiAiProviderProfile, ResolvedPiAiProviderProfile } from './config.ts'
import { contentHasImage } from '@deepseek-ai/dsh-llm'
import { toPiContext } from './context.ts'
import { toStreamChunks } from './stream.ts'
/** Constructor options for {@link PiAiAdapter}. */
export interface PiAiAdapterOptions {
/** Validated provider profiles this adapter instance owns. */
profiles: readonly PiAiProviderProfile[]
/** Resolve durable image storage at request time so plugin load order does not become capability state. */
resolveAttachments?: () => AttachmentStore | undefined
}
/**
* Resolve a catalog model dynamically and apply only the configured endpoint
* override, preserving the catalog's API/capability/compatibility metadata.
*/
function resolvePiModel(
profile: Omit<PiAiProviderProfile, 'retryPolicy'>,
modelId: string,
): Model<Api> {
const model = getBuiltinModels(profile.provider as BuiltinProvider).find(candidate => candidate.id === modelId) as Model<Api> | undefined
if (model === undefined) {
throw new LlmError(`pi-ai provider "${profile.provider}" has no catalog model "${modelId}"`, 'UNKNOWN_MODEL')
}
return profile.baseURL === undefined ? model : { ...model, baseUrl: profile.baseURL }
}
/** Copy profile stream knobs into pi-ai's common option vocabulary. */
function profileOptions(
profile: Omit<PiAiProviderProfile, 'retryPolicy'>,
reasoning: ModelThinkingLevel | undefined,
): SimpleStreamOptions {
const enabledReasoning: ThinkingLevel | undefined = reasoning === 'off' ? undefined : reasoning
return {
...profile.apiKey === undefined ? {} : { apiKey: profile.apiKey },
...enabledReasoning === undefined ? {} : { reasoning: enabledReasoning },
...profile.thinkingBudgets === undefined ? {} : { thinkingBudgets: profile.thinkingBudgets },
...profile.cacheRetention === undefined ? {} : { cacheRetention: profile.cacheRetention },
...profile.transport === undefined ? {} : { transport: profile.transport },
...profile.timeoutMs === undefined ? {} : { timeoutMs: profile.timeoutMs },
...profile.websocketConnectTimeoutMs === undefined ? {} : { websocketConnectTimeoutMs: profile.websocketConnectTimeoutMs },
// The agent recovery layer owns visible attempts; one adapter call is one SDK attempt.
maxRetries: 0,
}
}
/** Validate an explicit Harness/profile effort without invoking pi-ai's clamp. */
function resolveReasoningLevel(
model: Model<Api>,
effort: ReasoningEffortIdType | ModelThinkingLevel | undefined,
): ModelThinkingLevel | undefined {
if (effort === undefined) return undefined
const supported = getSupportedThinkingLevels(model)
if (supported.some(level => level === effort)) return effort as ModelThinkingLevel
throw new LlmError(
`pi-ai provider "${model.provider}" model "${model.id}" does not support reasoning effort "${effort}"`,
'UNSUPPORTED_REASONING_EFFORT',
)
}
/** Merge deployment headers while removing case-insensitive attribution collisions. */
function requestHeaders(headers: Readonly<Record<string, string>> | undefined): Record<string, string> {
const attribution = attributionHeaders()
const reserved = new Set(Object.keys(attribution).map(name => name.toLowerCase()))
return {
...Object.fromEntries(Object.entries(headers ?? {}).filter(([name]) => !reserved.has(name.toLowerCase()))),
...attribution,
}
}
/**
* pi-ai-backed multi-provider adapter. Model descriptors are resolved for each
* request, so models need not be registered during the Cordis lifecycle.
*/
export class PiAiAdapter extends LlmAdapter {
private readonly profiles: ReadonlyMap<string, ResolvedPiAiProviderProfile>
private readonly resolveAttachments: () => AttachmentStore | undefined
constructor(options: PiAiAdapterOptions) {
super()
this.profiles = new Map(resolveProfiles(options.profiles).map(profile => [profile.provider, profile]))
this.resolveAttachments = options.resolveAttachments ?? (() => undefined)
}
override providerRetryPolicy(provider: string): ResolvedRetryPolicy | undefined {
return this.profiles.get(provider)?.retryPolicy
}
override listModels(provider: string): Promise<readonly LlmModelInfo[]> {
const profile = this.profiles.get(provider)
if (profile === undefined) {
return Promise.reject(new LlmError(`pi-ai adapter does not own provider "${provider}"`, 'NO_ADAPTER'))
}
return Promise.resolve(getBuiltinModels(profile.provider as BuiltinProvider).map(model => ({
provider,
id: model.id,
name: model.name,
inputModalities: [...model.input],
})))
}
override resolveModel(
provider: string,
model: string,
_signal?: AbortSignal,
): Promise<LlmResolvedModelInfo> {
const profile = this.profiles.get(provider)
if (profile === undefined) {
return Promise.reject(new LlmError(
`pi-ai adapter does not own provider "${provider}"`,
'NO_ADAPTER',
))
}
return Promise.resolve().then(() => {
const resolvedModel = resolvePiModel(profile, model)
const levels = getSupportedThinkingLevels(resolvedModel)
const defaultLevel = resolveReasoningLevel(resolvedModel, profile.reasoning)
return {
provider,
id: model,
name: resolvedModel.name,
inputModalities: [...resolvedModel.input],
context: { contextWindow: resolvedModel.contextWindow },
reasoning: {
efforts: levels.map(level => ({
id: ReasoningEffortId(level),
name: `${level.charAt(0).toUpperCase()}${level.slice(1)}`,
})),
...defaultLevel === undefined
? {}
: { defaultEffort: ReasoningEffortId(defaultLevel) },
},
}
})
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
if (options.stop !== undefined) {
throw new LlmError('llm-pi-ai does not support GenerateOptions.stop', 'UNSUPPORTED_OPTION')
}
const profile = this.profiles.get(options.provider)
if (profile === undefined) {
throw new LlmError(`pi-ai adapter does not own provider "${options.provider}"`, 'NO_ADAPTER')
}
const model = resolvePiModel(profile, options.model)
const reasoning = resolveReasoningLevel(
model,
options.reasoningEffort ?? profile.reasoning,
)
const consumer = new AbortController()
const upstream = options.signal === undefined
? consumer.signal
: AbortSignal.any([options.signal, consumer.signal])
const streamIdleTimeoutMs = profile.streamIdleTimeoutMs
using watchdog = idleWatchdog(upstream, streamIdleTimeoutMs, 'LLM_STREAM_IDLE_TIMEOUT')
try {
const containsImage = options.messages.some(message => contentHasImage(message.content))
if (containsImage && !model.input.includes('image')) {
throw new LlmError(`pi-ai model "${model.id}" does not support image input`, 'UNSUPPORTED_CONTENT')
}
const attachments = containsImage ? this.resolveAttachments() : undefined
if (containsImage && attachments === undefined) {
throw new LlmError('pi-ai image input requires the durable attachment service', 'UNSUPPORTED_CONTENT')
}
const context = attachments === undefined
? toPiContext(options)
: await toPiContext(options, attachments)
const events = streamSimple(model, context, {
...profileOptions(profile, reasoning),
...options.temperature === undefined ? {} : { temperature: options.temperature },
...options.maxTokens === undefined ? {} : { maxTokens: options.maxTokens },
...options.sessionId === undefined ? {} : { sessionId: String(options.sessionId) },
signal: watchdog.signal,
// Profile headers are deployment-owned; attribution names are
// Harness-owned and therefore win collisions.
headers: requestHeaders(profile.headers),
})
const iterator = toStreamChunks(events, model.contextWindow)[Symbol.asyncIterator]()
let exhausted = false
try {
while (true) {
const result = await watchdog.next(iterator)
const timeout = timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT')
if (timeout !== undefined) throw timeout
if (result.done) {
exhausted = true
return
}
yield result.value
}
} finally {
if (!exhausted) {
consumer.abort('pi-ai stream consumer stopped')
try {
await iterator.return(undefined)
} catch (_abortedSdkTeardown) {
// The stable signal already owns SDK termination; return-time abort cannot add an outcome.
}
}
}
} catch (error: unknown) {
if (timeoutOf(watchdog.signal, 'LLM_STREAM_IDLE_TIMEOUT') !== undefined) {
throw new LlmError(`pi-ai stream idle timeout after ${streamIdleTimeoutMs}ms`, 'TIMEOUT', { cause: error })
}
if (options.signal?.aborted) {
throw new LlmError('pi-ai request aborted by caller', 'ABORTED', { cause: error })
}
throw error
} finally {
consumer.abort('pi-ai stream consumer stopped')
}
}
}