Merge master into codex/simp-session-log-representation

This commit is contained in:
Tianyi Cui
2026-07-17 22:33:42 +08:00
310 changed files with 4718 additions and 1903 deletions

View File

@@ -6,6 +6,6 @@ The LLM seam and its provider adapters. The interface package (`llm`) owns the a
|---|---|---|
| `llm/` | Abstract LLM service + content-block vocabulary + chunk assembler | `ctx.llm` |
| `llm-deepseek/` | DeepSeek API adapter (hand-rolled fetch/SSE) | (registers on `ctx.llm`) |
| `llm-pi-ai/` | DeepSeek adapter via `@earendil-works/pi-ai` (design twin) | (registers on `ctx.llm`) |
| `llm-pi-ai/` | Multi-provider adapter via `@earendil-works/pi-ai` | (registers on `ctx.llm`) |
The interface lives at `llm/llm/`; adapters are flat siblings under the group. A new provider adapter joins here and registers on `ctx.llm` without touching the interface. See [twin LLM adapters](../../docs/rfc/implemented/architecture/2026-06-13-twin-llm-adapters.md) for why two adapters exist.
The interface lives at `llm/llm/`; adapters are flat siblings under the group. Requests route by `provider`, while `model` is passed through to the selected adapter. A new provider adapter joins here and registers one or more provider routes on `ctx.llm` without touching the interface. See [twin LLM adapters](../../docs/rfc/implemented/architecture/2026-06-13-twin-llm-adapters.md) for the contract-validation origin of the two shipping implementations.

View File

@@ -2,7 +2,7 @@
DeepSeek chat-completions adapter for the harness LLM seam: hand-rolled `fetch` + SSE translation from the official wire format (source of truth: the API docs — guides/thinking_mode, guides/tool_calls, api/create-chat-completion) into the `StreamChunk` protocol.
A second, independent implementation of the same seam exists in `@deepseek-ai/dsh-llm-pi-ai` (library-backed). Same Config shape — pick one per context (registering both for the same model names throws by design).
A second, library-backed implementation of the same seam exists in `@deepseek-ai/dsh-llm-pi-ai`. This package always owns the `deepseek` provider route; mounting a pi-ai profile with `provider: deepseek` in the same context throws `LlmError('DUPLICATE_ADAPTER')` by design.
## Config
@@ -12,12 +12,16 @@ A second, independent implementation of the same seam exists in `@deepseek-ai/ds
config:
apiKey: !!js process.env.DEEPSEEK_API_KEY # or rely on the env fallback
baseURL: !!js process.env.DEEPSEEK_BASE_URL # default: https://api.deepseek.com
models: [deepseek-v4-flash, deepseek-v4-pro] # one adapter, registered for each name
thinking: enabled # optional; provider default is enabled
reasoningEffort: high # optional; high | max — omitted ⇒ not sent
models: # optional; defaults to V4 Flash and V4 Pro
- id: deepseek-v4-flash
name: DeepSeek V4 Flash
- id: private-reasoner
description: Company-hosted reasoning model
```
`models` lists every model name this one adapter instance serves: the adapter registers itself for each (the harness model name IS the wire `model` string), so a `generate`/`stream` call routes to it whenever `options.model` is any of them. Registering a second adapter for a name already taken throws `LlmError('DUPLICATE_ADAPTER')` (the LLM service enforces one adapter per model, all-or-nothing).
The plugin registers the single provider route `deepseek`. A request selects it with `provider: deepseek`; its `model` is passed through as the wire `model` string, so changing DeepSeek models does not require lifecycle-time registration. Omitting `models` advertises `deepseek-v4-flash` and `deepseek-v4-pro`; an explicit list replaces those defaults, while `models: []` advertises none. Catalog entries are exposed through `ctx.llm.listModels('deepseek')` for clients such as ACP editors, but remain advisory: unlisted model ids still pass through unchanged. An omitted entry name defaults to its id. Registering another adapter for `deepseek` throws `LlmError('DUPLICATE_ADAPTER')`.
`reasoningEffort` is **omitted by default** — when unset, the `reasoning_effort` wire field is not sent and the server applies its own default for the model. The only accepted values are `high` and `max` (DeepSeek's official effort levels). It is meaningful only with thinking enabled (the provider default).

View File

@@ -6,13 +6,23 @@
*/
import { attributionHeaders, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } 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'
import { parseSse } from './sse.ts'
import { translate } from './translate.ts'
import type { WireError } from './types.ts'
/** One optional model entry advertised by the hand-written adapter. */
export interface DeepSeekCatalogModel {
/** Wire model id accepted by the configured endpoint. */
id: string
/** Selector label; defaults to {@link id}. */
name?: string
/** Optional selector detail for deployments with similar model variants. */
description?: string
}
/** 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. */
@@ -21,6 +31,8 @@ export interface DeepSeekAdapterOptions {
baseURL: string
/** Request defaults applied to every call (thinking mode, effort). */
defaults?: RequestDefaults
/** Advisory models exposed to discovery consumers; requests remain unrestricted. */
models?: readonly DeepSeekCatalogModel[]
}
/**
@@ -49,6 +61,19 @@ export class DeepSeekAdapter extends LlmAdapter {
super()
}
override providerInfo(provider: string): LlmProviderInfo {
return { id: provider, name: 'DeepSeek' }
}
override listModels(provider: string): Promise<readonly LlmModelInfo[]> {
return Promise.resolve((this.options.models ?? []).map(model => ({
provider,
id: model.id,
name: model.name ?? model.id,
...model.description === undefined ? {} : { description: model.description },
})))
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
const body = serializeRequest(options, this.options.defaults ?? {})

View File

@@ -1,5 +1,5 @@
/**
* Register a {@link DeepSeekAdapter} for configured model names on `ctx.llm`. Configuration uses
* 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.
* @module @deepseek-ai/dsh-llm-deepseek
@@ -9,9 +9,10 @@ import type { Context } from 'cordis'
import z from 'schemastery'
import type {} from '@deepseek-ai/dsh-llm'
import { DeepSeekAdapter } from './adapter.ts'
import type { DeepSeekCatalogModel } from './adapter.ts'
export { DeepSeekAdapter, httpErrorCode } from './adapter.ts'
export type { DeepSeekAdapterOptions } from './adapter.ts'
export type { DeepSeekAdapterOptions, DeepSeekCatalogModel } from './adapter.ts'
export { serializeMessages, serializeRequest } from './serialize.ts'
export type { RequestDefaults } from './serialize.ts'
export { DONE, parseSse } from './sse.ts'
@@ -21,6 +22,11 @@ export type * from './types.ts'
export const name = 'llm-deepseek'
export const inject = ['llm']
const DEFAULT_MODELS: DeepSeekCatalogModel[] = [
{ id: 'deepseek-v4-flash' },
{ id: 'deepseek-v4-pro' },
]
/**
* Plugin config, validated by the same-named schemastery schema. Every field
* is optional in yml: credentials/endpoint fall back to the environment (a
@@ -32,40 +38,62 @@ export interface Config {
apiKey?: string
/** Endpoint base; falls back to $DEEPSEEK_BASE_URL, then the public API. */
baseURL?: string
/** Model names to register (sent verbatim on the wire). */
models?: string[]
/** Thinking-mode default for every request (provider default: enabled). */
thinking?: 'enabled' | 'disabled'
/** Thinking effort (only meaningful with thinking enabled). */
reasoningEffort?: 'high' | 'max'
/** Advisory models shown by discovery consumers; defaults to V4 Flash and V4 Pro. */
models?: DeepSeekCatalogModel[]
}
const catalogModel: z<DeepSeekCatalogModel> = z.object({
id: z.string().required(),
name: z.string(),
description: z.string(),
})
export const Config: z<Config> = z.object({
apiKey: z.string(),
baseURL: z.string(),
models: z.array(z.string()).default(['deepseek-v4-flash', 'deepseek-v4-pro']),
thinking: z.union(['enabled', 'disabled']),
reasoningEffort: z.union(['high', 'max']),
models: z.array(catalogModel).default(DEFAULT_MODELS),
})
/** Public API default; the internal endpoint comes from $DEEPSEEK_BASE_URL. */
export const PUBLIC_BASE_URL = 'https://api.deepseek.com'
/** Resolve, validate, and detach the advisory model catalog. */
function resolveModels(models: readonly DeepSeekCatalogModel[] | undefined): DeepSeekCatalogModel[] {
const seen = new Set<string>()
return (models ?? DEFAULT_MODELS).map((model) => {
if (model.id.length === 0) throw new Error('llm-deepseek: catalog model ids must be non-empty')
if (model.name !== undefined && model.name.length === 0) {
throw new Error(`llm-deepseek: catalog model "${model.id}" has an empty name`)
}
if (seen.has(model.id)) throw new Error(`llm-deepseek: duplicate catalog model "${model.id}"`)
seen.add(model.id)
return {
id: model.id,
...model.name === undefined ? {} : { name: model.name },
...model.description === undefined ? {} : { description: model.description },
}
})
}
export function apply(ctx: Context, config: Config): void {
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)')
}
const baseURL = config.baseURL ?? process.env.DEEPSEEK_BASE_URL ?? PUBLIC_BASE_URL
// schemastery's .default() guarantees models is set after validation.
const models = config.models as string[]
ctx.llm.registerAdapter(models, new DeepSeekAdapter({
ctx.llm.registerAdapter(['deepseek'], new DeepSeekAdapter({
apiKey,
baseURL,
defaults: {
thinking: config.thinking,
reasoningEffort: config.reasoningEffort,
},
models: resolveModels(config.models),
}))
}

View File

@@ -16,11 +16,11 @@ const FLASH = 'deepseek-v4-flash'
const PRO = 'deepseek-v4-pro'
const contexts: Context[] = []
async function harness(model: string, config: Partial<Config> = {}) {
async function harness(_model: string, config: Partial<Config> = {}) {
const ctx = new Context()
contexts.push(ctx)
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, { models: [model], ...config })
await ctx.plugin(LlmDeepSeek, config)
return ctx
}
@@ -134,6 +134,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', ()
const ctx = await harness(FLASH, { thinking: 'disabled' })
const kinds: string[] = []
for await (const chunk of ctx.llm.stream({
provider: 'deepseek',
model: FLASH,
messages: ask('Count from 1 to 5, digits only.'),
maxTokens: 50,

View File

@@ -86,7 +86,7 @@ const textEvents = [
async function harness(baseURL: string, config: object = {}) {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, { apiKey: 'test-key', baseURL, models: ['deepseek-v4-flash'], ...config })
await ctx.plugin(LlmDeepSeek, { apiKey: 'test-key', baseURL, ...config })
return ctx
}
@@ -123,6 +123,7 @@ describe('DeepSeekAdapter against a mock server', () => {
const kinds: string[] = []
for await (const chunk of ctx.llm.stream({
provider: 'deepseek',
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
})) {
@@ -198,7 +199,7 @@ describe('DeepSeekAdapter against a mock server', () => {
)
try {
const iterate = async (): Promise<void> => {
for await (const _chunk of adapter.stream({ model: 'm', messages: [] })) { /* drain */ }
for await (const _chunk of adapter.stream({ provider: 'deepseek', model: 'm', messages: [] })) { /* drain */ }
}
await expect(iterate()).rejects.toThrow(/no response body/)
} finally {
@@ -224,6 +225,7 @@ describe('DeepSeekAdapter against a mock server', () => {
const pending = (async () => {
const chunks = []
for await (const chunk of ctx.llm.stream({
provider: 'deepseek',
model: 'deepseek-v4-flash',
messages: [],
signal: controller.signal,
@@ -239,25 +241,81 @@ describe('DeepSeekAdapter against a mock server', () => {
})
describe('plugin registration and config', () => {
it('registers the configured models and unregisters on dispose (HMR safety)', async () => {
it('registers the deepseek provider and unregisters on dispose (HMR safety)', async () => {
const server = await mockServer([])
const ctx = new Context()
await ctx.plugin(LlmService)
const fiber = await ctx.plugin(LlmDeepSeek, {
apiKey: 'k',
baseURL: server.url,
models: ['deepseek-v4-flash', 'deepseek-v4-pro'],
})
expect(ctx.llm.models().sort()).toEqual(['deepseek-v4-flash', 'deepseek-v4-pro'])
expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }])
await fiber.dispose()
expect(ctx.llm.models()).toEqual([])
expect(ctx.llm.listProviders()).toEqual([])
})
it('defaults the model list', async () => {
it('owns the deepseek provider and advertises the default models', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, { apiKey: 'k', baseURL: 'http://127.0.0.1:1' })
expect(ctx.llm.models().sort()).toEqual(['deepseek-v4-flash', 'deepseek-v4-pro'])
expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }])
await expect(ctx.llm.listModels('deepseek')).resolves.toEqual([
{ provider: 'deepseek', id: 'deepseek-v4-flash', name: 'deepseek-v4-flash' },
{ provider: 'deepseek', id: 'deepseek-v4-pro', name: 'deepseek-v4-pro' },
])
})
it('uses the default model catalog when apply is called directly', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
LlmDeepSeek.apply(ctx, { apiKey: 'k', baseURL: 'http://127.0.0.1:1' })
await expect(ctx.llm.listModels('deepseek')).resolves.toEqual([
{ provider: 'deepseek', id: 'deepseek-v4-flash', name: 'deepseek-v4-flash' },
{ provider: 'deepseek', id: 'deepseek-v4-pro', name: 'deepseek-v4-pro' },
])
})
it('advertises configured models without restricting arbitrary request ids', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, {
apiKey: 'k',
baseURL: 'http://127.0.0.1:1',
models: [
{ id: 'private-fast' },
{ id: 'private-reasoner', name: 'Private Reasoner', description: 'Higher reasoning budget' },
],
})
await expect(ctx.llm.listModels('deepseek')).resolves.toEqual([
{ provider: 'deepseek', id: 'private-fast', name: 'private-fast' },
{ provider: 'deepseek', id: 'private-reasoner', name: 'Private Reasoner', description: 'Higher reasoning budget' },
])
})
it('allows an explicit empty model catalog', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, {
apiKey: 'k',
baseURL: 'http://127.0.0.1:1',
models: [],
})
await expect(ctx.llm.listModels('deepseek')).resolves.toEqual([])
})
it.each([
[[{ id: '' }], /ids must be non-empty/],
[[{ id: 'm', name: '' }], /empty name/],
[[{ id: 'm' }, { id: 'm' }], /duplicate catalog model/],
] as const)('rejects invalid advisory model config', async (models, message) => {
const ctx = new Context()
await ctx.plugin(LlmService)
await expect(ctx.plugin(LlmDeepSeek, {
apiKey: 'k',
baseURL: 'http://127.0.0.1:1',
models: [...models],
})).rejects.toThrow(message)
expect(ctx.llm.listProviders()).toEqual([])
})
it('falls back to DEEPSEEK_API_KEY and DEEPSEEK_BASE_URL env vars', async () => {
@@ -266,7 +324,7 @@ describe('plugin registration and config', () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, {})
expect(ctx.llm.models().length).toBeGreaterThan(0)
expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }])
})
it('throws a clear error when no API key is available', async () => {
@@ -275,7 +333,7 @@ describe('plugin registration and config', () => {
await ctx.plugin(LlmService)
await expect(ctx.plugin(LlmDeepSeek, {}))
.rejects.toThrow(/an API key is required/)
expect(ctx.llm.models()).toEqual([])
expect(ctx.llm.listProviders()).toEqual([])
})
it('prefers explicit config over env for key and base URL', async () => {
@@ -292,7 +350,7 @@ describe('plugin registration and config', () => {
vi.stubEnv('DEEPSEEK_BASE_URL', server.url)
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, { apiKey: 'k', models: ['deepseek-v4-flash'] })
await ctx.plugin(LlmDeepSeek, { apiKey: 'k' })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(server.requests).toHaveLength(1)
})
@@ -304,11 +362,12 @@ describe('plugin registration and config', () => {
await ctx.plugin(LlmService)
// Registration succeeds; no call is made (would hit api.deepseek.com).
await ctx.plugin(LlmDeepSeek, {})
expect(ctx.llm.models().length).toBeGreaterThan(0)
expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek', name: 'DeepSeek' }])
})
it('adapter is constructible directly for embedding', () => {
it('adapter is constructible directly for embedding', async () => {
const adapter = new DeepSeekAdapter({ apiKey: 'k', baseURL: 'http://127.0.0.1:1' })
expect(adapter).toBeInstanceOf(DeepSeekAdapter)
await expect(adapter.listModels('deepseek')).resolves.toEqual([])
})
})

View File

@@ -15,11 +15,19 @@ export interface AssembledResult {
finish: FinishReason
}
export async function assemble(ctx: Context, options: GenerateOptions): Promise<AssembledResult> {
export async function assemble(ctx: Context, options: Omit<GenerateOptions, 'provider'> & { provider?: string }): Promise<AssembledResult> {
const assembler = new BlockAssembler()
for await (const chunk of ctx.llm.stream(options)) assembler.push(chunk)
const request = { provider: 'deepseek', ...options }
for await (const chunk of ctx.llm.stream(request)) assembler.push(chunk)
return {
message: assembler.message(),
message: {
...assembler.message(),
provenance: {
provider: request.provider,
model: request.model,
...assembler.replayState === undefined ? {} : { replayState: assembler.replayState },
},
},
...assembler.usage !== undefined ? { usage: assembler.usage } : {},
finish: assembler.finish,
}

View File

@@ -4,7 +4,7 @@ import type { ContentBlock, GenerateOptions, Message } from '@deepseek-ai/dsh-ll
import { serializeMessages, serializeRequest } from '@deepseek-ai/dsh-llm-deepseek'
function request(overrides: Partial<GenerateOptions> = {}): GenerateOptions {
return { model: 'deepseek-v4-flash', messages: [], ...overrides }
return { provider: 'deepseek', model: 'deepseek-v4-flash', messages: [], ...overrides }
}
describe('serializeMessages', () => {

View File

@@ -1,60 +1,79 @@
# @deepseek-ai/dsh-llm-pi-ai
DeepSeek adapter for the harness LLM seam backed by [`@earendil-works/pi-ai`](https://www.npmjs.com/package/@earendil-works/pi-ai) (the LLM library behind the pi agent).
## Why a second adapter exists
`@deepseek-ai/dsh-llm-deepseek` already talks to the same endpoint. This package is its **design-verification twin**: same models, same wire protocol, completely different internals — a unified LLM library with its own event vocabulary versus hand-rolled fetch/SSE. Anything the harness `StreamChunk` protocol cannot express for BOTH implementations is a core-vocabulary bug. The differences it exercised on purpose:
- pi-ai hands tool-call `arguments` around as **parsed objects**; the harness keeps raw JSON strings. The adapter patches replay payloads back to the original raw strings before sending them, and re-stringifies parsed output tool calls at `block-end`.
- pi-ai reports failures as **in-stream error events** (it never throws mid-stream); these map to `finish {kind:'error'|'aborted'}` chunks — the protocol's other sanctioned error path besides throwing (which llm-deepseek uses).
- pi-ai folds reasoning tokens into `usage.output`; there is no separate reasoning count to map.
- pi-ai's options omit some DeepSeek/OpenAI-compatible details; the adapter uses its `onPayload` hook to preserve the harness contract (`stop`, scrubbing pi-ai's own per-tool `strict` default — the hand-rolled twin sends no such field — omitted reasoning effort, raw replayed tool arguments).
Generic multi-provider adapter for the harness LLM seam backed by [`@earendil-works/pi-ai`](https://www.npmjs.com/package/@earendil-works/pi-ai). One plugin instance owns an explicit list of provider profiles; every request selects a profile with `GenerateOptions.provider` and resolves `GenerateOptions.model` dynamically from pi-ai's installed catalog.
## Config
Same shape as llm-deepseek (one-line swap in cordis.yml), with pi-ai's thinking-level vocabulary:
Configure credentials and deployment-specific transport settings per provider. Omitting `apiKey` delegates authentication to pi-ai's provider-native ambient discovery. `baseURL` overrides only the endpoint of the selected catalog model, preserving its API family and compatibility metadata, so private proxies such as `https://proxy.example.com:8443` remain supported.
```yaml
- id: llm
name: '@deepseek-ai/dsh-llm-pi-ai'
config:
apiKey: !!js process.env.DEEPSEEK_API_KEY
baseURL: !!js process.env.DEEPSEEK_BASE_URL
models: [deepseek-v4-flash, deepseek-v4-pro]
reasoning: high # off | high | xhigh (xhigh → wire 'max')
providers:
- provider: openai
apiKey: !!js process.env.OPENAI_API_KEY
baseURL: https://proxy.example.com:8443
reasoning: high
- provider: anthropic
apiKey: !!js process.env.ANTHROPIC_API_KEY
maxRetries: 2
- provider: openrouter
apiKey: !!js process.env.OPENROUTER_API_KEY
headers:
X-Deployment: production
```
Each provider name must exist in pi-ai's installed catalog and may appear only once in this plugin instance. Registration with `ctx.llm` is atomic: a collision with any provider route already owned by another adapter fails plugin loading without registering the remaining routes. Model ids are not lifecycle config; an unknown model fails before any provider request with `LlmError('UNKNOWN_MODEL')`.
The adapter exposes each configured provider's installed pi-ai models through `ctx.llm.listModels(provider)`. This is provider-neutral selector metadata derived from `getModels(provider)`; request-time resolution still performs the authoritative catalog lookup, so discovery does not create a second model registry.
Supported profile fields are `provider`, `apiKey`, `baseURL`, `headers`, `reasoning`, `thinkingBudgets`, `cacheRetention`, `transport`, `timeoutMs`, `websocketConnectTimeoutMs`, `maxRetries`, and `maxRetryDelayMs`. They map to pi-ai's common stream options. Harness app attribution wins a conflicting configured header name.
## Provider/model routing and replay
The selected pi-ai catalog descriptor supplies the protocol implementation. This includes native API differences such as OpenAI models whose descriptor uses the Responses API rather than Chat Completions; the harness adapter does not hardcode endpoint selection by model name.
Successful assistant responses store a versioned, lossless-JSON replay state beside their durable provider/model provenance. At request time, `LlmService` passes replay state only when the historical provider route and target provider route are currently owned by this same `PiAiAdapter` instance. The adapter validates the state and restores pi-ai response ids and provider signatures even when the target provider or model changes; pi-ai then decides which metadata its target API can reuse. History without replay state is translated as foreign provider-neutral content and never impersonates a native pi-ai response.
If a listener rewrites assembled assistant content, the loop drops replay state before logging the message because its provider metadata no longer describes the content. Invalid versions, malformed metadata, provenance provider/model mismatches, and content/block mismatches fail explicitly with `LlmError('INVALID_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 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.
## App attribution
Every request carries the shared attribution header from dsh-llm's `attributionHeaders()`, passed through pi-ai's `headers` stream option (pi-ai merges caller headers last, so it always reaches the wire - the unit suite asserts arrival on the mock server, same as llm-deepseek). OpenRouter-specific app attribution headers are intentionally not sent by this adapter contract; they are deferred to a future explicit OpenRouter adapter or mode. See [dsh-llm § App attribution](../llm/README.md#app-attribution-attributionts).
Every request carries the shared attribution header from dsh-llm's `attributionHeaders()`, merged through pi-ai's `headers` stream option. Provider-specific app-attribution headers are not synthesized. See [dsh-llm § App attribution](../llm/README.md#app-attribution-attributionts).
## Dependency weight
pi-ai declares the openai/anthropic/google/mistral/AWS SDKs as install-time dependencies. They are lazy-loaded — only the openai SDK actually loads for this adapter — but they do land in `node_modules`. Accepted for a package whose purpose is design verification.
pi-ai installs several provider SDKs and lazy-loads the one selected by the catalog model. The dependency weight is isolated to this opt-in adapter package.
## Testing
Unit suites run against a local `node:http` mock SSE server (pi-ai's openai SDK happily talks to any base URL). Real-API coverage in `tests/adapter.e2e.ts` (`pnpm run test:e2e`, key-gated): V4 Flash + V4 Pro across all exposed reasoning levels (off/high/xhigh), the thinking+tools round trip, and a cross-adapter structural-equivalence check against llm-deepseek.
Unit tests use pi-ai catalog models redirected to local mock servers and cover provider/profile routing, native API selection, endpoint overrides, attribution, conversion, replay-state validation, and cross-provider/model replay within one adapter instance. Real-API coverage remains key-gated under `pnpm run test:e2e`.
## Model Experience
### DeepSeek request through pi-ai
### Provider request through pi-ai
**What the model sees**: The selected model receives the same logical system prompt, history, tools, stop sequences, and raw replayed tool arguments as the hand-written adapter. This package adds no prompt prose and removes pi-ai's own per-tool `strict` default to preserve that contract.
**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.
**Token effect**: Provider tokenization governs exact input. Reasoning level changes generated and passback content; pi-ai reports reasoning inside output usage rather than as a separate count.
**Token effect**: Provider tokenization governs exact input. Conversion adds no model-visible text; replay metadata may let a native API reuse provider-side state.
### DeepSeek response
### Provider response
**What the model sees**: pi-ai events become harness reasoning, text, tool-call, usage, and finish chunks; parsed tool arguments are restored to raw JSON strings at the harness boundary.
**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.
**Token effect**: Generated content affects later inputs only after the loop records it; adapter conversion adds no model-visible text.
**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.
## Known Limitations and Deferred Work
- **`tool_choice` is not mapped** — same MVP contract as llm-deepseek.
- **In-history `system`-role messages fold into `user`-role wire messages** — pi-ai exposes a single `systemPrompt` slot, diverging from the hand-rolled twin's `role: 'system'` passthrough.
- **`LlmError.status` is never set** — pi-ai reports failures as in-stream events with no HTTP status, so error codes are regex-classified from the error text.
- **`buildModel` hardcodes descriptor metadata** — `contextWindow: 128000`, `maxTokens: 64000`, zero cost, identically for every registered model name; not configurable.
- **pi-ai's built-in retries are disabled (`maxRetries: 0`)** — failures surface immediately; retry policy belongs to `llm/stream` listeners.
- **Catalog membership is required** — custom model ids that are absent from the installed pi-ai catalog fail with `UNKNOWN_MODEL`, even when a provider profile supplies a custom endpoint.
- **`GenerateOptions.stop` is unsupported** — pi-ai's common stream options cannot guarantee stop-sequence behavior across providers, so the adapter rejects the field.
- **In-history `system` messages use pi-ai's common context conversion** — provider-specific placement follows pi-ai rather than a harness-owned wire override.
- **`LlmError.status` is unavailable for in-stream failures** — pi-ai error events do not expose a stable HTTP status across providers.

View File

@@ -1,147 +1,101 @@
/**
* Pi-ai-backed DeepSeek adapter and design twin of the hand-rolled adapter.
* Both implementations must fit the same provider-neutral stream vocabulary.
* Generic pi-ai-backed implementation of the Harness LLM seam.
*
* @module dsh-llm-pi-ai/adapter
*/
import { stream as piStream } from '@earendil-works/pi-ai'
import type { Model } from '@earendil-works/pi-ai'
import { attributionHeaders, LlmAdapter } from '@deepseek-ai/dsh-llm'
import { CallId } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import { toPiContext, toStreamChunks } from './convert.ts'
import {
getModels,
streamSimple,
} from '@earendil-works/pi-ai'
import type {
Api,
KnownProvider,
Model,
SimpleStreamOptions,
} from '@earendil-works/pi-ai'
import { attributionHeaders, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, LlmModelInfo, StreamChunk } from '@deepseek-ai/dsh-llm'
import type { PiAiProviderProfile } from './config.ts'
import { toPiContext } from './context.ts'
import { toStreamChunks } from './stream.ts'
/** Reasoning levels surfaced by this adapter (DeepSeek wire: high|max). */
export type PiAiReasoning = 'off' | 'high' | 'xhigh'
/** Constructor options for {@link PiAiAdapter}; the plugin's `apply` resolves them from Config + environment. */
/** Constructor options for {@link PiAiAdapter}. */
export interface PiAiAdapterOptions {
/** Bearer token pi-ai sends on every request. */
apiKey: string
/** Endpoint base; `/chat/completions` is appended. */
baseURL: string
/** Thinking level applied to every request ('off' disables thinking). */
reasoning?: PiAiReasoning | undefined
/** Validated provider profiles this adapter instance owns. */
profiles: readonly PiAiProviderProfile[]
}
/**
* Build the inline pi-ai model descriptor for one DeepSeek model name.
* @param modelId - harness model name; sent verbatim on the wire.
* @param options - adapter options; only `baseURL` is read here (key and reasoning apply per request, not per descriptor).
* @returns a descriptor with every DeepSeek compat flag explicit — pi-ai's URL-based auto-detection is never relied on.
* Resolve a catalog model dynamically and apply only the configured endpoint
* override, preserving the catalog's API/capability/compatibility metadata.
*/
export function buildModel(modelId: string, options: PiAiAdapterOptions): Model<'openai-completions'> {
function resolveModel(profile: PiAiProviderProfile, modelId: string): Model<Api> {
const model = getModels(profile.provider as KnownProvider).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: PiAiProviderProfile): SimpleStreamOptions {
return {
id: modelId,
name: modelId,
api: 'openai-completions',
provider: 'deepseek',
baseUrl: options.baseURL,
// Keep reasoning support enabled so `off` can send DeepSeek's explicit
// disabled marker rather than falling back to the provider's enabled default.
reasoning: true,
// DeepSeek's official effort levels: high|max (xhigh maps to max).
thinkingLevelMap: { minimal: null, low: null, medium: null, high: 'high', xhigh: 'max' },
input: ['text'],
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
contextWindow: 128_000,
maxTokens: 64_000,
compat: {
// Auto-detection only fires for *.deepseek.com base URLs; the internal
// endpoint (and test mocks) need these set explicitly.
thinkingFormat: 'deepseek',
requiresReasoningContentOnAssistantMessages: true,
supportsReasoningEffort: true,
// DeepSeek documents max_tokens (not OpenAI's max_completion_tokens).
maxTokensField: 'max_tokens',
},
...profile.apiKey === undefined ? {} : { apiKey: profile.apiKey },
...profile.reasoning === undefined ? {} : { reasoning: profile.reasoning },
...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 },
...profile.maxRetries === undefined ? {} : { maxRetries: profile.maxRetries },
...profile.maxRetryDelayMs === undefined ? {} : { maxRetryDelayMs: profile.maxRetryDelayMs },
}
}
type Payload = {
tools?: { function?: { strict?: unknown } }[]
messages?: {
role?: unknown
tool_calls?: { id?: unknown; function?: { arguments?: unknown } }[]
}[]
reasoning_effort?: unknown
stop?: unknown
}
function rawToolArguments(options: GenerateOptions): Map<CallId, string> {
const raw = new Map<CallId, string>()
for (const message of options.messages) {
if (message.role !== 'assistant') continue
for (const block of message.content) {
if (block.type === 'tool-call') raw.set(block.id, block.arguments)
}
/** 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,
}
return raw
}
function patchPayload(payload: unknown, options: GenerateOptions, reasoning: PiAiReasoning | undefined): unknown {
/* v8 ignore next -- pi-ai onPayload always receives an object; tolerate unusual future hooks defensively */
if (typeof payload !== 'object' || payload === null) return payload
const body = payload as Payload
if (reasoning === undefined) {
delete body.reasoning_effort
}
if (options.stop !== undefined) {
body.stop = options.stop
}
// pi-ai stamps its own `strict` default on every serialized tool; the
// harness tool contract has no strict field and the hand-rolled twin sends
// none, so scrub it for wire parity.
for (const tool of body.tools ?? []) {
/* v8 ignore next -- malformed pi-ai payload guard: real tool entries always carry function */
if (tool.function === undefined) continue
delete tool.function.strict
}
const rawById = rawToolArguments(options)
/* v8 ignore next -- defensive for non-chat payloads; OpenAI chat payloads always carry messages */
for (const message of body.messages ?? []) {
if (message.role !== 'assistant') continue
/* v8 ignore next -- assistant messages without tool_calls need no raw-argument patch */
for (const call of message.tool_calls ?? []) {
/* v8 ignore next -- malformed pi-ai payload guard: real tool calls always carry a string id */
if (typeof call.id !== 'string') continue
const raw = rawById.get(CallId(call.id))
/* v8 ignore next -- pi-ai always emits a function object for assistant tool_calls; guard malformed payloads defensively */
if (raw !== undefined && call.function !== undefined) call.function.arguments = raw
}
}
return body
}
/**
* pi-ai-backed adapter. One instance serves every registered model name.
*
* Implementation notes:
* - `onPayload` patches provider payload details pi-ai cannot express directly:
* stop sequences, scrubbing pi-ai's own per-tool `strict` default (the
* hand-rolled twin sends no such field), omitted reasoning effort, and raw
* replayed tool-call arguments.
* - pi-ai reports request failures as in-stream error events; convert.ts
* maps them to `finish {kind:'error'|'aborted'}` chunks rather than
* throwing — both are sanctioned StreamChunk error paths.
* 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 {
constructor(private readonly options: PiAiAdapterOptions) {
private readonly profiles: ReadonlyMap<string, PiAiProviderProfile>
constructor(options: PiAiAdapterOptions) {
super()
this.profiles = new Map(options.profiles.map(profile => [profile.provider, profile]))
}
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(getModels(profile.provider as KnownProvider).map(model => ({
provider,
id: model.id,
name: model.name,
})))
}
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
const model = buildModel(options.model, this.options)
// Undefined config means "provider default" (DeepSeek: thinking ENABLED),
// matching llm-deepseek's omission semantics. pi-ai derives the wire
// thinking toggle from whether reasoningEffort is passed, so undefined maps
// internally to 'high' to get `thinking: enabled`; patchPayload then removes
// `reasoning_effort` so the provider chooses its default effort.
const reasoning = this.options.reasoning ?? 'high'
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 = resolveModel(profile, options.model)
// Pi-ai has no iterator-return cancellation hook. Chain an internal signal
// and abort it when this generator exits so early consumers stop the HTTP stream.
@@ -151,19 +105,16 @@ export class PiAiAdapter extends LlmAdapter {
else options.signal?.addEventListener('abort', onCallerAbort, { once: true })
try {
const events = piStream(model, toPiContext(options), {
apiKey: this.options.apiKey,
// pi-ai merges caller headers last over its provider defaults, so the
// harness attribution always reaches the wire.
headers: attributionHeaders(),
...options.temperature !== undefined ? { temperature: options.temperature } : {},
...options.maxTokens !== undefined ? { maxTokens: options.maxTokens } : {},
const events = streamSimple(model, toPiContext(options), {
...profileOptions(profile),
...options.temperature === undefined ? {} : { temperature: options.temperature },
...options.maxTokens === undefined ? {} : { maxTokens: options.maxTokens },
...options.sessionId === undefined ? {} : { sessionId: String(options.sessionId) },
signal: controller.signal,
...reasoning !== 'off' ? { reasoningEffort: reasoning } : {},
onPayload: payload => patchPayload(payload, options, this.options.reasoning),
maxRetries: 0,
// Profile headers are deployment-owned; attribution names are
// Harness-owned and therefore win collisions.
headers: requestHeaders(profile.headers),
})
yield* toStreamChunks(events)
} finally {
options.signal?.removeEventListener('abort', onCallerAbort)

View File

@@ -0,0 +1,99 @@
/**
* Configuration schema and provider-profile validation for the pi-ai adapter.
*
* @module dsh-llm-pi-ai/config
*/
import { getProviders } from '@earendil-works/pi-ai'
import type { CacheRetention, ThinkingBudgets, ThinkingLevel, Transport } from '@earendil-works/pi-ai'
import z from 'schemastery'
/** Configuration for one pi-ai provider route. */
export interface PiAiProviderProfile {
/** pi-ai provider catalog name and Harness route key. */
provider: string
/** Provider credential; when absent pi-ai uses its provider-native ambient discovery. */
apiKey?: string
/** Override the selected catalog model's endpoint without changing its protocol metadata. */
baseURL?: string
/** Provider request headers; Harness attribution wins reserved names. */
headers?: Record<string, string>
/** Provider-neutral pi-ai reasoning level. */
reasoning?: ThinkingLevel
/** Token budgets used by reasoning providers that support them. */
thinkingBudgets?: ThinkingBudgets
/** Prompt-cache retention preference. */
cacheRetention?: CacheRetention
/** Streaming transport preference. */
transport?: Transport
/** HTTP/provider SDK timeout in milliseconds. */
timeoutMs?: number
/** WebSocket connection timeout in milliseconds. */
websocketConnectTimeoutMs?: number
/** Provider SDK retry count. */
maxRetries?: number
/** Maximum provider-requested retry delay in milliseconds. */
maxRetryDelayMs?: number
}
/** Plugin configuration: the non-empty provider profiles this instance owns. */
export interface Config {
/** Non-empty set of pi-ai provider routes this adapter instance owns. */
providers: PiAiProviderProfile[]
}
const thinkingBudgets = z.object({
minimal: z.number(),
low: z.number(),
medium: z.number(),
high: z.number(),
})
const profile = z.object({
provider: z.string().required(),
apiKey: z.string(),
baseURL: z.string(),
headers: z.dict(z.string()),
reasoning: z.union(['minimal', 'low', 'medium', 'high', 'xhigh']),
thinkingBudgets,
cacheRetention: z.union(['none', 'short', 'long']),
transport: z.union(['sse', 'websocket', 'websocket-cached', 'auto']),
timeoutMs: z.natural(),
websocketConnectTimeoutMs: z.natural(),
maxRetries: z.natural(),
maxRetryDelayMs: z.natural(),
})
/** Runtime schema for {@link Config}. */
export const Config: z<Config> = z.object({
providers: z.array(profile).required(),
})
/**
* Validate profiles against the installed pi-ai catalog and return a detached
* shallow copy suitable for adapter construction.
* @param profiles - configured provider profiles.
* @returns validated profiles in configuration order.
*/
export function resolveProfiles(profiles: readonly PiAiProviderProfile[]): PiAiProviderProfile[] {
if (profiles.length === 0) throw new Error('llm-pi-ai: providers must contain at least one profile')
const supported = new Set<string>(getProviders())
const seen = new Set<string>()
return profiles.map((source) => {
if (source.provider.length === 0) throw new Error('llm-pi-ai: provider names must be non-empty')
if (!supported.has(source.provider)) throw new Error(`llm-pi-ai: unknown pi-ai provider "${source.provider}"`)
if (seen.has(source.provider)) throw new Error(`llm-pi-ai: duplicate provider profile "${source.provider}"`)
if (source.apiKey !== undefined && source.apiKey.trim().length === 0) {
throw new Error(`llm-pi-ai: provider "${source.provider}" has an empty apiKey; omit it to use ambient authentication`)
}
if (source.baseURL !== undefined && source.baseURL.length === 0) {
throw new Error(`llm-pi-ai: provider "${source.provider}" has an empty baseURL`)
}
seen.add(source.provider)
return {
...source,
...source.headers === undefined ? {} : { headers: { ...source.headers } },
...source.thinkingBudgets === undefined ? {} : { thinkingBudgets: { ...source.thinkingBudgets } },
}
})
}

View File

@@ -0,0 +1,85 @@
/**
* Harness request-history conversion into pi-ai's Context vocabulary.
*
* @module dsh-llm-pi-ai/context
*/
import { CallId } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
import type { Context as PiContext, Message as PiMessage, Tool as PiTool } from '@earendil-works/pi-ai'
import { toPiAssistant } from './replay.ts'
/** Join the text blocks of a harness message. */
function flattenText(message: Message): string {
return message.content
.filter(block => block.type === 'text')
.map(block => block.text)
.join('')
}
/**
* Convert harness history to a pi-ai Context. Tool results need the tool
* NAME (pi-ai's `toolName`), which the harness doesn't carry on the result
* block — it is recovered from the preceding assistant tool-call with the
* same id.
* @param options - the harness request; `options.system` maps to pi-ai's single `systemPrompt` slot.
* @returns the pi-ai context; `tools` is omitted entirely when the request declares none.
*/
export function toPiContext(options: GenerateOptions): PiContext {
const toolNames = new Map<CallId, string>()
const messages: PiMessage[] = []
for (const message of options.messages) {
if (message.role === 'system') {
// pi-ai has a single systemPrompt slot; in-history system messages are
// folded into user messages to preserve order (rare in practice — the
// harness sends the system prompt via options.system).
messages.push({ role: 'user', content: flattenText(message), timestamp: 0 })
continue
}
if (message.role === 'assistant') {
const assistant = toPiAssistant(message)
for (const block of assistant.content) {
if (block.type === 'toolCall') toolNames.set(CallId(block.id), block.name)
}
messages.push(assistant)
continue
}
// user role: text + tool results (each result becomes its own message).
const text = flattenText(message)
const results = message.content.filter(block => block.type === 'tool-result')
if (text.length > 0 || results.length === 0) {
messages.push({ role: 'user', content: text, timestamp: 0 })
}
for (const result of results) {
messages.push({
role: 'toolResult',
toolCallId: result.toolCallId,
toolName: toolNames.get(result.toolCallId) ?? 'unknown',
content: [{
type: 'text',
text: result.content
.filter(block => block.type === 'text')
.map(block => block.text)
.join('') || '(no output)',
}],
isError: result.isError ?? false,
timestamp: 0,
})
}
}
const tools: PiTool[] | undefined = options.tools?.map(tool => ({
name: tool.name,
description: tool.description,
// ToolSchema.parameters is a JSON Schema object; pi-ai's TSchema
// (TypeBox) is structurally JSON Schema, so it assigns directly.
parameters: tool.parameters,
}))
return {
...options.system !== undefined ? { systemPrompt: options.system } : {},
messages,
...tools !== undefined && tools.length > 0 ? { tools } : {},
}
}

View File

@@ -1,76 +1,45 @@
/**
* pi-ai-backed DeepSeek adapter plugin. Same Config shape as
* `@deepseek-ai/dsh-llm-deepseek` (one-line swap in cordis.yml), different
* implementation underneath — see `./adapter.ts` for why both exist.
* Generic pi-ai-backed LLM adapter plugin. One plugin instance registers an
* explicit set of provider profiles; requests select a profile by provider and
* resolve the model dynamically from pi-ai's installed catalog.
*
* ```yaml
* - id: llm
* name: '@deepseek-ai/dsh-llm-pi-ai'
* config:
* apiKey: !!js process.env.DEEPSEEK_API_KEY
* baseURL: !!js process.env.DEEPSEEK_BASE_URL
* models: [deepseek-v4-flash, deepseek-v4-pro]
* reasoning: high
* providers:
* - provider: openai
* apiKey: !!js process.env.OPENAI_API_KEY
* - provider: anthropic
* apiKey: !!js process.env.ANTHROPIC_API_KEY
* - provider: openrouter
* apiKey: !!js process.env.OPENROUTER_API_KEY
* baseURL: https://proxy.example.com/v1
* ```
*
* @module @deepseek-ai/dsh-llm-pi-ai
*/
import type { Context } from 'cordis'
import z from 'schemastery'
import type {} from '@deepseek-ai/dsh-llm'
import { PiAiAdapter } from './adapter.ts'
import type { PiAiReasoning } from './adapter.ts'
import { Config, resolveProfiles } from './config.ts'
export { buildModel, PiAiAdapter } from './adapter.ts'
export type { PiAiAdapterOptions, PiAiReasoning } from './adapter.ts'
export { mapStopReason, mapUsage, toPiContext, toStreamChunks } from './convert.ts'
export { PiAiAdapter } from './adapter.ts'
export type { PiAiAdapterOptions } from './adapter.ts'
export { Config, resolveProfiles } from './config.ts'
export type { PiAiProviderProfile } from './config.ts'
export { toPiContext } from './context.ts'
export { toPiReplayState } from './replay.ts'
export type { PiAiReplayState } from './replay.ts'
export { mapStopReason, mapUsage, toStreamChunks } from './stream.ts'
export const name = 'llm-pi-ai'
export const inject = ['llm']
/**
* 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).
*/
export interface Config {
/** API key; falls back to $DEEPSEEK_API_KEY. Required one way or the other. */
apiKey?: string
/** Endpoint base; falls back to $DEEPSEEK_BASE_URL, then the public API. */
baseURL?: string
/** Model names to register (sent verbatim on the wire). */
models?: string[]
/**
* Thinking level for every request: 'off' disables thinking mode; 'high'
* and 'xhigh' (wire 'max') set the effort. Omitted = provider default
* (thinking enabled), matching llm-deepseek's omission semantics.
*/
reasoning?: PiAiReasoning
}
export const Config: z<Config> = z.object({
apiKey: z.string(),
baseURL: z.string(),
models: z.array(z.string()).default(['deepseek-v4-flash', 'deepseek-v4-pro']),
reasoning: z.union(['off', 'high', 'xhigh']),
})
/** Public API default; the internal endpoint comes from $DEEPSEEK_BASE_URL. */
export const PUBLIC_BASE_URL = 'https://api.deepseek.com'
/** Register one generic pi-ai adapter for all configured provider routes. */
export function apply(ctx: Context, config: Config): void {
const apiKey = config.apiKey ?? process.env.DEEPSEEK_API_KEY
if (apiKey === undefined || apiKey.length === 0) {
throw new Error('llm-pi-ai: an API key is required (Config.apiKey or $DEEPSEEK_API_KEY)')
}
const baseURL = config.baseURL ?? process.env.DEEPSEEK_BASE_URL ?? PUBLIC_BASE_URL
// schemastery's .default() guarantees models is set after validation.
const models = config.models as string[]
ctx.llm.registerAdapter(models, new PiAiAdapter({
apiKey,
baseURL,
reasoning: config.reasoning,
}))
const profiles = resolveProfiles(config.providers)
const adapter = new PiAiAdapter({ profiles })
ctx.llm.registerAdapter(profiles.map(entry => entry.provider), adapter)
}

View File

@@ -0,0 +1,211 @@
/**
* Durable pi-ai replay metadata and assistant-history reconstruction.
*
* Harness content remains the durable source for text and tool calls. This
* module stores only the provider-native metadata needed to reconstruct a
* pi-ai assistant message on a later request.
*
* @module dsh-llm-pi-ai/replay
*/
import { LlmError } from '@deepseek-ai/dsh-llm'
import type { Message } from '@deepseek-ai/dsh-llm'
import type { Api, AssistantMessage, Usage as PiUsage } from '@earendil-works/pi-ai'
type PiAiReplayBlock =
| { type: 'text'; textSignature?: string }
| { type: 'reasoning'; thinkingSignature?: string; redacted?: boolean }
| { type: 'tool-call'; thoughtSignature?: string }
/** Versioned adapter-private projection required to replay a pi-ai response. */
export interface PiAiReplayState {
kind: 'pi-ai'
version: 1
api: Api
provider: string
model: string
responseModel?: string
responseId?: string
stopReason: AssistantMessage['stopReason']
blocks: PiAiReplayBlock[]
}
/** Parse tool-call argument JSON; tolerate model malformations with {}. */
function parseArguments(raw: string): Record<string, unknown> {
try {
const parsed: unknown = JSON.parse(raw)
if (typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed)) {
return parsed as Record<string, unknown>
}
} catch {
// fall through
}
return {}
}
/** Construct the zero usage value required by historical pi-ai messages. */
function emptyPiUsage(): PiUsage {
return {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
}
}
/**
* Project a successful pi-ai response into the minimal durable replay state.
* @param message - completed native pi-ai assistant response.
* @returns the versioned lossless-JSON replay projection.
*/
export function toPiReplayState(message: AssistantMessage): PiAiReplayState {
return {
kind: 'pi-ai',
version: 1,
api: message.api,
provider: message.provider,
model: message.model,
...message.responseModel === undefined ? {} : { responseModel: message.responseModel },
...message.responseId === undefined ? {} : { responseId: message.responseId },
stopReason: message.stopReason,
blocks: message.content.map((block): PiAiReplayBlock => {
switch (block.type) {
case 'text': return {
type: 'text',
...block.textSignature === undefined ? {} : { textSignature: block.textSignature },
}
case 'thinking': return {
type: 'reasoning',
...block.thinkingSignature === undefined ? {} : { thinkingSignature: block.thinkingSignature },
...block.redacted === undefined ? {} : { redacted: block.redacted },
}
case 'toolCall': return {
type: 'tool-call',
...block.thoughtSignature === undefined ? {} : { thoughtSignature: block.thoughtSignature },
}
}
}),
}
}
function invalidReplay(message: string): never {
throw new LlmError(`invalid pi-ai replay state: ${message}`, 'INVALID_REPLAY_STATE')
}
/** Validate the adapter-private state before it reaches pi-ai. */
function readReplayState(value: unknown): PiAiReplayState {
if (typeof value !== 'object' || value === null || Array.isArray(value)) return invalidReplay('expected an object')
const state = value as Record<string, unknown>
if (state['kind'] !== 'pi-ai') return invalidReplay('unknown state kind')
if (state['version'] !== 1) return invalidReplay(`unsupported version ${String(state['version'])}`)
for (const key of ['api', 'provider', 'model'] as const) {
if (typeof state[key] !== 'string' || state[key].length === 0) return invalidReplay(`${key} must be a non-empty string`)
}
if (!['stop', 'length', 'toolUse', 'error', 'aborted'].includes(String(state['stopReason']))) {
return invalidReplay('unknown stopReason')
}
if (state['responseModel'] !== undefined && typeof state['responseModel'] !== 'string') return invalidReplay('responseModel must be a string')
if (state['responseId'] !== undefined && typeof state['responseId'] !== 'string') return invalidReplay('responseId must be a string')
if (!Array.isArray(state['blocks'])) return invalidReplay('blocks must be an array')
for (const [index, value] of state['blocks'].entries()) {
if (typeof value !== 'object' || value === null || Array.isArray(value)) return invalidReplay(`block ${index} must be an object`)
const block = value as Record<string, unknown>
if (!['text', 'reasoning', 'tool-call'].includes(String(block['type']))) return invalidReplay(`block ${index} has an unknown type`)
for (const signature of ['textSignature', 'thinkingSignature', 'thoughtSignature'] as const) {
if (block[signature] !== undefined && typeof block[signature] !== 'string') return invalidReplay(`block ${index} ${signature} must be a string`)
}
if (block['redacted'] !== undefined && typeof block['redacted'] !== 'boolean') return invalidReplay(`block ${index} redacted must be boolean`)
}
return state as unknown as PiAiReplayState
}
/** Convert provider-neutral blocks without trusting them as same-model replay. */
function foreignAssistant(message: Message): AssistantMessage {
const content: AssistantMessage['content'] = []
for (const block of message.content) {
switch (block.type) {
case 'text': content.push({ type: 'text', text: block.text }); break
case 'reasoning': content.push({ type: 'thinking', thinking: block.text }); break
case 'tool-call': content.push({
type: 'toolCall',
id: block.id,
name: block.name,
arguments: parseArguments(block.arguments),
}); break
default:
// plugin-added block types are not representable in pi-ai.
break
}
}
return {
role: 'assistant',
content,
// Deliberately never equals a catalog API: absent replay state is foreign
// even if provenance names the same provider/model as this request.
api: 'dsh-foreign',
provider: message.provenance?.provider ?? 'dsh-foreign',
model: message.provenance?.model ?? 'dsh-foreign',
usage: emptyPiUsage(),
stopReason: content.some(piece => piece.type === 'toolCall') ? 'toolUse' : 'stop',
timestamp: 0,
}
}
/** Recombine durable Harness content with validated pi-ai replay metadata. */
function replayedAssistant(message: Message, rawState: unknown): AssistantMessage {
const state = readReplayState(rawState)
const provenance = message.provenance
if (state.provider !== provenance?.provider) return invalidReplay('provider does not match assistant provenance')
if (state.model !== provenance.model) return invalidReplay('model does not match assistant provenance')
if (state.blocks.length !== message.content.length) return invalidReplay('block count does not match assistant content')
const content: AssistantMessage['content'] = message.content.map((block, index) => {
const replay = state.blocks[index]
if (replay === undefined || replay.type !== block.type) return invalidReplay(`block ${index} does not match assistant content`)
switch (block.type) {
case 'text': return {
type: 'text',
text: block.text,
...replay.type === 'text' && replay.textSignature !== undefined ? { textSignature: replay.textSignature } : {},
}
case 'reasoning': return {
type: 'thinking',
thinking: block.text,
...replay.type === 'reasoning' && replay.thinkingSignature !== undefined ? { thinkingSignature: replay.thinkingSignature } : {},
...replay.type === 'reasoning' && replay.redacted !== undefined ? { redacted: replay.redacted } : {},
}
case 'tool-call': return {
type: 'toolCall',
id: block.id,
name: block.name,
arguments: parseArguments(block.arguments),
...replay.type === 'tool-call' && replay.thoughtSignature !== undefined ? { thoughtSignature: replay.thoughtSignature } : {},
}
/* v8 ignore next -- readReplayState rejects unknown replay tags, so an equal plugin-added Harness tag cannot reach this switch */
default: return invalidReplay(`block ${index} has an unsupported Harness type`)
}
})
return {
role: 'assistant',
content,
api: state.api,
provider: state.provider,
model: state.model,
...state.responseModel === undefined ? {} : { responseModel: state.responseModel },
...state.responseId === undefined ? {} : { responseId: state.responseId },
usage: emptyPiUsage(),
stopReason: state.stopReason,
timestamp: 0,
}
}
/**
* Convert one durable Harness assistant message into pi-ai history.
* @param message - assistant content with optional adapter-owned replay metadata.
* @returns a native pi-ai assistant message reconstructed from durable content.
*/
export function toPiAssistant(message: Message): AssistantMessage {
const replayState = message.provenance?.replayState
return replayState === undefined ? foreignAssistant(message) : replayedAssistant(message, replayState)
}

View File

@@ -1,152 +1,17 @@
/**
* Bidirectional mapping between the harness vocabulary and pi-ai's:
* Convert harness requests to pi-ai context and pi-ai assistant events to harness stream chunks.
* pi-ai parses tool arguments while the harness preserves raw JSON, so conversion parses inbound
* arguments and re-stringifies outbound values while the adapter restores provider payloads.
* In-stream pi-ai errors become harness error/aborted finishes, and its reasoning tokens remain
* folded into output usage because it reports no separate count.
* @module dsh-llm-pi-ai/convert
* pi-ai assistant event translation into the Harness streaming protocol.
*
* pi-ai tool-call arguments are parsed objects while the Harness keeps their
* raw JSON representation. pi-ai also reports failures as terminal stream
* events, which this module maps into Harness finish chunks.
*
* @module dsh-llm-pi-ai/stream
*/
import { CallId, LlmError } from '@deepseek-ai/dsh-llm'
import type { FinishReason, GenerateOptions, Message, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm'
import type {
AssistantMessage,
AssistantMessageEvent,
Context as PiContext,
Message as PiMessage,
Tool as PiTool,
Usage as PiUsage,
} from '@earendil-works/pi-ai'
/** Join the text blocks of a harness message. */
function flattenText(message: Message): string {
return message.content
.filter(block => block.type === 'text')
.map(block => block.text)
.join('')
}
/** Parse tool-call argument JSON; tolerate model malformations with {}. */
function parseArguments(raw: string): Record<string, unknown> {
try {
const parsed: unknown = JSON.parse(raw)
if (typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed)) {
return parsed as Record<string, unknown>
}
} catch {
// fall through
}
return {}
}
/**
* Convert harness history to a pi-ai Context. Tool results need the tool
* NAME (pi-ai's `toolName`), which the harness doesn't carry on the result
* block it is recovered from the preceding assistant tool-call with the
* same id.
* @param options - the harness request; `options.system` maps to pi-ai's single `systemPrompt` slot.
* @returns the pi-ai context; `tools` is omitted entirely when the request declares none.
*/
export function toPiContext(options: GenerateOptions): PiContext {
const toolNames = new Map<CallId, string>()
const messages: PiMessage[] = []
for (const message of options.messages) {
if (message.role === 'system') {
// pi-ai has a single systemPrompt slot; in-history system messages are
// folded into user messages to preserve order (rare in practice — the
// harness sends the system prompt via options.system).
messages.push({ role: 'user', content: flattenText(message), timestamp: 0 })
continue
}
if (message.role === 'assistant') {
const content: AssistantMessage['content'] = []
for (const block of message.content) {
switch (block.type) {
case 'text':
content.push({ type: 'text', text: block.text })
break
case 'reasoning':
// Without this wire-field name, pi-ai replays an empty `reasoning_content`, violating
// DeepSeek's thinking-mode passback rule on tool-call turns.
content.push({ type: 'thinking', thinking: block.text, thinkingSignature: 'reasoning_content' })
break
case 'tool-call':
toolNames.set(block.id, block.name)
content.push({
type: 'toolCall',
id: block.id,
name: block.name,
arguments: parseArguments(block.arguments),
})
break
default:
// plugin-added block types: not representable here.
break
}
}
messages.push({
role: 'assistant',
content,
api: 'openai-completions',
provider: 'deepseek',
model: options.model,
usage: emptyPiUsage(),
stopReason: content.some(piece => piece.type === 'toolCall') ? 'toolUse' : 'stop',
timestamp: 0,
})
continue
}
// user role: text + tool results (each result becomes its own message).
const text = flattenText(message)
const results = message.content.filter(block => block.type === 'tool-result')
if (text.length > 0 || results.length === 0) {
messages.push({ role: 'user', content: text, timestamp: 0 })
}
for (const result of results) {
messages.push({
role: 'toolResult',
toolCallId: result.toolCallId,
toolName: toolNames.get(result.toolCallId) ?? 'unknown',
content: [{
type: 'text',
text: result.content
.filter(block => block.type === 'text')
.map(block => block.text)
.join('') || '(no output)',
}],
isError: result.isError ?? false,
timestamp: 0,
})
}
}
const tools: PiTool[] | undefined = options.tools?.map(tool => ({
name: tool.name,
description: tool.description,
// ToolSchema.parameters is a JSON Schema object; pi-ai's TSchema
// (TypeBox) is structurally JSON Schema, so it assigns directly.
parameters: tool.parameters,
}))
return {
...options.system !== undefined ? { systemPrompt: options.system } : {},
messages,
...tools !== undefined && tools.length > 0 ? { tools } : {},
}
}
function emptyPiUsage(): PiUsage {
return {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
}
}
import type { FinishReason, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm'
import type { AssistantMessage, AssistantMessageEvent, Usage as PiUsage } from '@earendil-works/pi-ai'
import { toPiReplayState } from './replay.ts'
/**
* Map pi-ai usage (reasoning folded into output by pi-ai).
@@ -259,7 +124,7 @@ 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) }
yield { type: 'finish', reason: mapStopReason(event.message), replayState: toPiReplayState(event.message) }
return
case 'error':
// In-stream error delivery (pi-ai's style) → error finish chunk

View File

@@ -3,26 +3,33 @@ import { Context } from 'cordis'
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
import type { Message, ToolSchema } from '@deepseek-ai/dsh-llm'
import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai'
import type { Config } from '@deepseek-ai/dsh-llm-pi-ai'
import type { PiAiProviderProfile } from '@deepseek-ai/dsh-llm-pi-ai'
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
import { assemble, type AssembledResult } from './assemble.ts'
/**
* Real-API e2e for the pi-ai-backed adapter: V4 Flash + V4 Pro across all
* reasoning levels the adapter exposes (off / high / xhigh→wire 'max').
* Mirrors the llm-deepseek matrix so the two independent implementations
* verify the same StreamChunk contract. Key-gated.
* Real-API e2e for the pi-ai-backed adapter: V4 Flash + V4 Pro with provider
* defaults and representative high/xhigh reasoning. Mirrors the native
* adapter's StreamChunk contract and exercises a replayed tool follow-up.
* Key-gated.
*/
const FLASH = 'deepseek-v4-flash'
const PRO = 'deepseek-v4-pro'
const contexts: Context[] = []
async function harness(model: string, config: Partial<Config> = {}) {
async function harness(_model: string, config: Partial<PiAiProviderProfile> = {}) {
const ctx = new Context()
contexts.push(ctx)
await ctx.plugin(LlmService)
await ctx.plugin(LlmPiAi, { models: [model], ...config })
await ctx.plugin(LlmPiAi, {
providers: [{
provider: 'deepseek',
...process.env.DEEPSEEK_API_KEY === undefined ? {} : { apiKey: process.env.DEEPSEEK_API_KEY },
...process.env.DEEPSEEK_BASE_URL === undefined ? {} : { baseURL: process.env.DEEPSEEK_BASE_URL },
...config,
}],
})
return ctx
}
@@ -56,8 +63,8 @@ const weatherTool: ToolSchema = {
}
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () => {
it.each([FLASH, PRO])('%s + reasoning off: plain text generation', async (model) => {
const ctx = await harness(model, { reasoning: 'off' })
it.each([FLASH, PRO])('%s + provider-default reasoning: plain text generation', async (model) => {
const ctx = await harness(model)
const result = await assemble(ctx,{
model,
messages: ask('Reply with exactly the word: pong'),
@@ -65,7 +72,6 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () =>
})
expect(result.finish.kind).toBe('stop')
expect(textOf(result).toLowerCase()).toContain('pong')
expect(result.message.content.some(block => block.type === 'reasoning')).toBe(false)
})
it.each([FLASH, PRO])('%s + reasoning high: reasoning blocks present', async (model) => {
@@ -99,7 +105,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () =>
model: PRO,
messages: [
...ask('What is the weather in Paris right now? Use the get_weather tool.'),
{ role: 'assistant', content: first.message.content },
first.message,
{
role: 'user',
content: [{
@@ -123,9 +129,9 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () =>
const deepseekCtx = new Context()
contexts.push(deepseekCtx)
await deepseekCtx.plugin(LlmService)
await deepseekCtx.plugin(LlmDeepSeek, { models: [FLASH], thinking: 'disabled' })
await deepseekCtx.plugin(LlmDeepSeek, { thinking: 'disabled' })
const piCtx = await harness(FLASH, { reasoning: 'off' })
const piCtx = await harness(FLASH)
const prompt = ask('Reply with exactly the word: pong')
const [fromDeepSeek, fromPiAi] = await Promise.all([

View File

@@ -2,34 +2,35 @@ 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, { CallId, userAgent } from '@deepseek-ai/dsh-llm'
import LlmService, { LlmError, userAgent } from '@deepseek-ai/dsh-llm'
import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai'
import { buildModel, PiAiAdapter } from '@deepseek-ai/dsh-llm-pi-ai'
import { PiAiAdapter, resolveProfiles } from '@deepseek-ai/dsh-llm-pi-ai'
import { assemble } from './assemble.ts'
/** Scripted SSE responses, one per request (OpenAI chat-completions shape). */
interface MockServer {
url: string
paths: string[]
requests: unknown[]
/** Header bags of received requests, in order (parallel to `requests`). */
headers: IncomingMessage['headers'][]
close(): Promise<void>
}
const servers: Server[] = []
afterEach(async () => {
vi.unstubAllEnvs()
await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve))))
})
async function mockServer(script: { status?: number; events?: string[]; body?: string }[]): Promise<MockServer> {
async function mockServer(script: { status?: number; events?: string[]; body?: string; delayMs?: number }[]): Promise<MockServer> {
const paths: string[] = []
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))
paths.push(request.url ?? '')
requests.push(body.length === 0 ? undefined : JSON.parse(body))
headers.push(request.headers)
const behavior = script.shift() ?? { status: 500, body: 'script exhausted' }
if (behavior.status !== undefined && behavior.status !== 200) {
@@ -38,20 +39,22 @@ async function mockServer(script: { status?: number; events?: string[]; body?: s
return
}
response.writeHead(200, { 'content-type': 'text/event-stream' })
for (const event of behavior.events ?? []) response.write(`data: ${event}\n\n`)
response.end()
let index = 0
const writeNext = (): void => {
const event = behavior.events?.[index++]
if (event === undefined) { response.end(); return }
response.write(`data: ${event}\n\n`)
if (behavior.delayMs === undefined) writeNext()
else setTimeout(writeNext, behavior.delayMs)
}
writeNext()
})
})
servers.push(server)
await new Promise<void>(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,
close: () => new Promise(resolve => server.close(() => { resolve() })),
}
return { url: `http://127.0.0.1:${address.port}`, paths, requests, headers }
}
const textEvents = [
@@ -61,347 +64,231 @@ const textEvents = [
'[DONE]',
]
const toolEvents = [
'{"choices":[{"delta":{"role":"assistant","content":null},"index":0,"finish_reason":null}]}',
'{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"get_weather","arguments":""}}]},"index":0,"finish_reason":null}]}',
'{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"city\\":\\"Paris\\"}"}}]},"index":0,"finish_reason":null}]}',
'{"choices":[{"delta":{},"index":0,"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":20,"completion_tokens":6}}',
'[DONE]',
]
const thinkingEvents = [
'{"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""},"index":0,"finish_reason":null}]}',
'{"choices":[{"delta":{"reasoning_content":"pondering"},"index":0,"finish_reason":null}]}',
'{"choices":[{"delta":{"content":"answer","reasoning_content":null},"index":0,"finish_reason":null}]}',
'{"choices":[{"delta":{},"index":0,"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":9}}',
'[DONE]',
]
async function harness(baseURL: string, config: object = {}) {
async function harness(baseURL: string, overrides: Record<string, unknown> = {}): Promise<Context> {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmPiAi, { apiKey: 'test-key', baseURL, models: ['deepseek-v4-flash'], ...config })
await ctx.plugin(LlmPiAi, {
providers: [{ provider: 'deepseek', apiKey: 'test-key', baseURL, ...overrides }],
})
return ctx
}
describe('PiAiAdapter against a mock server', () => {
it('streams a text generation through the assembler', async () => {
describe('PiAiAdapter provider routing', () => {
it('resolves a catalog model dynamically and uses a private endpoint', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
const result = await assemble(ctx, {
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
})
expect(result.message.content).toEqual([{ type: 'text', text: 'hello' }])
expect(result.finish).toEqual({ kind: 'stop' })
expect(result.usage).toMatchObject({ inputTokens: 3, outputTokens: 1 })
expect(result.usage).toEqual({ inputTokens: 3, outputTokens: 1 })
expect(server.paths).toEqual(['/chat/completions'])
})
// Attribution reaches the wire through pi-ai's headers hook: the exact
// shared User-Agent, and no provider-specific headers under the
// User-Agent-only contract.
it('merges profile headers with Harness attribution winning', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url, {
headers: { 'x-company': 'private', 'User-Agent': 'wrong' },
})
await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(server.headers[0]?.['x-company']).toBe('private')
expect(server.headers[0]?.['user-agent']).toBe(userAgent())
expect(server.headers[0]).not.toHaveProperty('http-referer')
expect(server.headers[0]).not.toHaveProperty('x-openrouter-title')
expect(server.headers[0]).not.toHaveProperty('x-openrouter-categories')
})
it('streams tool calls with re-stringified arguments', async () => {
const server = await mockServer([{ events: toolEvents }])
const ctx = await harness(server.url)
const result = await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'weather?' }] }],
tools: [{
name: 'get_weather',
description: 'Get weather',
parameters: { type: 'object', properties: { city: { type: 'string' } } },
}],
it('forwards common stream options and profile reasoning', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url, {
reasoning: 'xhigh',
cacheRetention: 'none',
transport: 'sse',
timeoutMs: 5000,
websocketConnectTimeoutMs: 3000,
maxRetries: 0,
maxRetryDelayMs: 10,
thinkingBudgets: { high: 2048 },
})
expect(result.finish).toEqual({ kind: 'tool-calls' })
const call = result.message.content.find(block => block.type === 'tool-call')
expect(call).toMatchObject({ name: 'get_weather', arguments: '{"city":"Paris"}' })
})
it('maps reasoning_content streams to reasoning blocks', async () => {
const server = await mockServer([{ events: thinkingEvents }])
const ctx = await harness(server.url, { reasoning: 'high' })
const result = await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'think' }] }],
})
expect(result.message.content).toEqual([
{ type: 'reasoning', text: 'pondering' },
{ type: 'text', text: 'answer' },
])
})
it('sends DeepSeek thinking fields when reasoning is configured', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url, { reasoning: 'xhigh' })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(server.requests[0]).toMatchObject({
thinking: { type: 'enabled' },
reasoning_effort: 'max', // xhigh maps to max via thinkingLevelMap
})
})
it('disables thinking for reasoning: off', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url, { reasoning: 'off' })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(server.requests[0]).toMatchObject({ thinking: { type: 'disabled' } })
})
it('injects stop sequences through onPayload', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [], stop: ['END'] })
expect(server.requests[0]).toMatchObject({ stop: ['END'] })
})
it('scrubs pi-ai\'s own per-tool strict default through onPayload', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
await assemble(ctx,{
await assemble(ctx, {
model: 'deepseek-v4-flash',
messages: [],
tools: [
{ name: 'alpha', description: 'a', parameters: {} },
{ name: 'beta', description: 'b', parameters: {} },
],
temperature: 0.2,
maxTokens: 77,
sessionId: 'session-for-pi' as never,
})
// pi-ai stamps `strict` on every serialized tool function; the harness
// contract has none and the hand-rolled twin sends no such field, so the
// payload fixup must have deleted it from every tool.
const request = server.requests[0] as { tools: { function: { name: string; strict?: boolean } }[] }
expect(request.tools.map(tool => tool.function.name)).toEqual(['alpha', 'beta'])
for (const tool of request.tools) {
expect('strict' in tool.function).toBe(false)
}
})
it('preserves raw replayed tool-call arguments in the provider payload', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
await assemble(ctx,{
expect(server.requests[0]).toMatchObject({
model: 'deepseek-v4-flash',
messages: [{
role: 'assistant',
content: [{ type: 'tool-call', id: CallId('broken'), name: 'f', arguments: '{broken' }],
}],
temperature: 0.2,
max_completion_tokens: 77,
thinking: { type: 'enabled' },
reasoning_effort: 'max',
})
const request = server.requests[0] as { messages: { role: string; tool_calls?: { id: string; function: { arguments: string } }[] }[] }
const assistant = request.messages.find(message => message.role === 'assistant')
expect(assistant?.tool_calls?.[0]?.function.arguments).toBe('{broken')
})
it('maps HTTP errors to error finish chunks (pi-ai in-stream style)', async () => {
const server = await mockServer([{
status: 401,
body: JSON.stringify({ error: { message: 'bad key' } }),
}])
it('preserves omitted profile options when constructing the adapter directly', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['deepseek'], new PiAiAdapter({
profiles: [{ provider: 'deepseek', apiKey: 'test-key', baseURL: server.url }],
}))
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.message.content).toEqual([{ type: 'text', text: 'hello' }])
})
it('rejects stop sequences rather than silently ignoring them', async () => {
const server = await mockServer([])
const ctx = await harness(server.url)
const result = await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toMatchObject({ kind: 'error', code: 'AUTH' })
expect((result.finish as { message: string }).message).toMatch(/bad key|401/)
await expect(assemble(ctx, { model: 'deepseek-v4-flash', messages: [], stop: ['END'] }))
.rejects.toMatchObject({ code: 'UNSUPPORTED_OPTION' })
expect(server.requests).toEqual([])
})
it('rejects unknown catalog models before network I/O', async () => {
const server = await mockServer([])
const ctx = await harness(server.url)
await expect(assemble(ctx, { model: 'not-in-the-catalog', messages: [] }))
.rejects.toMatchObject({ code: 'UNKNOWN_MODEL' })
expect(server.requests).toEqual([])
})
it('uses the catalog API implementation, including OpenAI Responses', async () => {
const server = await mockServer([{ status: 401, body: JSON.stringify({ error: { message: 'expected mock failure' } }) }])
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmPiAi, {
providers: [{ provider: 'openai', apiKey: 'test-key', baseURL: `${server.url}/v1`, maxRetries: 0 }],
})
const result = await assemble(ctx, { provider: 'openai', model: 'gpt-4.1', messages: [] })
expect(result.finish.kind).toBe('error')
expect(server.paths).toEqual(['/v1/responses'])
})
it.each([
[401, 'AUTH'],
[400, 'INVALID_REQUEST'],
[429, 'RATE_LIMIT'],
[500, 'SERVER'],
] as const)('maps HTTP %s to stable error code %s', async (status, code) => {
] as const)('maps HTTP %s failures to %s', async (status, code) => {
const server = await mockServer([{ status, body: JSON.stringify({ error: { message: `provider ${status}` } }) }])
const ctx = await harness(server.url)
const result = await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
const ctx = await harness(server.url, { maxRetries: 0 })
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toMatchObject({ kind: 'error', code })
})
})
it('registers/unregisters models on the llm service (HMR safety)', async () => {
describe('provider profile lifecycle', () => {
it('registers every profile atomically and unregisters on dispose', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const fiber = await ctx.plugin(LlmPiAi, { apiKey: 'k', baseURL: 'http://127.0.0.1:1' })
expect(ctx.llm.models().sort()).toEqual(['deepseek-v4-flash', 'deepseek-v4-pro'])
const fiber = await ctx.plugin(LlmPiAi, {
providers: [{ provider: 'openai' }, { provider: 'anthropic' }],
})
expect(ctx.llm.listProviders()).toEqual([
{ id: 'openai', name: 'openai' },
{ id: 'anthropic', name: 'anthropic' },
])
await fiber.dispose()
expect(ctx.llm.models()).toEqual([])
expect(ctx.llm.listProviders()).toEqual([])
})
it('throws a clear error when no API key is available', async () => {
const previous = process.env.DEEPSEEK_API_KEY
delete process.env.DEEPSEEK_API_KEY
try {
const ctx = new Context()
await ctx.plugin(LlmService)
await expect(ctx.plugin(LlmPiAi, {})).rejects.toThrow(/an API key is required/)
} finally {
if (previous !== undefined) process.env.DEEPSEEK_API_KEY = previous
}
})
})
describe('option spreads and env fallbacks', () => {
it('forwards temperature, maxTokens, and signal', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
const controller = new AbortController()
await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
temperature: 0.5,
maxTokens: 40,
signal: controller.signal,
it('exposes the installed pi-ai model catalog through provider-neutral metadata', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmPiAi, { providers: [{ provider: 'openai' }] })
const models = await ctx.llm.listModels('openai')
expect(models.find(model => model.id === 'gpt-4.1')).toEqual({
provider: 'openai', id: 'gpt-4.1', name: 'GPT-4.1',
})
expect(server.requests[0]).toMatchObject({ temperature: 0.5, max_tokens: 40 })
expect(models.every(model => model.provider === 'openai')).toBe(true)
})
it('falls back to DEEPSEEK_API_KEY / DEEPSEEK_BASE_URL env vars', async () => {
it('accepts absent credentials for pi-ai ambient authentication', async () => {
vi.stubEnv('DEEPSEEK_API_KEY', 'ambient-key')
const server = await mockServer([{ events: textEvents }])
vi.stubEnv('DEEPSEEK_API_KEY', 'env-key')
vi.stubEnv('DEEPSEEK_BASE_URL', server.url)
try {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmPiAi, { models: ['deepseek-v4-flash'] })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(server.requests).toHaveLength(1)
} finally {
vi.unstubAllEnvs()
const ctx = await harness(server.url, { apiKey: undefined })
await assemble(ctx, { model: 'deepseek-v4-flash', messages: [] })
expect(server.headers[0]?.authorization).toBe('Bearer ambient-key')
})
it('validates empty, duplicate, unknown, and explicitly blank profiles', () => {
expect(() => resolveProfiles([])).toThrow(/at least one/)
expect(() => resolveProfiles([{ provider: '' }])).toThrow(/non-empty/)
expect(() => resolveProfiles([{ provider: 'not-real' }])).toThrow(/unknown/)
expect(() => resolveProfiles([{ provider: 'openai' }, { provider: 'openai' }])).toThrow(/duplicate/)
expect(() => resolveProfiles([{ provider: 'openai', apiKey: '' }])).toThrow(/empty apiKey/)
expect(() => resolveProfiles([{ provider: 'openai', apiKey: ' ' }])).toThrow(/empty apiKey/)
expect(() => resolveProfiles([{ provider: 'openai', baseURL: '' }])).toThrow(/empty baseURL/)
})
it('rejects negative or fractional stream tunables at schema validation', () => {
const invalid = [
{ timeoutMs: -1 },
{ websocketConnectTimeoutMs: -1 },
{ maxRetries: -1 },
{ maxRetries: 0.5 },
{ maxRetryDelayMs: -1 },
]
for (const entry of invalid) {
expect(() => new LlmPiAi.Config({ providers: [{ provider: 'openai', ...entry }] })).toThrow()
}
})
it('defaults to the public base URL without config or env', async () => {
vi.stubEnv('DEEPSEEK_API_KEY', 'k')
vi.stubEnv('DEEPSEEK_BASE_URL', undefined)
try {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmPiAi, {})
expect(ctx.llm.models().length).toBeGreaterThan(0)
} finally {
vi.unstubAllEnvs()
}
it('constructs the adapter directly and rejects routes it does not own', async () => {
const adapter = new PiAiAdapter({ profiles: [{ provider: 'openai' }] })
await expect(adapter.listModels('anthropic')).rejects.toMatchObject({ code: 'NO_ADAPTER' })
await expect((async () => {
for await (const _chunk of adapter.stream({ provider: 'anthropic', model: 'claude-sonnet-4', messages: [] })) { /* drain */ }
})()).rejects.toMatchObject({ code: 'NO_ADAPTER' })
expect(new LlmError('x', 'X')).toBeInstanceOf(Error)
})
})
describe('buildModel', () => {
it('builds a DeepSeek-compat openai-completions model descriptor', () => {
const model = buildModel('deepseek-v4-pro', { apiKey: 'k', baseURL: 'http://x', reasoning: 'high' })
expect(model).toMatchObject({
id: 'deepseek-v4-pro',
api: 'openai-completions',
describe('abort wiring', () => {
it('resolves catalog endpoints without an override before honoring pre-abort', async () => {
const adapter = new PiAiAdapter({ profiles: [{ provider: 'deepseek', apiKey: 'test-key', maxRetries: 0 }] })
const controller = new AbortController()
controller.abort('already stopped')
const chunks = []
for await (const chunk of adapter.stream({
provider: 'deepseek',
baseUrl: 'http://x',
reasoning: true,
compat: { thinkingFormat: 'deepseek', requiresReasoningContentOnAssistantMessages: true },
})
})
it('keeps reasoning true even for off (pi-ai gates the thinking field on it)', () => {
// 'off' yields {thinking: {type: 'disabled'}} on the wire — pi-ai only
// emits the field at all when model.reasoning is true.
expect(buildModel('m', { apiKey: 'k', baseURL: 'http://x', reasoning: 'off' }).reasoning).toBe(true)
})
it('adapter is constructible directly for embedding', () => {
expect(new PiAiAdapter({ apiKey: 'k', baseURL: 'http://x' })).toBeInstanceOf(PiAiAdapter)
})
})
describe('provider reasoning, passback, and early-stream cancellation', () => {
it('defaults omitted reasoning config to thinking ENABLED (provider default)', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url) // no reasoning key at all
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
const request = server.requests[0] as Record<string, unknown>
expect(request.thinking).toEqual({ type: 'enabled' })
expect('reasoning_effort' in request).toBe(false)
})
it('replays reasoning_content on assistant tool-call turns (passback rule)', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [
{ role: 'user', content: [{ type: 'text', text: 'weather?' }] },
{
role: 'assistant',
content: [
{ type: 'reasoning', text: 'I should check.' },
{ type: 'tool-call', id: CallId('c1'), name: 'get_weather', arguments: '{"city":"Paris"}' },
],
},
{
role: 'user',
content: [{ type: 'tool-result', toolCallId: CallId('c1'), content: [{ type: 'text', text: 'Sunny' }] }],
},
],
})
const request = server.requests[0] as { messages: { role: string; reasoning_content?: string }[] }
const assistant = request.messages.find(message => message.role === 'assistant')
expect(assistant?.reasoning_content).toBe('I should check.')
})
it('aborts the upstream request when the consumer stops streaming early', async () => {
// Slow server: write one chunk, then hold the connection open and record
// whether the socket closes (the adapter must cancel on early break).
let socketClosed = false
const server = createServer((request: IncomingMessage, response: ServerResponse) => {
request.on('data', () => undefined)
request.on('end', () => {
response.writeHead(200, { 'content-type': 'text/event-stream' })
response.write(`data: ${textEvents[0]}\n\n`)
response.write(`data: ${textEvents[1]}\n\n`)
// never finish; rely on client abort
request.socket.on('close', () => { socketClosed = true })
})
})
servers.push(server)
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
const address = server.address()
if (address === null || typeof address === 'string') throw new Error('no port')
const ctx = await harness(`http://127.0.0.1:${address.port}`)
for await (const chunk of ctx.llm.stream({ model: 'deepseek-v4-flash', messages: [] })) {
if (chunk.type === 'text-delta') break // stop early mid-stream
}
// The finally-abort must reach the server as a closed socket.
await vi.waitFor(() => { expect(socketClosed).toBe(true) }, { timeout: 5_000 })
})
})
describe('caller cancellation', () => {
it('honors a pre-aborted caller signal', async () => {
const ctx = await harness('http://127.0.0.1:1')
const controller = new AbortController()
controller.abort('already cancelled')
// pi-ai surfaces the abort as an in-stream error event → aborted finish.
const result = await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
signal: controller.signal,
})
})) chunks.push(chunk)
expect(chunks.at(-1)).toMatchObject({ type: 'finish', reason: { kind: 'aborted' } })
})
it('honors a pre-aborted caller signal', async () => {
const server = await mockServer([{ events: textEvents, delayMs: 20 }])
const ctx = await harness(server.url)
const controller = new AbortController()
controller.abort('already stopped')
const result = await assemble(ctx, { model: 'deepseek-v4-flash', messages: [], signal: controller.signal })
expect(result.finish.kind).toBe('aborted')
})
it('propagates a mid-stream caller abort to the upstream request', async () => {
const server = await mockServer([{ events: textEvents }])
it('forwards an abort that arrives while provider streaming is active', async () => {
const server = await mockServer([{ events: textEvents, delayMs: 30 }])
const ctx = await harness(server.url)
const controller = new AbortController()
const pending = assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
signal: controller.signal,
const resultPromise = assemble(ctx, {
model: 'deepseek-v4-flash', messages: [], signal: controller.signal,
})
controller.abort()
const result = await pending
// Either the abort lands before any chunk (aborted) or after the tiny
// mock stream finished (stop) — both are valid races; never a hang.
expect(['aborted', 'stop']).toContain(result.finish.kind)
setTimeout(() => { controller.abort('stopped during stream') }, 10)
const result = await resultPromise
expect(result.finish.kind).toBe('aborted')
})
it('aborts upstream when a consumer stops early', async () => {
const server = await mockServer([{ events: textEvents, delayMs: 30 }])
const ctx = await harness(server.url)
for await (const chunk of ctx.llm.stream({ provider: 'deepseek', model: 'deepseek-v4-flash', messages: [] })) {
if (chunk.type === 'block-start') break
}
await new Promise(resolve => setTimeout(resolve, 20))
expect(server.requests).toHaveLength(1)
})
})

View File

@@ -15,11 +15,19 @@ export interface AssembledResult {
finish: FinishReason
}
export async function assemble(ctx: Context, options: GenerateOptions): Promise<AssembledResult> {
export async function assemble(ctx: Context, options: Omit<GenerateOptions, 'provider'> & { provider?: string }): Promise<AssembledResult> {
const assembler = new BlockAssembler()
for await (const chunk of ctx.llm.stream(options)) assembler.push(chunk)
const request = { provider: 'deepseek', ...options }
for await (const chunk of ctx.llm.stream(request)) assembler.push(chunk)
return {
message: assembler.message(),
message: {
...assembler.message(),
provenance: {
provider: request.provider,
model: request.model,
...assembler.replayState === undefined ? {} : { replayState: assembler.replayState },
},
},
...assembler.usage !== undefined ? { usage: assembler.usage } : {},
finish: assembler.finish,
}

View File

@@ -1,8 +1,8 @@
import { describe, expect, it } from 'vitest'
import { CallId } from '@deepseek-ai/dsh-llm'
import { CallId, 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 { mapStopReason, mapUsage, toPiContext, toStreamChunks } from '@deepseek-ai/dsh-llm-pi-ai'
import { mapStopReason, mapUsage, toPiContext, toPiReplayState, toStreamChunks } from '@deepseek-ai/dsh-llm-pi-ai'
function usage(input = 0, output = 0, cacheRead = 0, cacheWrite = 0): Usage {
return {
@@ -42,6 +42,7 @@ async function collect(stream: AsyncIterable<StreamChunk>): Promise<StreamChunk[
describe('toPiContext', () => {
it('maps system prompt, user text, and tools', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'deepseek-v4-flash',
system: 'be helpful',
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
@@ -55,13 +56,14 @@ describe('toPiContext', () => {
})
it('omits empty tools and absent system prompt', () => {
const context = toPiContext({ model: 'm', messages: [], tools: [] })
const context = toPiContext({ provider: 'deepseek', model: 'm', messages: [], tools: [] })
expect(context.systemPrompt).toBeUndefined()
expect(context.tools).toBeUndefined()
})
it('maps assistant text/reasoning/tool-call blocks', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
@@ -76,8 +78,7 @@ describe('toPiContext', () => {
expect(message.role).toBe('assistant')
expect(message.stopReason).toBe('toolUse')
expect(message.content).toEqual([
// thinkingSignature names the replay field — DeepSeek's passback rule.
{ type: 'thinking', thinking: 'hmm', thinkingSignature: 'reasoning_content' },
{ type: 'thinking', thinking: 'hmm' },
{ type: 'text', text: 'calling' },
{ type: 'toolCall', id: 'c1', name: 'f', arguments: { a: 1 } },
])
@@ -85,6 +86,7 @@ describe('toPiContext', () => {
it('marks tool-call-free assistant messages with stopReason stop', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{ role: 'assistant', content: [{ type: 'text', text: 'done' }] }],
})
@@ -93,6 +95,7 @@ describe('toPiContext', () => {
it('parses malformed tool-call arguments to {}', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
@@ -105,6 +108,7 @@ describe('toPiContext', () => {
it('parses non-object argument JSON (arrays, scalars) to {}', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
@@ -116,6 +120,7 @@ describe('toPiContext', () => {
it('recovers toolName for tool results from the preceding assistant call', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [
{
@@ -140,6 +145,7 @@ describe('toPiContext', () => {
it('labels unmatched tool results with toolName unknown and keeps isError', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'user',
@@ -156,6 +162,7 @@ describe('toPiContext', () => {
it('splits mixed user text + tool results and folds history system messages', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [
{ role: 'system', content: [{ type: 'text', text: 'rule' }] },
@@ -173,6 +180,7 @@ describe('toPiContext', () => {
it('skips plugin-added (unknown) blocks in assistant content', () => {
const context = toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
@@ -184,6 +192,195 @@ describe('toPiContext', () => {
})
expect((context.messages[0] as AssistantMessage).content).toEqual([{ type: 'text', text: 'visible' }])
})
it('recombines durable content with pi-ai replay metadata across target providers and models', () => {
const state = toPiReplayState(assistant({
api: 'openai-responses',
provider: 'openai',
model: 'gpt-5',
responseModel: 'gpt-5-2026-01-01',
responseId: 'resp_123',
stopReason: 'toolUse',
content: [
{ type: 'thinking', thinking: 'private reasoning', thinkingSignature: 'think-sig', redacted: true },
{ type: 'text', text: 'calling', textSignature: 'text-sig' },
{ type: 'toolCall', id: 'c1', name: 'f', arguments: { a: 1 }, thoughtSignature: 'tool-sig' },
],
}))
const context = toPiContext({
provider: 'anthropic',
model: 'claude-next',
messages: [{
role: 'assistant',
content: [
{ type: 'reasoning', text: 'private reasoning' },
{ type: 'text', text: 'calling' },
{ type: 'tool-call', id: CallId('c1'), name: 'f', arguments: '{"a":1}' },
],
provenance: { provider: 'openai', model: 'gpt-5', replayState: state },
}],
})
expect(context.messages[0]).toMatchObject({
role: 'assistant',
api: 'openai-responses',
provider: 'openai',
model: 'gpt-5',
responseModel: 'gpt-5-2026-01-01',
responseId: 'resp_123',
stopReason: 'toolUse',
content: [
{ type: 'thinking', thinking: 'private reasoning', thinkingSignature: 'think-sig', redacted: true },
{ type: 'text', text: 'calling', textSignature: 'text-sig' },
{ type: 'toolCall', id: 'c1', name: 'f', arguments: { a: 1 }, thoughtSignature: 'tool-sig' },
],
})
})
it('replays all native block kinds when optional metadata is absent', () => {
const state = toPiReplayState(assistant({
content: [
{ type: 'thinking', thinking: 'private reasoning' },
{ type: 'text', text: 'calling' },
{ type: 'toolCall', id: 'c1', name: 'f', arguments: { a: 1 } },
],
}))
const context = toPiContext({
provider: 'deepseek',
model: 'new-model',
messages: [{
role: 'assistant',
content: [
{ type: 'reasoning', text: 'private reasoning' },
{ type: 'text', text: 'calling' },
{ type: 'tool-call', id: CallId('c1'), name: 'f', arguments: '{"a":1}' },
],
provenance: { provider: 'deepseek', model: 'deepseek-v4-flash', replayState: state },
}],
})
expect(context.messages[0]).toMatchObject({
role: 'assistant',
content: [
{ type: 'thinking', thinking: 'private reasoning' },
{ type: 'text', text: 'calling' },
{ type: 'toolCall', id: 'c1', name: 'f', arguments: { a: 1 } },
],
})
expect(context.messages[0]).not.toHaveProperty('responseModel')
expect(context.messages[0]).not.toHaveProperty('responseId')
})
it('rejects unsupported replay-state versions with a stable error code', () => {
try {
toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
content: [{ type: 'text', text: 'done' }],
provenance: {
provider: 'deepseek',
model: 'old',
replayState: { kind: 'pi-ai', version: 2 },
},
}],
})
expect.fail('expected invalid replay state')
} catch (error: unknown) {
expect(error).toBeInstanceOf(LlmError)
expect((error as LlmError).code).toBe('INVALID_REPLAY_STATE')
expect((error as Error).message).toContain('unsupported version 2')
}
})
it('rejects replay metadata whose blocks do not match the durable content', () => {
const state = toPiReplayState(assistant({ content: [{ type: 'text', text: 'done' }] }))
expect(() => toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
content: [{ type: 'reasoning', text: 'done' }],
provenance: { provider: 'deepseek', model: 'deepseek-v4-flash', replayState: state },
}],
})).toThrow(/block 0 does not match assistant content/)
})
it('rejects replay metadata whose block count differs from durable content', () => {
const state = toPiReplayState(assistant())
expect(() => toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
content: [{ type: 'text', text: 'done' }],
provenance: { provider: 'deepseek', model: 'deepseek-v4-flash', replayState: state },
}],
})).toThrow(/block count does not match assistant content/)
})
const validReplay = {
kind: 'pi-ai',
version: 1,
api: 'openai-completions',
provider: 'deepseek',
model: 'deepseek-v4-flash',
stopReason: 'stop',
blocks: [{ type: 'text' }],
}
it.each([
['provider', { ...validReplay, provider: 'openai' }],
['model', { ...validReplay, model: 'deepseek-v4-pro' }],
])('rejects replay metadata whose %s differs from assistant provenance', (field, replayState) => {
try {
toPiContext({
provider: 'deepseek',
model: 'next-model',
messages: [{
role: 'assistant',
content: [{ type: 'text', text: 'done' }],
provenance: { provider: 'deepseek', model: 'deepseek-v4-flash', replayState },
}],
})
expect.fail('expected invalid replay state')
} catch (error: unknown) {
expect(error).toBeInstanceOf(LlmError)
expect((error as LlmError).code).toBe('INVALID_REPLAY_STATE')
expect((error as Error).message).toContain(`${field} does not match assistant provenance`)
}
})
it.each([
['number state', 1, 'expected an object'],
['null state', null, 'expected an object'],
['array state', [], 'expected an object'],
['unknown kind', { ...validReplay, kind: 'other' }, 'unknown state kind'],
['non-string api', { ...validReplay, api: 1 }, 'api must be a non-empty string'],
['empty provider', { ...validReplay, provider: '' }, 'provider must be a non-empty string'],
['missing model', { ...validReplay, model: undefined }, 'model must be a non-empty string'],
['unknown stop reason', { ...validReplay, stopReason: 'pause' }, 'unknown stopReason'],
['non-string response model', { ...validReplay, responseModel: 1 }, 'responseModel must be a string'],
['non-string response id', { ...validReplay, responseId: 1 }, 'responseId must be a string'],
['non-array blocks', { ...validReplay, blocks: 'text' }, 'blocks must be an array'],
['number block', { ...validReplay, blocks: [1] }, 'block 0 must be an object'],
['null block', { ...validReplay, blocks: [null] }, 'block 0 must be an object'],
['array block', { ...validReplay, blocks: [[]] }, 'block 0 must be an object'],
['unknown block type', { ...validReplay, blocks: [{ type: 'audio' }] }, 'block 0 has an unknown type'],
['non-string signature', { ...validReplay, blocks: [{ type: 'text', textSignature: 1 }] }, 'textSignature must be a string'],
['non-boolean redaction', { ...validReplay, blocks: [{ type: 'reasoning', redacted: 'yes' }] }, 'redacted must be boolean'],
])('rejects malformed replay state: %s', (_name, replayState, message) => {
expect(() => toPiContext({
provider: 'deepseek',
model: 'm',
messages: [{
role: 'assistant',
content: [{ type: 'text', text: 'done' }],
provenance: { provider: 'deepseek', model: 'deepseek-v4-flash', replayState },
}],
})).toThrow(message)
})
})
describe('toStreamChunks', () => {
@@ -205,7 +402,19 @@ describe('toStreamChunks', () => {
{ type: 'text-delta', index: 0, text: 'hi' },
{ type: 'block-end', index: 0, block: { type: 'text', text: 'hi' } },
{ type: 'usage', usage: { inputTokens: 3, outputTokens: 2 } },
{ type: 'finish', reason: { kind: 'stop' } },
{
type: 'finish',
reason: { kind: 'stop' },
replayState: {
kind: 'pi-ai',
version: 1,
api: 'openai-completions',
provider: 'deepseek',
model: 'deepseek-v4-flash',
stopReason: 'stop',
blocks: [{ type: 'text' }],
},
},
])
})
@@ -234,7 +443,7 @@ describe('toStreamChunks', () => {
toolCall: { type: 'toolCall', id: 'call-1', name: 'f', arguments: { a: 1 } },
partial: partialWithToolCall,
},
{ type: 'done', reason: 'toolUse', message: assistant({ stopReason: 'toolUse' }) },
{ type: 'done', reason: 'toolUse', message: assistant({ content: partialWithToolCall.content, stopReason: 'toolUse' }) },
)))
expect(chunks).toEqual([
{ type: 'block-start', index: 0, blockType: 'tool-call' },
@@ -242,7 +451,19 @@ describe('toStreamChunks', () => {
{ type: 'tool-call-delta', index: 0, id: 'call-1', name: 'f', argumentsDelta: ':1}' },
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: 'call-1', name: 'f', arguments: '{"a":1}' } },
{ type: 'usage', usage: { inputTokens: 0, outputTokens: 0 } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
{
type: 'finish',
reason: { kind: 'tool-calls' },
replayState: {
kind: 'pi-ai',
version: 1,
api: 'openai-completions',
provider: 'deepseek',
model: 'deepseek-v4-flash',
stopReason: 'toolUse',
blocks: [{ type: 'tool-call' }],
},
},
])
})

View File

@@ -8,10 +8,13 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
### Public API
- `ctx.llm.registerAdapter(models: string[], adapter: LlmAdapter): () => void` Register an adapter for the given model names. Disposed with the calling fiber.
- `ctx.llm.models(): string[]` — model names with a registered adapter.
- `ctx.llm.registerAdapter(providers: string[], adapter: LlmAdapter): () => void` Register one adapter instance for the given provider routes. Registration is all-or-nothing, and is disposed with the calling fiber.
- `ctx.llm.listProviders(): LlmProviderInfo[]` Describe registered provider routes in registration order.
- `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`.
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
| Event | Mode | Purpose |
@@ -20,18 +23,18 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
### Extension points
- Subclass `LlmAdapter` and call `ctx.llm.registerAdapter(models, adapter)` to add a new model provider.
- Subclass `LlmAdapter` and call `ctx.llm.registerAdapter(providers, adapter)` to add one or more provider routes. `GenerateOptions.provider` selects the adapter; `GenerateOptions.model` is adapter-owned and may be resolved dynamically. Override `providerInfo()` and asynchronous `listModels()` to expose selector metadata; their defaults use the route id as its name and advertise no models.
- Wrap `llm/stream` via `ctx.on()` waterfall listeners for caching, retry, logging, rate-limiting, etc.
### Content-block vocabulary (`types.ts`)
Messages are arrays of typed content blocks: `text`, `reasoning`, `tool-call`, `tool-result`. The union is derived from the merge-extensible `ContentBlockMap`, so plugins can add block types via declaration merging. The core set is limited to blocks every shipping path honors — multimodal content (images, audio, …) has no core block type; a feature that needs one adds it via the map together with the adapter/UI/compaction support that honors it.
Messages are arrays of typed content blocks: `text`, `reasoning`, `tool-call`, `tool-result`. The union is derived from the merge-extensible `ContentBlockMap`, so plugins can add block types via declaration merging. Assistant messages produced by the loop also carry provider/model provenance and optional adapter-private replay state. Before dispatch, `LlmService` retains that state only when the historical provider route and target provider route are currently owned by the exact same adapter instance; the adapter then decides whether it can restore or convert the state across models/providers. The core block set is limited to blocks every shipping path honors — multimodal content (images, audio, …) has no core block type; a feature that needs one adds it via the map together with the adapter/UI/compaction support that honors it.
Streaming is a raw chunk protocol (`block-start`, `text-delta`, `reasoning-delta`, `tool-call-delta`, `block-end`, `usage`, `finish`). `BlockAssembler` is the single shared implementation that assembles chunks into blocks/messages.
### Call configuration (`call-config.ts`)
`LlmCallConfig` is the model + sampling scalars of one conversation's requests (`model`, `temperature`, `maxTokens`, `stop` — each mapping 1:1 onto the same-named `GenerateOptions` field). It is per-conversation state recorded in the session log as part of the request header (see the dsh-session `request/header` events), never a silently-adjustable per-call knob: the `agent/request` waterfall proposes a replacement and the loop logs a real change. `callConfigEquals(a, b)` is the field-wise real-change detector; `deepFreeze(value)` is the ownership helper the loop applies to every built request before dispatch (`llm/stream` listeners and adapters read, never rewrite).
`LlmCallConfig` is the provider + model + sampling scalars of one conversation's requests (`provider`, `model`, `temperature`, `maxTokens`, `stop` — each mapping 1:1 onto the same-named `GenerateOptions` field). It is per-conversation state recorded in the session log as part of the request header (see the dsh-session `request/header` events), never a silently-adjustable per-call knob: the `agent/request` waterfall proposes a replacement and the loop logs a real change. `callConfigEquals(a, b)` is the field-wise real-change detector; `deepFreeze(value)` is the ownership helper the loop applies to every built request before dispatch (`llm/stream` listeners and adapters read, never rewrite).
### App attribution (`attribution.ts`)
@@ -46,7 +49,7 @@ Every product adapter sends application identity on provider HTTP requests. `att
### Real adapters
Two adapters implement `LlmAdapter` on different internals: [`@deepseek-ai/dsh-llm-deepseek`](../llm-deepseek) uses hand-rolled fetch/SSE, while [`@deepseek-ai/dsh-llm-pi-ai`](../llm-pi-ai) uses `@earendil-works/pi-ai`. Both follow the `StreamChunk` conventions in `types.ts`: usage precedes finish, tool arguments remain raw strings, and errors take one of two sanctioned paths. See [the twin LLM adapters](../../../docs/rfc/implemented/architecture/2026-06-13-twin-llm-adapters.md) for the design rationale.
Two adapters implement `LlmAdapter` on different internals: [`@deepseek-ai/dsh-llm-deepseek`](../llm-deepseek) uses hand-rolled fetch/SSE for the `deepseek` route, while [`@deepseek-ai/dsh-llm-pi-ai`](../llm-pi-ai) dynamically resolves configured provider/model pairs through `@earendil-works/pi-ai`. Both follow the `StreamChunk` conventions in `types.ts`: usage precedes finish, tool arguments remain raw strings, and errors take one of two sanctioned paths. See [the twin LLM adapters](../../../docs/rfc/implemented/architecture/2026-06-13-twin-llm-adapters.md) for the design rationale.
## Model Experience

View File

@@ -36,6 +36,7 @@ export class BlockAssembler {
private order: number[] = []
private _usage: TokenUsage | undefined
private _finish: FinishReason | undefined
private _replayState: unknown = undefined
/**
* Feed one chunk. Returns the completed block when the chunk closes one
@@ -85,6 +86,7 @@ export class BlockAssembler {
}
case 'finish': {
this._finish = chunk.reason
this._replayState = chunk.replayState
return
}
default: return assertNever(chunk, 'BlockAssembler.push')
@@ -142,6 +144,11 @@ export class BlockAssembler {
return this._finish ?? { kind: 'stop' }
}
/** Adapter-private replay state from the terminal finish chunk, if any. */
get replayState(): unknown {
return this._replayState
}
/**
* The assembled assistant message.
* @returns an assistant-role message over `blocks()` (same open-block assembly rules).

View File

@@ -1,17 +1,18 @@
/**
* Conversation call configuration and freeze utilities. Model and sampling
* values are request-header state that can affect cache reuse; request
* waterfalls replace them and the loop logs changed snapshots instead of
* allowing silent per-call drift.
* Conversation call configuration and freeze utilities. Provider routing,
* model, and sampling values are request-header state that can affect cache
* reuse; request waterfalls replace them and the loop logs changed snapshots
* instead of allowing silent per-call drift.
* @module dsh-llm/call-config
*/
/**
* Model + sampling scalars of one conversation's requests. Every field maps
* Provider + model + sampling scalars of one conversation's requests. Every field maps
* 1:1 onto the same-named `GenerateOptions` field; the loop builds requests
* from the logged header rather than accepting these per call.
*/
export interface LlmCallConfig {
provider: string
model: string
temperature?: number
maxTokens?: number
@@ -27,7 +28,7 @@ export interface LlmCallConfig {
* @returns whether every field (including the `stop` list, element-wise) matches.
*/
export function callConfigEquals(a: LlmCallConfig, b: LlmCallConfig): boolean {
if (a.model !== b.model || a.temperature !== b.temperature || a.maxTokens !== b.maxTokens) return false
if (a.provider !== b.provider || a.model !== b.model || a.temperature !== b.temperature || a.maxTokens !== b.maxTokens) return false
if (a.stop === undefined || b.stop === undefined) return a.stop === b.stop
return a.stop.length === b.stop.length && a.stop.every((s, i) => s === b.stop?.[i])
}

View File

@@ -7,8 +7,9 @@
*/
import { Context, Service } from 'cordis'
import type { GenerateOptions, StreamChunk } from './types.ts'
import type { GenerateOptions, LlmModelInfo, LlmProviderInfo, Message, StreamChunk } from './types.ts'
import { HarnessError } from './error.ts'
import { deepFreeze } from './call-config.ts'
export * from './attribution.ts'
export * from './brand.ts'
@@ -55,11 +56,31 @@ export class LlmError extends HarnessError {
/**
* Provider-wire adapter for the harness message and stream vocabulary. Register implementations
* with `ctx.llm.registerAdapter(models, adapter)`. Every provider HTTP request must include
* with `ctx.llm.registerAdapter(providers, adapter)`. Every provider HTTP request must include
* `attributionHeaders()`; prove that at the wire or library header-hook boundary. The hand-rolled
* DeepSeek and pi-ai adapters intentionally exercise this contract through different internals.
*/
export abstract class LlmAdapter {
/**
* Describe one provider route owned by this adapter.
* @param provider - a route passed to `registerAdapter()` for this instance.
* @returns detached display metadata whose id must equal `provider`.
*/
providerInfo(provider: string): LlmProviderInfo {
return { id: provider, name: provider }
}
/**
* List models this adapter can currently advertise for one owned provider.
* The result is advisory: an adapter may accept unlisted model ids, and
* consumers must not turn absence into request rejection.
* @param _provider - one provider route owned by this adapter.
* @returns discoverable models in adapter-preferred order.
*/
listModels(_provider: string): Promise<readonly LlmModelInfo[]> {
return Promise.resolve([])
}
/**
* Stream one model call as raw chunks. The only required method.
* @param options - the fully-assembled request; implementations must honor `options.signal`.
@@ -73,30 +94,40 @@ export abstract class LlmAdapter {
* surface, interceptable via the `llm/stream` waterfall.
*/
export class LlmService extends Service {
private adapters = new Map<string, LlmAdapter>()
private adapters = new Map<string, { adapter: LlmAdapter; provider: LlmProviderInfo }>()
constructor(ctx: Context) {
super(ctx, 'llm')
}
/**
* Register an adapter for the given model names. Throws `LlmError` with code
* `DUPLICATE_ADAPTER` if any model already has an adapter (all-or-nothing).
* Register an adapter for the given provider routes. Throws `LlmError` with code
* `DUPLICATE_ADAPTER` if any provider already has an adapter (all-or-nothing).
* Disposed with the fiber.
* @param models - every model name this adapter should serve.
* @param adapter - the adapter that streams calls for those models.
* @param providers - every provider route this adapter should serve.
* @param adapter - the adapter that streams calls for those providers.
* @returns the disposer that unregisters all of them.
*/
registerAdapter(models: string[], adapter: LlmAdapter): () => void {
registerAdapter(providers: string[], adapter: LlmAdapter): () => void {
const dispose = this.ctx.effect(function* (this: LlmService) {
for (const model of models) {
if (this.adapters.has(model)) {
throw new LlmError(`an adapter for model "${model}" is already registered`, 'DUPLICATE_ADAPTER')
if (providers.length === 0) throw new LlmError('an adapter must register at least one provider', 'INVALID_ADAPTER')
const unique = new Set<string>()
const registrations: { adapter: LlmAdapter; provider: LlmProviderInfo }[] = []
for (const provider of providers) {
if (provider.length === 0) throw new LlmError('adapter provider names must be non-empty', 'INVALID_ADAPTER')
if (unique.has(provider) || this.adapters.has(provider)) {
throw new LlmError(`an adapter for provider "${provider}" is already registered`, 'DUPLICATE_ADAPTER')
}
const info = adapter.providerInfo(provider)
if (typeof info.id !== 'string' || info.id !== provider || typeof info.name !== 'string' || info.name.length === 0) {
throw new LlmError(`adapter metadata for provider "${provider}" must preserve its id and have a non-empty name`, 'INVALID_ADAPTER')
}
unique.add(provider)
registrations.push({ adapter, provider: { id: info.id, name: info.name } })
}
for (const model of models) this.adapters.set(model, adapter)
for (const registration of registrations) this.adapters.set(registration.provider.id, registration)
yield () => {
for (const model of models) this.adapters.delete(model)
for (const provider of providers) this.adapters.delete(provider)
}
}.bind(this), 'llm.registerAdapter()')
// ctx.effect's disposer returns Promise<void>; our disposer API is
@@ -105,29 +136,81 @@ export class LlmService extends Service {
}
/**
* Model names with a registered adapter.
* @returns the registered names, in registration order.
* Describe provider routes with a registered adapter.
* @returns detached provider metadata in registration order.
*/
models(): string[] {
return [...this.adapters.keys()]
listProviders(): LlmProviderInfo[] {
return [...this.adapters.values()].map(({ provider }) => ({ ...provider }))
}
private adapter(model: string): LlmAdapter {
const adapter = this.adapters.get(model)
if (!adapter) throw new LlmError(`no adapter registered for model "${model}"`, 'NO_ADAPTER')
return adapter
/**
* Discover models advertised by one registered provider. Catalog membership
* is advisory and never changes routing or request validation.
* @param provider - registered provider route to inspect.
* @returns detached model metadata in adapter-preferred order.
*/
async listModels(provider: string): Promise<LlmModelInfo[]> {
const adapter = this.registration(provider).adapter
const models = await adapter.listModels(provider)
const seen = new Set<string>()
return models.map((model) => {
if (
typeof model.provider !== 'string'
|| model.provider !== provider
|| typeof model.id !== 'string'
|| model.id.length === 0
|| typeof model.name !== 'string'
|| model.name.length === 0
|| (model.description !== undefined && typeof model.description !== 'string')
|| seen.has(model.id)
) {
throw new LlmError(`adapter returned invalid or duplicate model metadata for provider "${provider}"`, 'INVALID_CATALOG')
}
seen.add(model.id)
return {
provider: model.provider,
id: model.id,
name: model.name,
...model.description === undefined ? {} : { description: model.description },
}
})
}
private registration(provider: string): { adapter: LlmAdapter; provider: LlmProviderInfo } {
const registration = this.adapters.get(provider)
if (!registration) throw new LlmError(`no adapter registered for provider "${provider}"`, 'NO_ADAPTER')
return registration
}
/** Remove replay state whose historical route is owned by another adapter. */
private forAdapter(options: GenerateOptions, adapter: LlmAdapter): GenerateOptions {
const messages: Message[] = options.messages.map((message) => {
const provenance = message.provenance
if (message.role !== 'assistant' || provenance?.replayState === undefined) return message
if (this.adapters.get(provenance.provider)?.adapter === adapter) return message
return {
...message,
provenance: { provider: provenance.provider, model: provenance.model },
}
})
if (messages.every((message, index) => message === options.messages[index])) return options
const filtered = { ...options, messages }
return Object.isFrozen(options) ? deepFreeze(filtered) : filtered
}
/**
* Stream one model call as raw chunks (token-level deltas). Throws
* `LlmError` with code `NO_ADAPTER` if no adapter is registered for
* `options.model`. Dispatches through the `llm/stream` waterfall.
* @param options - the full request; `options.model` selects the adapter.
* `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.
* @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, () => {
return this.adapter(options.model).stream(options)
const adapter = this.registration(options.provider).adapter
return adapter.stream(this.forAdapter(options, adapter))
})
}
}

View File

@@ -53,10 +53,29 @@ export type ContentBlockType = keyof ContentBlockMap
/** Any known content block, derived from {@link ContentBlockMap}; switch on `type` and fall through unknowns (merge-extensible). */
export type ContentBlock = ContentBlockMap[ContentBlockType]
/** A single message in a conversation history. */
/** Provider ownership and adapter-private replay data for an assistant message. */
export interface AssistantProvenance {
/** Provider route that produced the message. */
provider: string
/** Provider model id that produced the message. */
model: string
/**
* Lossless-JSON adapter state needed to replay the provider response.
* `LlmService` exposes it to a target adapter only when that adapter instance
* currently owns both this historical provider and the target provider.
*/
replayState?: unknown
}
/**
* A single message in a conversation history. Loop-derived assistant messages
* always carry provenance; callers may omit it on hand-built foreign history.
*/
export interface Message {
role: 'system' | 'user' | 'assistant'
content: ContentBlock[]
/** Present only on assistant messages produced by a routed adapter. */
provenance?: AssistantProvenance
}
/**
@@ -102,6 +121,26 @@ export interface TokenUsage {
reasoningTokens?: number
}
/** Display metadata for one registered provider route. */
export interface LlmProviderInfo {
/** Provider route key used by {@link GenerateOptions.provider}. */
id: string
/** Human-readable provider name for selectors and diagnostics. */
name: string
}
/** One adapter-discovered model; catalog membership is advisory, not request validation. */
export interface LlmModelInfo {
/** Provider route that owns this model entry. */
provider: string
/** Model id passed to {@link GenerateOptions.model}. */
id: string
/** Human-readable model name for selectors. */
name: string
/** Optional user-facing distinction from otherwise similar models. */
description?: string
}
/**
* Raw streaming protocol emitted by adapters.
* Block indexes correlate interleaved deltas, and `block-end` carries the
@@ -116,7 +155,12 @@ export type StreamChunk =
| { type: 'tool-call-delta'; index: number; id: CallId; name?: string; argumentsDelta: string }
| { type: 'block-end'; index: number; block: ContentBlock }
| { type: 'usage'; usage: TokenUsage }
| { type: 'finish'; reason: FinishReason }
| {
type: 'finish'
reason: FinishReason
/** Adapter-private lossless-JSON state for replaying a successful response. */
replayState?: unknown
}
/**
* JSON-schema description of a tool, as sent to the model.
@@ -134,6 +178,8 @@ export interface ToolSchema {
/** A single model request, fully assembled. */
export interface GenerateOptions {
/** Registered provider route selecting the adapter instance. */
provider: string
model: string
/**
* Ordered conversation messages, exactly as the provider sees them (after

View File

@@ -9,14 +9,16 @@ import { callConfigEquals, deepFreeze } from '../src/call-config.ts'
describe('callConfigEquals', () => {
it('compares every field, including the stop list element-wise', () => {
expect(callConfigEquals({ model: 'm' }, { model: 'm' })).toBe(true)
expect(callConfigEquals({ model: 'm' }, { model: 'x' })).toBe(false)
expect(callConfigEquals({ model: 'm', temperature: 0.5 }, { model: 'm' })).toBe(false)
expect(callConfigEquals({ model: 'm', maxTokens: 1 }, { model: 'm', maxTokens: 2 })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a'] }, { model: 'm' })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a'] }, { model: 'm', stop: ['a', 'b'] })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a'] }, { model: 'm', stop: ['b'] })).toBe(false)
expect(callConfigEquals({ model: 'm', stop: ['a', 'b'] }, { model: 'm', stop: ['a', 'b'] })).toBe(true)
const base = { provider: 'p', model: 'm' }
expect(callConfigEquals(base, base)).toBe(true)
expect(callConfigEquals(base, { provider: 'x', model: 'm' })).toBe(false)
expect(callConfigEquals(base, { provider: 'p', model: 'x' })).toBe(false)
expect(callConfigEquals({ ...base, temperature: 0.5 }, base)).toBe(false)
expect(callConfigEquals({ ...base, maxTokens: 1 }, { ...base, maxTokens: 2 })).toBe(false)
expect(callConfigEquals({ ...base, stop: ['a'] }, base)).toBe(false)
expect(callConfigEquals({ ...base, stop: ['a'] }, { ...base, stop: ['a', 'b'] })).toBe(false)
expect(callConfigEquals({ ...base, stop: ['a'] }, { ...base, stop: ['b'] })).toBe(false)
expect(callConfigEquals({ ...base, stop: ['a', 'b'] }, { ...base, stop: ['a', 'b'] })).toBe(true)
})
})

View File

@@ -1,6 +1,7 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { GenerateOptions, LlmAdapter, LlmError, StreamChunk } from '@deepseek-ai/dsh-llm'
import type { LlmModelInfo, LlmProviderInfo } from '@deepseek-ai/dsh-llm'
class ScriptedAdapter extends LlmAdapter {
constructor(private script: StreamChunk[]) {
@@ -12,6 +13,32 @@ class ScriptedAdapter extends LlmAdapter {
}
}
class RecordingAdapter extends ScriptedAdapter {
lastOptions: GenerateOptions | undefined
override async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
this.lastOptions = options
yield * super.stream(options)
}
}
class CatalogAdapter extends ScriptedAdapter {
constructor(
private readonly provider: LlmProviderInfo,
private readonly models: readonly LlmModelInfo[],
) {
super(SCRIPT)
}
override providerInfo(_provider: string): LlmProviderInfo {
return this.provider
}
override listModels(_provider: string): Promise<readonly LlmModelInfo[]> {
return Promise.resolve(this.models)
}
}
const SCRIPT: StreamChunk[] = [
{ type: 'block-start', index: 0, blockType: 'text' },
{ type: 'text-delta', index: 0, text: 'hi' },
@@ -22,18 +49,18 @@ describe('LlmService', () => {
it('routes stream() to the registered adapter', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], new ScriptedAdapter(SCRIPT))
ctx.llm.registerAdapter(['test-provider'], new ScriptedAdapter(SCRIPT))
const chunks: StreamChunk[] = []
for await (const chunk of ctx.llm.stream({ model: 'test-model', messages: [] })) chunks.push(chunk)
for await (const chunk of ctx.llm.stream({ provider: 'test-provider', model: 'test-model', messages: [] })) chunks.push(chunk)
expect(chunks).toEqual(SCRIPT)
})
it('throws NO_ADAPTER for unregistered models', async () => {
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({ model: 'nope', messages: [] })) { /* drain */ }
for await (const _ of ctx.llm.stream({ provider: 'nope', model: 'any-model', messages: [] })) { /* drain */ }
})()).rejects.toThrow('no adapter registered')
})
@@ -44,10 +71,80 @@ describe('LlmService', () => {
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
inner.llm.registerAdapter(['scoped-model'], new ScriptedAdapter(SCRIPT))
}, { inject: ['llm'] }))
expect(ctx.llm.models()).toEqual(['scoped-model'])
expect(ctx.llm.listProviders()).toEqual([{ id: 'scoped-model', name: 'scoped-model' }])
await fiber.dispose()
expect(ctx.llm.models()).toEqual([])
expect(ctx.llm.listProviders()).toEqual([])
})
it('discovers detached provider and advisory model metadata', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const provider = { id: 'catalog', name: 'Catalog Provider' }
const model = { provider: 'catalog', id: 'fast', name: 'Fast', description: 'Low latency' }
ctx.llm.registerAdapter(['catalog'], new CatalogAdapter(provider, [model]))
const providers = ctx.llm.listProviders()
const models = await ctx.llm.listModels('catalog')
expect(providers).toEqual([provider])
expect(models).toEqual([model])
providers[0]!.name = 'mutated'
models[0]!.name = 'mutated'
provider.name = 'source mutated'
model.name = 'source mutated'
expect(ctx.llm.listProviders()).toEqual([{ id: 'catalog', name: 'Catalog Provider' }])
await expect(ctx.llm.listModels('catalog')).resolves.toEqual([{
provider: 'catalog', id: 'fast', name: 'source mutated', description: 'Low latency',
}])
})
it('defaults adapters to their route name and an empty advisory model list', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['plain'], new ScriptedAdapter(SCRIPT))
expect(ctx.llm.listProviders()).toEqual([{ id: 'plain', name: 'plain' }])
await expect(ctx.llm.listModels('plain')).resolves.toEqual([])
await expect(ctx.llm.listModels('missing')).rejects.toMatchObject({ code: 'NO_ADAPTER' })
})
it.each([
[{ id: 1, name: 'Name' }, 'non-string id'],
[{ id: 'other', name: 'Name' }, 'mismatched id'],
[{ id: 'route', name: 1 }, 'non-string name'],
[{ id: 'route', name: '' }, 'empty name'],
] as const)('rejects invalid provider metadata atomically (%s: %s)', async (metadata, _label) => {
const ctx = new Context()
await ctx.plugin(LlmService)
const adapter = new CatalogAdapter(metadata as unknown as LlmProviderInfo, [])
expect(() => ctx.llm.registerAdapter(['route'], adapter)).toThrow(expect.objectContaining({ code: 'INVALID_ADAPTER' }))
expect(ctx.llm.listProviders()).toEqual([])
})
it.each([
[{ provider: 1, id: 'm', name: 'M' }, 'non-string provider'],
[{ provider: 'other', id: 'm', name: 'M' }, 'mismatched provider'],
[{ provider: 'route', id: 1, name: 'M' }, 'non-string id'],
[{ provider: 'route', id: '', name: 'M' }, 'empty id'],
[{ provider: 'route', id: 'm', name: 1 }, 'non-string name'],
[{ provider: 'route', id: 'm', name: '' }, 'empty name'],
[{ provider: 'route', id: 'm', name: 'M', description: 1 }, 'non-string description'],
] as const)('rejects invalid model metadata (%s: %s)', async (metadata, _label) => {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['route'], new CatalogAdapter(
{ id: 'route', name: 'Route' },
[metadata as unknown as LlmModelInfo],
))
await expect(ctx.llm.listModels('route')).rejects.toMatchObject({ code: 'INVALID_CATALOG' })
})
it('rejects duplicate model ids in one provider catalog', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const model = { provider: 'route', id: 'same', name: 'Same' }
ctx.llm.registerAdapter(['route'], new CatalogAdapter({ id: 'route', name: 'Route' }, [model, model]))
await expect(ctx.llm.listModels('route')).rejects.toMatchObject({ code: 'INVALID_CATALOG' })
})
it('lets llm/stream waterfall listeners wrap the underlying stream', async () => {
@@ -64,11 +161,90 @@ describe('LlmService', () => {
})
const chunks: StreamChunk[] = []
for await (const chunk of ctx.llm.stream({ model: 'test-model', messages: [] })) chunks.push(chunk)
for await (const chunk of ctx.llm.stream({ provider: 'test-model', model: 'dynamic-model', messages: [] })) chunks.push(chunk)
expect(chunks).toHaveLength(4)
expect(chunks[0]).toMatchObject({ index: 99 })
})
it('resolves the provider after llm/stream listeners have had a chance to route it', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const adapter = new RecordingAdapter(SCRIPT)
ctx.llm.registerAdapter(['routed'], adapter)
ctx.on('llm/stream', (options, next) => {
options.provider = 'routed'
return next()
})
for await (const _chunk of ctx.llm.stream({ provider: 'initial', model: 'm', messages: [] })) { /* drain */ }
expect(adapter.lastOptions?.provider).toBe('routed')
})
it('keeps replay state when historical and target providers belong to the same adapter instance', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const adapter = new RecordingAdapter(SCRIPT)
ctx.llm.registerAdapter(['historical', 'target'], adapter)
const replayState = { private: 'state' }
for await (const _chunk of ctx.llm.stream({
provider: 'target',
model: 'new-model',
messages: [{
role: 'assistant',
content: [{ type: 'text', text: 'old response' }],
provenance: { provider: 'historical', model: 'old-model', replayState },
}],
})) { /* drain */ }
expect(adapter.lastOptions?.messages[0]?.provenance).toEqual({
provider: 'historical', model: 'old-model', replayState,
})
})
it('strips replay state but preserves provenance when the target uses a different adapter instance', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['historical'], new RecordingAdapter(SCRIPT))
const target = new RecordingAdapter(SCRIPT)
ctx.llm.registerAdapter(['target'], target)
for await (const _chunk of ctx.llm.stream({
provider: 'target',
model: 'new-model',
messages: [{
role: 'assistant',
content: [{ type: 'text', text: 'old response' }],
provenance: { provider: 'historical', model: 'old-model', replayState: { private: 'state' } },
}],
})) { /* drain */ }
expect(target.lastOptions?.messages[0]?.provenance).toEqual({ provider: 'historical', model: 'old-model' })
})
it('preserves immutability while stripping replay state from frozen requests', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['historical'], new RecordingAdapter(SCRIPT))
const target = new RecordingAdapter(SCRIPT)
ctx.llm.registerAdapter(['target'], target)
const options = Object.freeze({
provider: 'target',
model: 'new-model',
messages: [{
role: 'assistant' as const,
content: [{ type: 'text' as const, text: 'old response' }],
provenance: { provider: 'historical', model: 'old-model', replayState: { private: 'state' } },
}],
})
for await (const _chunk of ctx.llm.stream(options)) { /* drain */ }
expect(target.lastOptions).not.toBe(options)
expect(Object.isFrozen(target.lastOptions)).toBe(true)
expect(target.lastOptions?.messages[0]?.provenance).toEqual({ provider: 'historical', model: 'old-model' })
})
it('creates LlmError with a code for programmatic handling', () => {
const err = new LlmError('something went wrong', 'CUSTOM_CODE')
expect(err).toBeInstanceOf(Error)
@@ -104,9 +280,9 @@ describe('LlmService', () => {
await ctx.plugin(LlmService)
const dispose = ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))
expect(ctx.llm.models()).toEqual(['m1'])
expect(ctx.llm.listProviders()).toEqual([{ id: 'm1', name: 'm1' }])
dispose()
expect(ctx.llm.models()).toEqual([])
expect(ctx.llm.listProviders()).toEqual([])
})
it('rejects duplicate adapter registration with DUPLICATE_ADAPTER code', async () => {
@@ -123,19 +299,30 @@ describe('LlmService', () => {
}
})
it('rejects empty and internally duplicated provider registrations atomically', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const adapter = new ScriptedAdapter(SCRIPT)
expect(() => ctx.llm.registerAdapter([], adapter)).toThrow(expect.objectContaining({ code: 'INVALID_ADAPTER' }))
expect(() => ctx.llm.registerAdapter([''], adapter)).toThrow(expect.objectContaining({ code: 'INVALID_ADAPTER' }))
expect(() => ctx.llm.registerAdapter(['first', 'first'], adapter)).toThrow(expect.objectContaining({ code: 'DUPLICATE_ADAPTER' }))
expect(ctx.llm.listProviders()).toEqual([])
})
it('re-registers a model after its prior registration is disposed', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const dispose = ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))
expect(ctx.llm.models()).toEqual(['m1'])
expect(ctx.llm.listProviders()).toEqual([{ id: 'm1', name: 'm1' }])
dispose()
expect(ctx.llm.models()).toEqual([])
expect(ctx.llm.listProviders()).toEqual([])
// The duplicate check is not wedged: the same model registers cleanly again.
const disposeAgain = ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))
expect(ctx.llm.models()).toEqual(['m1'])
expect(ctx.llm.listProviders()).toEqual([{ id: 'm1', name: 'm1' }])
disposeAgain()
expect(ctx.llm.models()).toEqual([])
expect(ctx.llm.listProviders()).toEqual([])
})
})