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

Merge remote app attribution branch

Browse Source 
Resolve the RFC and implementation to defer OpenRouter-specific attribution headers and keep mandatory attribution to User-Agent only.
This commit is contained in:
Tianyi Cui
2026-07-04 22:47:53 +08:00
parent 1891cb52d5 ac42c46a65
commit 544cc0439e
692 changed files with 49021 additions and 4970 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/llm/llm-deepseek/README.md
View File

@@ -23,6 +23,10 @@ A second, independent implementation of the same seam exists in `@deepseek-ai/ds
`thinking`/`reasoningEffort` are adapter-level request defaults serialized as the official top-level `thinking: {type}` / `reasoning_effort` wire fields. They live in adapter config (not `GenerateOptions`) to keep the core vocabulary provider-neutral.
## App attribution
Every request carries the shared attribution header from dsh-llm's `attributionHeaders()` - the mandatory `User-Agent` baseline identifying the harness (see [dsh-llm § App attribution](../llm/README.md#app-attribution-attributionts)). Direct DeepSeek requests and OpenAI-compatible gateway requests get no provider-specific app-attribution headers under this adapter contract; OpenRouter app attribution is deferred to a future explicit OpenRouter adapter or mode.
## Wire-format notes (verified live + against the official docs)
- Streaming only (`stream_options.include_usage` always on). `usage` may arrive attached to the finish chunk or as a trailing usage-only chunk — the translator defers both to `[DONE]`, so `usage` always precedes `finish` and nothing follows `finish`.

8
packages/llm/llm-deepseek/package.json
View File

@@ -5,17 +5,19 @@
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/index.d.ts",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/index.d.ts",
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib",
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",

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

@@ -5,7 +5,7 @@
* @module dsh-llm-deepseek/adapter
*/
import { LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import { attributionHeaders, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import { serializeRequest } from './serialize.ts'
import type { RequestDefaults } from './serialize.ts'
@@ -21,13 +21,6 @@ export interface DeepSeekAdapterOptions {
defaults?: RequestDefaults
}
/**
* Attribution header sent on every request so the provider can identify the
* client. Bump in lockstep with this package's version (no build-time version
* injection is wired in this repo yet).
*/
const USER_AGENT = 'deepseek-harness/0.0.1'
/** Map an HTTP status to a stable LlmError code. */
export function httpErrorCode(status: number): string {
if (status === 401 || status === 403) return 'AUTH'
@@ -67,7 +60,7 @@ export class DeepSeekAdapter extends LlmAdapter {
'authorization': `Bearer ${this.options.apiKey}`,
'content-type': 'application/json',
'accept': 'text/event-stream',
'user-agent': USER_AGENT,
...attributionHeaders(),
},
body: JSON.stringify(body),
...options.signal ? { signal: options.signal } : {},

15
packages/llm/llm-deepseek/tests/adapter.e2e.ts
View File

@@ -1,9 +1,10 @@
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
import type { GenerateResult, Message, ToolSchema } from '@deepseek-ai/dsh-llm'
import type { Message, ToolSchema } from '@deepseek-ai/dsh-llm'
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
import type { Config } from '@deepseek-ai/dsh-llm-deepseek'
import { assemble, type AssembledResult } from './assemble.ts'
/**
* Real-API e2e for the hand-rolled adapter: V4 Flash + V4 Pro across
@@ -31,7 +32,7 @@ function ask(text: string): Message[] {
return [{ role: 'user', content: [{ type: 'text', text }] }]
}
function textOf(result: GenerateResult): string {
function textOf(result: AssembledResult): string {
return result.message.content
.filter(block => block.type === 'text')
.map(block => block.text)
@@ -51,7 +52,7 @@ const weatherTool: ToolSchema = {
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', () => {
it('flash + thinking disabled: plain text generation', async () => {
const ctx = await harness(FLASH, { thinking: 'disabled' })
const result = await ctx.llm.generate({
const result = await assemble(ctx,{
model: FLASH,
messages: ask('Reply with exactly the word: pong'),
maxTokens: 50,
@@ -65,7 +66,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', ()
it('flash + thinking enabled (effort high): reasoning blocks + reasoning tokens', async () => {
const ctx = await harness(FLASH, { thinking: 'enabled', reasoningEffort: 'high' })
const result = await ctx.llm.generate({
const result = await assemble(ctx,{
model: FLASH,
messages: ask('Which is larger, 9.11 or 9.8? Answer with just the number.'),
maxTokens: 2000,
@@ -82,7 +83,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', ()
const ctx = await harness(PRO, { thinking: 'enabled', reasoningEffort: effort })
// Turn 1: the model must call the tool (and think before it).
const first = await ctx.llm.generate({
const first = await assemble(ctx,{
model: PRO,
messages: ask('What is the weather in Paris right now? Use the get_weather tool.'),
tools: [weatherTool],
@@ -96,7 +97,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', ()
// Turn 2: send the tool result back WITH the assistant's reasoning
// block in history (the official thinking+tools passback rule).
const second = await ctx.llm.generate({
const second = await assemble(ctx,{
model: PRO,
messages: [
...ask('What is the weather in Paris right now? Use the get_weather tool.'),
@@ -120,7 +121,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-deepseek e2e (real API)', ()
it('pro + thinking disabled: plain generation without reasoning blocks', async () => {
const ctx = await harness(PRO, { thinking: 'disabled' })
const result = await ctx.llm.generate({
const result = await assemble(ctx,{
model: PRO,
messages: ask('Reply with exactly the word: pong'),
maxTokens: 50,

33
packages/llm/llm-deepseek/tests/adapter.spec.ts
View File

@@ -2,9 +2,10 @@ import { createServer } from 'node:http'
import type { IncomingMessage, Server, ServerResponse } from 'node:http'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import LlmService, { LlmError } from '@deepseek-ai/dsh-llm'
import LlmService, { LlmError, userAgent } from '@deepseek-ai/dsh-llm'
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
import { DeepSeekAdapter, httpErrorCode } from '@deepseek-ai/dsh-llm-deepseek'
import { assemble } from './assemble.ts'
/** One scripted behavior for the next request the mock server receives. */
type Behavior =
@@ -90,11 +91,11 @@ async function harness(baseURL: string, config: object = {}) {
}
describe('DeepSeekAdapter against a mock server', () => {
it('streams a text generation end to end through ctx.llm.generate', async () => {
it('streams a text generation end to end through the assembler', async () => {
const server = await mockServer([{ kind: 'sse', events: textEvents }])
const ctx = await harness(server.url)
const result = await ctx.llm.generate({
const result = await assemble(ctx, {
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
})
@@ -108,8 +109,12 @@ describe('DeepSeekAdapter against a mock server', () => {
stream: true,
stream_options: { include_usage: true },
})
// Attribution header identifies the harness to the provider.
expect(server.headers[0]?.['user-agent']).toMatch(/^deepseek-harness\//)
// Attribution reaches the wire: the exact shared User-Agent, and no
// provider-specific headers without an explicitly configured target.
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 raw chunks through ctx.llm.stream', async () => {
@@ -130,7 +135,7 @@ describe('DeepSeekAdapter against a mock server', () => {
const server = await mockServer([{ kind: 'sse', events: textEvents }])
const ctx = await harness(server.url, { thinking: 'disabled', reasoningEffort: 'high' })
await ctx.llm.generate({
await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'hi' }] }],
})
@@ -155,15 +160,15 @@ describe('DeepSeekAdapter against a mock server', () => {
}
const server = await mockServer([behavior, behavior, behavior])
const ctx = await harness(server.url)
await expect(ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] }))
await expect(assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(`failed with ${status}`)
await expect(
ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
.catch((error: unknown) => (error as LlmError).code),
).resolves.toBe(code)
// The numeric HTTP status is carried on the error for explicit handling.
await expect(
ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
.catch((error: unknown) => (error as LlmError).status),
).resolves.toBe(status)
})
@@ -171,14 +176,14 @@ describe('DeepSeekAdapter against a mock server', () => {
it('keeps the status-line message for JSON error bodies without a message', async () => {
const server = await mockServer([{ kind: 'http-error', status: 500, body: '{"error":{"type":"x"}}' }])
const ctx = await harness(server.url)
await expect(ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] }))
await expect(assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(/HTTP 500/)
})
it('keeps the status-line message for non-JSON error bodies', async () => {
const server = await mockServer([{ kind: 'http-error', status: 502, body: 'Bad Gateway', contentType: 'text/plain' }])
const ctx = await harness(server.url)
await expect(ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] }))
await expect(assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(/HTTP 502/)
})
@@ -207,7 +212,7 @@ describe('DeepSeekAdapter against a mock server', () => {
events: ['{"choices":[{"delta":{"content":"par"}}]}'],
}])
const ctx = await harness(server.url)
await expect(ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] }))
await expect(assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] }))
.rejects.toThrow(/terminated|socket|without \[DONE\]/)
})
@@ -278,7 +283,7 @@ describe('plugin registration and config', () => {
vi.stubEnv('DEEPSEEK_BASE_URL', 'http://env-host:1')
const server = await mockServer([{ kind: 'sse', events: textEvents }])
const ctx = await harness(server.url) // harness passes explicit config
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(server.requests).toHaveLength(1) // hit the explicit URL, not env
})
@@ -288,7 +293,7 @@ describe('plugin registration and config', () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmDeepSeek, { apiKey: 'k', models: ['deepseek-v4-flash'] })
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(server.requests).toHaveLength(1)
})

26
packages/llm/llm-deepseek/tests/assemble.ts Normal file
View File

@@ -0,0 +1,26 @@
/**
* Test helper: drive `ctx.llm.stream()` through a `BlockAssembler` and return
* the assembled message + usage + finish reason. This exercises the same
* streaming path production uses (the loop), rather than a service-level
* one-shot convenience method.
*/
import { BlockAssembler } from '@deepseek-ai/dsh-llm'
import type { Context } from 'cordis'
import type { FinishReason, GenerateOptions, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
export interface AssembledResult {
message: Message
usage?: TokenUsage
finish: FinishReason
}
export async function assemble(ctx: Context, options: GenerateOptions): Promise<AssembledResult> {
const assembler = new BlockAssembler()
for await (const chunk of ctx.llm.stream(options)) assembler.push(chunk)
return {
message: assembler.message(),
...assembler.usage !== undefined ? { usage: assembler.usage } : {},
finish: assembler.finish,
}
}

2
packages/llm/llm-deepseek/tests/translate.spec.ts
View File

@@ -47,7 +47,7 @@ describe('translate: text', () => {
))) {
assembler.push(chunk)
}
const result = assembler.result()
const result = { message: assembler.message(), finish: assembler.finish }
expect(result.message.content).toEqual([{ type: 'text', text: 'hi' }])
expect(result.finish).toEqual({ kind: 'stop' })
})

2
packages/llm/llm-deepseek/tsconfig.json
View File

@@ -2,7 +2,7 @@
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib"
"outDir": "lib/types"
},
"include": [
"src"

4
packages/llm/llm-pi-ai/README.md
View File

@@ -25,6 +25,10 @@ Same shape as llm-deepseek (one-line swap in cordis.yml), with pi-ai's thinking-
reasoning: high # off | high | xhigh (xhigh → wire 'max')
```
## 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).
## 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.

8
packages/llm/llm-pi-ai/package.json
View File

@@ -5,17 +5,19 @@
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/index.d.ts",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/index.d.ts",
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib",
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",

12
packages/llm/llm-pi-ai/src/adapter.ts
View File

@@ -13,7 +13,8 @@
import { stream as piStream } from '@earendil-works/pi-ai'
import type { Model } from '@earendil-works/pi-ai'
import { LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import { attributionHeaders, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import { CallId } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk, ToolSchema } from '@deepseek-ai/dsh-llm'
import { toPiContext, toStreamChunks } from './convert.ts'
@@ -69,8 +70,8 @@ type Payload = {
stop?: unknown
}
function rawToolArguments(options: GenerateOptions): Map<string, string> {
const raw = new Map<string, string>()
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) {
@@ -116,7 +117,7 @@ function patchPayload(payload: unknown, options: GenerateOptions, reasoning: PiA
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(call.id)
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
}
@@ -170,6 +171,9 @@ export class PiAiAdapter extends LlmAdapter {
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 } : {},
signal: controller.signal,

2
packages/llm/llm-pi-ai/src/convert.ts
View File

@@ -57,7 +57,7 @@ function parseArguments(raw: string): Record<string, unknown> {
* same id.
*/
export function toPiContext(options: GenerateOptions): PiContext {
const toolNames = new Map<string, string>()
const toolNames = new Map<CallId, string>()
const messages: PiMessage[] = []
for (const message of options.messages) {

19
packages/llm/llm-pi-ai/tests/adapter.e2e.ts
View File

@@ -1,10 +1,11 @@
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
import type { GenerateResult, Message, ToolSchema } 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 * 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
@@ -33,14 +34,14 @@ function ask(text: string): Message[] {
return [{ role: 'user', content: [{ type: 'text', text }] }]
}
function textOf(result: GenerateResult): string {
function textOf(result: AssembledResult): string {
return result.message.content
.filter(block => block.type === 'text')
.map(block => block.text)
.join('')
}
function blockKinds(result: GenerateResult): string[] {
function blockKinds(result: AssembledResult): string[] {
return result.message.content.map(block => block.type)
}
@@ -57,7 +58,7 @@ 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' })
const result = await ctx.llm.generate({
const result = await assemble(ctx,{
model,
messages: ask('Reply with exactly the word: pong'),
maxTokens: 50,
@@ -69,7 +70,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () =>
it.each([FLASH, PRO])('%s + reasoning high: reasoning blocks present', async (model) => {
const ctx = await harness(model, { reasoning: 'high' })
const result = await ctx.llm.generate({
const result = await assemble(ctx,{
model,
messages: ask('Which is larger, 9.11 or 9.8? Answer with just the number.'),
maxTokens: 2000,
@@ -82,7 +83,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () =>
it('pro + reasoning xhigh (wire max): tool-call round trip', async () => {
const ctx = await harness(PRO, { reasoning: 'xhigh' })
const first = await ctx.llm.generate({
const first = await assemble(ctx,{
model: PRO,
messages: ask('What is the weather in Paris right now? Use the get_weather tool.'),
tools: [weatherTool],
@@ -94,7 +95,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () =>
expect(call!.name).toBe('get_weather')
expect(JSON.parse(call!.arguments)).toMatchObject({ city: expect.stringMatching(/paris/i) as string })
const second = await ctx.llm.generate({
const second = await assemble(ctx,{
model: PRO,
messages: [
...ask('What is the weather in Paris right now? Use the get_weather tool.'),
@@ -128,8 +129,8 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () =>
const prompt = ask('Reply with exactly the word: pong')
const [fromDeepSeek, fromPiAi] = await Promise.all([
deepseekCtx.llm.generate({ model: FLASH, messages: prompt, maxTokens: 50 }),
piCtx.llm.generate({ model: FLASH, messages: prompt, maxTokens: 50 }),
assemble(deepseekCtx, { model: FLASH, messages: prompt, maxTokens: 50 }),
assemble(piCtx, { model: FLASH, messages: prompt, maxTokens: 50 }),
])
expect(blockKinds(fromPiAi)).toEqual(blockKinds(fromDeepSeek))
expect(fromPiAi.finish.kind).toBe(fromDeepSeek.finish.kind)

52
packages/llm/llm-pi-ai/tests/adapter.spec.ts
View File

@@ -2,14 +2,17 @@ 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, LlmError } from '@deepseek-ai/dsh-llm'
import LlmService, { CallId, 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 { assemble } from './assemble.ts'
/** Scripted SSE responses, one per request (OpenAI chat-completions shape). */
interface MockServer {
url: string
requests: unknown[]
/** Header bags of received requests, in order (parallel to `requests`). */
headers: IncomingMessage['headers'][]
close(): Promise<void>
}
@@ -21,11 +24,13 @@ afterEach(async () => {
async function mockServer(script: { status?: number; events?: string[]; body?: string }[]): Promise<MockServer> {
const requests: unknown[] = []
const headers: IncomingMessage['headers'][] = []
const server = createServer((request: IncomingMessage, response: ServerResponse) => {
let body = ''
request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') })
request.on('end', () => {
requests.push(JSON.parse(body))
headers.push(request.headers)
const behavior = script.shift() ?? { status: 500, body: 'script exhausted' }
if (behavior.status !== undefined && behavior.status !== 200) {
response.writeHead(behavior.status, { 'content-type': 'application/json' })
@@ -44,6 +49,7 @@ async function mockServer(script: { status?: number; events?: string[]; body?: s
return {
url: `http://127.0.0.1:${address.port}`,
requests,
headers,
close: () => new Promise(resolve => server.close(() => { resolve() })),
}
}
@@ -79,24 +85,32 @@ async function harness(baseURL: string, config: object = {}) {
}
describe('PiAiAdapter against a mock server', () => {
it('streams a text generation through ctx.llm.generate', async () => {
it('streams a text generation through the assembler', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
const result = await ctx.llm.generate({
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 })
// Attribution reaches the wire through pi-ai's headers hook: the exact
// shared User-Agent, and no provider-specific headers without an
// explicitly configured target.
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 ctx.llm.generate({
const result = await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'weather?' }] }],
tools: [{
@@ -114,7 +128,7 @@ describe('PiAiAdapter against a mock server', () => {
const server = await mockServer([{ events: thinkingEvents }])
const ctx = await harness(server.url, { reasoning: 'high' })
const result = await ctx.llm.generate({
const result = await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [{ role: 'user', content: [{ type: 'text', text: 'think' }] }],
})
@@ -127,7 +141,7 @@ describe('PiAiAdapter against a mock server', () => {
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 ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
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
@@ -137,21 +151,21 @@ describe('PiAiAdapter against a mock server', () => {
it('disables thinking for reasoning: off', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url, { reasoning: 'off' })
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
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 ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [], stop: ['END'] })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [], stop: ['END'] })
expect(server.requests[0]).toMatchObject({ stop: ['END'] })
})
it('preserves per-tool strict exactly through onPayload', async () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
await ctx.llm.generate({
await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
tools: [
@@ -173,7 +187,7 @@ describe('PiAiAdapter against a mock server', () => {
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 ctx.llm.generate({
await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [{
role: 'assistant',
@@ -192,7 +206,7 @@ describe('PiAiAdapter against a mock server', () => {
body: JSON.stringify({ error: { message: 'bad key' } }),
}])
const ctx = await harness(server.url)
const result = await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
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/)
})
@@ -204,13 +218,13 @@ describe('PiAiAdapter against a mock server', () => {
] as const)('maps HTTP %s to stable error code %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 ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
const result = await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(result.finish).toMatchObject({ kind: 'error', code })
})
it('rejects prefill with UNSUPPORTED', async () => {
const ctx = await harness('http://127.0.0.1:1')
await expect(ctx.llm.generate({
await expect(assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
prefill: [{ type: 'text', text: 'Sure' }],
@@ -244,7 +258,7 @@ describe('option spreads and env fallbacks', () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
const controller = new AbortController()
await ctx.llm.generate({
await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
temperature: 0.5,
@@ -262,7 +276,7 @@ describe('option spreads and env fallbacks', () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(LlmPiAi, { models: ['deepseek-v4-flash'] })
await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
await assemble(ctx,{ model: 'deepseek-v4-flash', messages: [] })
expect(server.requests).toHaveLength(1)
} finally {
vi.unstubAllEnvs()
@@ -311,7 +325,7 @@ describe('review fixes', () => {
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 ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] })
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)
@@ -320,7 +334,7 @@ describe('review fixes', () => {
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 ctx.llm.generate({
await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [
{ role: 'user', content: [{ type: 'text', text: 'weather?' }] },
@@ -376,7 +390,7 @@ describe('review fixes: abort wiring', () => {
const controller = new AbortController()
controller.abort('already cancelled')
// pi-ai surfaces the abort as an in-stream error event → aborted finish.
const result = await ctx.llm.generate({
const result = await assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
signal: controller.signal,
@@ -388,7 +402,7 @@ describe('review fixes: abort wiring', () => {
const server = await mockServer([{ events: textEvents }])
const ctx = await harness(server.url)
const controller = new AbortController()
const pending = ctx.llm.generate({
const pending = assemble(ctx,{
model: 'deepseek-v4-flash',
messages: [],
signal: controller.signal,

26
packages/llm/llm-pi-ai/tests/assemble.ts Normal file
View File

@@ -0,0 +1,26 @@
/**
* Test helper: drive `ctx.llm.stream()` through a `BlockAssembler` and return
* the assembled message + usage + finish reason. This exercises the same
* streaming path production uses (the loop), rather than a service-level
* one-shot convenience method.
*/
import { BlockAssembler } from '@deepseek-ai/dsh-llm'
import type { Context } from 'cordis'
import type { FinishReason, GenerateOptions, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
export interface AssembledResult {
message: Message
usage?: TokenUsage
finish: FinishReason
}
export async function assemble(ctx: Context, options: GenerateOptions): Promise<AssembledResult> {
const assembler = new BlockAssembler()
for await (const chunk of ctx.llm.stream(options)) assembler.push(chunk)
return {
message: assembler.message(),
...assembler.usage !== undefined ? { usage: assembler.usage } : {},
finish: assembler.finish,
}
}

2
packages/llm/llm-pi-ai/tsconfig.json
View File

@@ -2,7 +2,7 @@
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib"
"outDir": "lib/types"
},
"include": [
"src"

17
packages/llm/llm/README.md
View File

@@ -4,28 +4,24 @@ Provider-neutral LLM vocabulary and abstract service. This package defines the c
## Service: `LlmService` (ctx key: `llm`)
An adapter registry plus streaming / non-streaming call surfaces. Both call surfaces are interceptable via waterfall events.
An adapter registry plus a single streaming call surface, interceptable via a waterfall event.
### 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.stream(options: GenerateOptions): AsyncIterable<StreamChunk>` Stream one model call as raw chunks (token-level deltas).
- `ctx.llm.streamBlocks(options: GenerateOptions): AsyncIterable<ContentBlock>` Stream as completed content blocks (convenience view).
- `ctx.llm.generate(options: GenerateOptions): Promise<GenerateResult>` One model call, fully assembled.
- `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`.
### Events
| Event | Mode | Purpose |
|---|---|---|
| `llm/stream` | waterfall | Intercept/wrap every streaming model call (retry, caching, routing) |
| `llm/generate` | waterfall | Intercept/wrap every non-streaming model call |
| `llm/adapter-change` | emit | An adapter was registered or unregistered |
### Extension points
- Subclass `LlmAdapter` and call `ctx.llm.registerAdapter(models, adapter)` to add a new model provider.
- Wrap `llm/stream` or `llm/generate` via `ctx.on()` waterfall listeners for caching, retry, logging, rate-limiting, etc.
- Wrap `llm/stream` via `ctx.on()` waterfall listeners for caching, retry, logging, rate-limiting, etc.
### Content-block vocabulary (`types.ts`)
@@ -33,11 +29,14 @@ Messages are arrays of typed content blocks: `text`, `reasoning`, `tool-call`, `
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.
### App attribution (`attribution.ts`)
Every product adapter must identify the application on every provider HTTP request - attribution is part of the adapter contract, not an adapter-local nicety. `attributionHeaders(identity?)` builds the standard `User-Agent` header (`product/version (+url)`, from `userAgent()`) for every request. The default `APP_IDENTITY` carries only static public product facts (its version is read from this package's manifest); a white-label deployment passes its own `AppIdentity`, and omission falls back to the default - nothing can suppress attribution. OpenRouter-specific app attribution headers are intentionally not supported by this contract. An adapter proves compliance with a wire-level test: a mock server asserting the received header (or, for a library-backed adapter, that the library's header hook delivers the same value). Policy and rationale: [Mandatory `User-Agent` attribution](../../../docs/rfc/implemented/architecture/2026-06-21-mandatory-app-attribution-headers.md).
### Classes
- `LlmAdapter` — abstract base class for provider adapters. The only required method is `stream()`.
- `BlockAssembler` — incrementally assembles raw chunks into complete content blocks and an assistant message. Used by the agent loop (raw chunks for replay
+ assembled for history) and by `streamBlocks()`/`generate()`.
- `BlockAssembler` — incrementally assembles raw chunks into complete content blocks and an assistant message. The agent loop feeds it raw chunks (logging them for replay) while reading the assembled blocks/message for history.
- `HarnessError` — base class for the harness error taxonomy: a stable `code` string (distinct from the human `message`) plus `cause` chaining. Lives here, in the leaf package every other imports, so a single base is shared without a new dependency edge. Per-package errors (`LlmError`, `ToolArgsError`, `InvariantError`, …) extend it. `isHarnessError(value)` narrows at seams.
- `LlmError` — extends `HarnessError`; `code` string (`NO_ADAPTER`, `DUPLICATE_ADAPTER`, and adapter codes like `AUTH`/`RATE_LIMIT`) plus an optional numeric `status` when the failure came from a non-2xx provider response.

10
packages/llm/llm/package.json
View File

@@ -5,24 +5,28 @@
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/index.d.ts",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/index.d.ts",
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib",
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-brand": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"devDependencies": {
"@deepseek-ai/dsh-brand": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

58
packages/llm/llm/src/assembler.ts
View File

@@ -1,13 +1,14 @@
/**
* Incremental chunk-to-message assembler. This is the single canonical assembly
* algorithm used by both the agent loop and the LLM service convenience views.
* algorithm used by the agent loop to build an assistant message from a chunk
* stream while logging the raw chunks for replay fidelity.
*
* @module @deepseek-ai/dsh-llm/assembler
*/
import { CallId } from './brand.ts'
import { assertNever } from './never.ts'
import type { ContentBlock, FinishReason, GenerateResult, Message, StreamChunk, TokenUsage } from './types.ts'
import type { ContentBlock, FinishReason, Message, StreamChunk, TokenUsage } from './types.ts'
interface PartialBlock {
blockType: string
@@ -23,9 +24,8 @@ interface PartialBlock {
* Incrementally assembles raw {@link StreamChunk}s into complete
* {@link ContentBlock}s and a final assistant {@link Message}.
*
* This is the single shared assembly implementation: the agent loop feeds it
* while logging raw chunks for replay fidelity, and `LlmService.generate()` /
* `streamBlocks()` use it to offer assembled views of the same stream.
* The agent loop feeds it while logging raw chunks for replay fidelity, then
* reads `blocks()` / `message()` / `usage` / `finish` once the stream ends.
*
* Tolerant of delta-only protocols (no block-start/end); deltas arriving for
* an index already closed by `block-end` are ignored (malformed stream) so a
@@ -34,7 +34,6 @@ interface PartialBlock {
export class BlockAssembler {
private partials = new Map<number, PartialBlock>()
private order: number[] = []
private flushed = 0
private _usage: TokenUsage | undefined
private _finish: FinishReason | undefined
@@ -129,44 +128,6 @@ export class BlockAssembler {
return this.order.map(index => this.assemble(this.mustGet(index), index))
}
/**
* Streaming flush: returns (once) every block that is complete AND has no
* incomplete block before it in stream order. Call after each `push()`;
* blocks come out strictly in stream order, so a streaming consumer sees
* exactly the sequence `blocks()` would produce.
*/
flushReady(): ContentBlock[] {
const ready: ContentBlock[] = []
while (this.flushed < this.order.length) {
const index = this.order[this.flushed]
/* v8 ignore next 3 -- noUncheckedIndexedAccess guard: loop condition guarantees index exists in a non-empty array */
if (index === undefined) break
const partial = this.mustGet(index)
if (!partial.block) break
ready.push(partial.block)
this.flushed += 1
}
return ready
}
/**
* End-of-stream flush: returns (once) all not-yet-flushed blocks, in stream
* order, assembling still-open ones from their deltas (delta-only
* protocols). After this, `flushReady()` + `flushRemaining()` together have
* yielded exactly `blocks()`.
*/
flushRemaining(): ContentBlock[] {
const remaining: ContentBlock[] = []
while (this.flushed < this.order.length) {
const index = this.order[this.flushed]
/* v8 ignore next 3 -- noUncheckedIndexedAccess guard: loop condition guarantees index exists */
if (index === undefined) break
remaining.push(this.assemble(this.mustGet(index), index))
this.flushed += 1
}
return remaining
}
get usage(): TokenUsage | undefined {
return this._usage
}
@@ -179,13 +140,4 @@ export class BlockAssembler {
message(): Message {
return { role: 'assistant', content: this.blocks() }
}
/** The assembled non-streaming result. */
result(): GenerateResult {
return {
message: this.message(),
...this._usage !== undefined ? { usage: this._usage } : {},
finish: this.finish,
}
}
}

71
packages/llm/llm/src/attribution.ts Normal file
View File

@@ -0,0 +1,71 @@
/**
* App-attribution vocabulary for provider requests.
*
* Every product LLM adapter must identify the application on every provider
* HTTP request (see the adapter contract on {@link ../index.ts LlmAdapter}):
* a static, non-secret product identity, sent as the standard `User-Agent`.
* Adapters obtain the headers from {@link attributionHeaders} instead of
* hand-copying constants, so the identity cannot drift between
* implementations. The policy and its rationale are pinned in
* docs/rfc/implemented/architecture/2026-06-21-mandatory-app-attribution-headers.md.
*
* @module @deepseek-ai/dsh-llm/attribution
*/
import { createRequire } from 'node:module'
// The package's own manifest is the single source of the version so the
// User-Agent cannot drift from what is published (`./package.json` is an
// export of this package; the relative path resolves from both `src/` and
// the bundled `lib/`).
const { version } = createRequire(import.meta.url)('../package.json') as { version: string }
/**
* Static public application identity sent to LLM providers.
*
* Every field is a public product fact, safe on every request: no secrets,
* local paths, session ids, prompt text, or per-user identifiers belong here,
* and nothing per-request may influence the values.
*/
export interface AppIdentity {
/** `User-Agent` product token (lowercase, hyphenated). */
product: string
/** Product version; sourced from package metadata, never hand-copied. */
version: string
/** Public home URL of the app, used as the `User-Agent` comment. */
url: string
}
/**
* The harness's own identity: the default every adapter sends. Deployments
* that need a white-label identity pass their own {@link AppIdentity} to
* {@link attributionHeaders} — omission falls back to this default; nothing
* can suppress attribution entirely.
*/
export const APP_IDENTITY: AppIdentity = {
product: 'deepseek-harness',
version,
// FIXME: create the public deepseek-ai/deepseek-harness-sdk repository this
// URL promises before the first release ships attribution pointing at it.
url: 'https://github.com/deepseek-ai/deepseek-harness-sdk',
}
/**
* The standard `User-Agent` value: `product/version (+url)`. The
* parenthesized `+url` comment is the conventional self-identification form
* (RFC 9110 §10.1.5 product + comment syntax).
*/
export function userAgent(identity: AppIdentity = APP_IDENTITY): string {
return `${identity.product}/${identity.version} (+${identity.url})`
}
/**
* Build the attribution headers an adapter must send on every provider
* request. Header names are lowercase (HTTP field names are case-insensitive
* on the wire).
*/
export function attributionHeaders(
identity: AppIdentity = APP_IDENTITY,
): Record<string, string> {
return { 'user-agent': userAgent(identity) }
}

21
packages/llm/llm/src/brand.ts
View File

@@ -1,24 +1,15 @@
/**
* Branded (nominal) ID types.
* dsh-llm's owned branded id: `CallId` (tool-call correlation).
*
* A brand makes structurally-identical strings non-interchangeable at the
* type level: an `AgentId` cannot be passed where a `CallId` is expected,
* even though both are strings at runtime. Construction goes through the
* per-type factory (a plain cast inside — zero runtime cost); comparison,
* logging, and serialization all behave as ordinary strings.
*
* Policy: core packages brand the IDs they own — `CallId` here (tool-call
* correlation), `SessionId` in dsh-session, `AgentId` in dsh-agent. Branding
* is for IDs that cross package boundaries and could plausibly be confused;
* not every string needs a brand.
* The `Branded<B>` primitive itself lives in `@deepseek-ai/dsh-brand` (a
* zero-dependency type-only package) so every owner of a cross-boundary id can
* brand it without depending on dsh-llm; see that package's README for the
* nominal-typing policy.
*
* @module @deepseek-ai/dsh-llm/brand
*/
declare const BRAND: unique symbol
/** A string carrying a compile-time-only brand `B`. */
export type Branded<B extends string> = string & { readonly [BRAND]: B }
import type { Branded } from '@deepseek-ai/dsh-brand'
/**
* Correlates a model-issued tool call with its result. Provider-issued for

73
packages/llm/llm/src/index.ts
View File

@@ -1,16 +1,16 @@
/**
* LLM service: adapter registry with waterfall-interceptable streaming and
* non-streaming call surfaces. Exports the `LlmService` default, the abstract
* `LlmAdapter` for provider backends, and `BlockAssembler` for chunk assembly.
* LLM service: adapter registry with a waterfall-interceptable streaming call
* surface. Exports the `LlmService` default, the abstract `LlmAdapter` for
* provider backends, and `BlockAssembler` for chunk assembly.
*
* @module @deepseek-ai/dsh-llm
*/
import { Context, Service } from 'cordis'
import type { ContentBlock, GenerateOptions, GenerateResult, StreamChunk } from './types.ts'
import { BlockAssembler } from './assembler.ts'
import type { GenerateOptions, StreamChunk } from './types.ts'
import { HarnessError } from './error.ts'
export * from './attribution.ts'
export * from './brand.ts'
export * from './never.ts'
export * from './error.ts'
@@ -30,17 +30,6 @@ declare module 'cordis' {
* @mode waterfall
*/
'llm/stream'(this: LlmService, options: GenerateOptions, next: () => AsyncIterable<StreamChunk>): AsyncIterable<StreamChunk>
/**
* Waterfall around every non-streaming model call. Bound to the
* {@link LlmService}; call `next()` to delegate to the adapter.
* @mode waterfall
*/
'llm/generate'(this: LlmService, options: GenerateOptions, next: () => Promise<GenerateResult>): Promise<GenerateResult>
/**
* An adapter was registered or unregistered (the model→adapter map changed).
* @mode emit
*/
'llm/adapter-change'(): void
}
}
@@ -68,6 +57,13 @@ export class LlmError extends HarnessError {
* fetch/SSE) and `@deepseek-ai/dsh-llm-pi-ai` (pi-ai-backed) — two
* deliberately different internals over the same contract; see the
* adapter contract documented on `StreamChunk` in `./types.ts`.
*
* App attribution is part of the adapter contract: every HTTP request to a
* provider carries the headers from `attributionHeaders()` (`./attribution.ts`)
* — the standard `User-Agent` baseline everywhere. An adapter proves it with
* a wire-level test (a mock server asserting the received header), or, for a
* library-backed adapter, by asserting the library's header hook delivers the
* same value to the wire.
*/
export abstract class LlmAdapter {
/** Stream one model call as raw chunks. The only required method. */
@@ -75,8 +71,8 @@ export abstract class LlmAdapter {
}
/**
* The abstract `llm` service: an adapter registry plus streaming /
* non-streaming call surfaces, both interceptable via waterfall events.
* The abstract `llm` service: an adapter registry plus a streaming model-call
* surface, interceptable via the `llm/stream` waterfall.
*/
export class LlmService extends Service {
private adapters = new Map<string, LlmAdapter>()
@@ -88,8 +84,7 @@ export class LlmService extends Service {
/**
* Register an adapter for the given model names. Throws `LlmError` with code
* `DUPLICATE_ADAPTER` if any model already has an adapter (all-or-nothing).
* Emits `llm/adapter-change` on registration and disposal. Disposed with the
* fiber.
* Disposed with the fiber.
*/
registerAdapter(models: string[], adapter: LlmAdapter): () => void {
const dispose = this.ctx.effect(function* (this: LlmService) {
@@ -99,17 +94,9 @@ export class LlmService extends Service {
}
}
for (const model of models) this.adapters.set(model, adapter)
// Yield the rollback BEFORE emitting the change event: a generator effect
// collects each yielded disposer before running the next step, so a
// throwing `llm/adapter-change` listener rolls the mutation back instead
// of leaking the entry (which would wedge the duplicate check until
// restart). The duplicate throws above fire before any mutation, so they
// correctly leak nothing.
yield () => {
for (const model of models) this.adapters.delete(model)
this.ctx.emit('llm/adapter-change')
}
this.ctx.emit('llm/adapter-change')
}.bind(this), 'llm.registerAdapter()')
// ctx.effect's disposer returns Promise<void>; our disposer API is
// synchronous fire-and-forget — discard the (always-resolved) promise.
@@ -137,36 +124,6 @@ export class LlmService extends Service {
return this.adapter(options.model).stream(options)
})
}
/**
* Stream one model call as completed content blocks — a convenience view
* for consumers that don't care about token-level deltas. Blocks are
* yielded strictly in stream order as soon as they (and everything before
* them) complete; blocks left open at end of stream (delta-only protocols)
* are assembled and flushed last, so the sequence always equals
* `generate()`'s `message.content`.
*/
async * streamBlocks(options: GenerateOptions): AsyncIterable<ContentBlock> {
const assembler = new BlockAssembler()
for await (const chunk of this.stream(options)) {
assembler.push(chunk)
yield * assembler.flushReady()
}
yield * assembler.flushRemaining()
}
/**
* One model call, fully assembled (drains the chunk stream). Dispatches
* through the `llm/generate` waterfall (and the inner stream through
* `llm/stream`). Same completion guarantees as `streamBlocks()`.
*/
generate(options: GenerateOptions): Promise<GenerateResult> {
return this.ctx.waterfall(this, 'llm/generate', options, async () => {
const assembler = new BlockAssembler()
for await (const chunk of this.stream(options)) assembler.push(chunk)
return assembler.result()
})
}
}
export default LlmService

22
packages/llm/llm/src/types.ts
View File

@@ -19,6 +19,7 @@
* ```
*/
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { CallId } from './brand.ts'
/** Cache hint attached to a content block (provider-interpreted). */
@@ -192,11 +193,18 @@ export interface GenerateOptions {
*/
stop?: string[]
signal?: AbortSignal
}
/** Non-streaming result, assembled from the chunk stream. */
export interface GenerateResult {
message: Message
usage?: TokenUsage
finish: FinishReason
/**
* The id of the session this request belongs to — stamped by the agent loop
* from `agent.session.id`. Adapters ignore it; it lets an `llm/stream` listener
* route a call by WHICH session issued it (the replay adapter keys its per-call
* cursor by session, so a parent and its in-process subagent — each with its
* own session on one context — replay from their own recorded scripts).
*
* Typed as `Branded<'SessionId'>` rather than importing `SessionId` from
* `dsh-session`: that package imports `Message` from here, so importing its
* `SessionId` back would cycle. `SessionId` IS `Branded<'SessionId'>`, so a
* real session id assigns with no cast. (A future ids package could own the
* brand and dissolve this note.)
*/
sessionId?: Branded<'SessionId'>
}

50
packages/llm/llm/tests/assembler.spec.ts
View File

@@ -81,36 +81,6 @@ describe('BlockAssembler', () => {
expect(() => assembler.blocks()).toThrow('BlockAssembler invariant violated')
})
it('assembles open blocks at end of stream via flushRemaining', () => {
const assembler = new BlockAssembler()
assembler.push({ type: 'text-delta', index: 0, text: 'open' })
assembler.push({ type: 'reasoning-delta', index: 1, text: 'thinking' })
// flushReady returns nothing because index 0 is incomplete and blocking
const ready = assembler.flushReady()
expect(ready).toEqual([])
// flushRemaining assembles everything still open
const remaining = assembler.flushRemaining()
expect(remaining).toEqual([
{ type: 'text', text: 'open' },
{ type: 'reasoning', text: 'thinking' },
])
// blocks() now matches the flushed view
expect(assembler.blocks()).toEqual(remaining)
})
it('result() omits usage key when no usage was received', () => {
const assembler = new BlockAssembler()
assembler.push({ type: 'text-delta', index: 0, text: 'msg' })
const result = assembler.result()
expect(result.message).toBeDefined()
expect(result.finish).toEqual({ kind: 'stop' })
// usage should NOT be present on the object at all
expect('usage' in result).toBe(false)
})
it('ignores duplicate block-start for the same index', () => {
const assembler = new BlockAssembler()
assembler.push({ type: 'block-start', index: 0, blockType: 'text' })
@@ -142,13 +112,11 @@ describe('BlockAssembler', () => {
])
})
it('includes usage in result() when usage was received', () => {
it('exposes usage via the getter when a usage chunk was received', () => {
const assembler = new BlockAssembler()
assembler.push({ type: 'text-delta', index: 0, text: 'msg' })
assembler.push({ type: 'usage', usage: { inputTokens: 5, outputTokens: 3 } })
const result = assembler.result()
expect(result.usage).toEqual({ inputTokens: 5, outputTokens: 3 })
expect('usage' in result).toBe(true)
expect(assembler.usage).toEqual({ inputTokens: 5, outputTokens: 3 })
})
})
@@ -172,25 +140,25 @@ describe('BlockAssembler regressions (property-test findings)', () => {
// Found by fast-check (the property-testing RFC): two block-ends at the same index made the
// streamed prefix (first block) disagree with final blocks() (second
// block). The first close must win — same straggler rule as post-close
// deltas — so streaming and one-shot assembly stay identical.
// deltas — so the prefix returned incrementally by push() and the final
// blocks() stay identical.
const chunks: StreamChunk[] = [
{ type: 'block-end', index: 0, block: { type: 'reasoning', text: 'first' } },
{ type: 'block-end', index: 0, block: { type: 'text', text: 'second' } },
]
const streaming = new BlockAssembler()
const flushed = []
const closed = []
for (const chunk of chunks) {
streaming.push(chunk)
flushed.push(...streaming.flushReady())
const block = streaming.push(chunk)
if (block) closed.push(block)
}
flushed.push(...streaming.flushRemaining())
const oneShot = new BlockAssembler()
for (const chunk of chunks) oneShot.push(chunk)
expect(flushed).toEqual([{ type: 'reasoning', text: 'first' }])
expect(closed).toEqual([{ type: 'reasoning', text: 'first' }])
expect(oneShot.blocks()).toEqual([{ type: 'reasoning', text: 'first' }])
expect(flushed).toEqual(oneShot.blocks())
expect(closed).toEqual(oneShot.blocks())
})
it('push returns undefined for a duplicate block-end (it closed nothing)', () => {

51
packages/llm/llm/tests/attribution.spec.ts Normal file
View File

@@ -0,0 +1,51 @@
import { createRequire } from 'node:module'
import { describe, expect, it } from 'vitest'
import { APP_IDENTITY, attributionHeaders, userAgent } from '@deepseek-ai/dsh-llm'
import type { AppIdentity } from '@deepseek-ai/dsh-llm'
const manifest = createRequire(import.meta.url)('../package.json') as { version: string }
/** A white-label identity exercising every override seam. */
const forkIdentity: AppIdentity = {
product: 'fork-agent',
version: '9.9.9',
url: 'https://example.com/fork-agent',
}
describe('APP_IDENTITY', () => {
it('sources the version from the package manifest, never a hand-copied constant', () => {
expect(APP_IDENTITY.version).toBe(manifest.version)
})
it('carries only static public product facts', () => {
expect(APP_IDENTITY).toEqual({
product: 'deepseek-harness',
version: manifest.version,
url: 'https://github.com/deepseek-ai/deepseek-harness-sdk',
})
})
})
describe('userAgent', () => {
it('renders product/version with the +url comment', () => {
expect(userAgent()).toBe(
`deepseek-harness/${manifest.version} (+https://github.com/deepseek-ai/deepseek-harness-sdk)`,
)
})
it('renders a custom identity', () => {
expect(userAgent(forkIdentity)).toBe('fork-agent/9.9.9 (+https://example.com/fork-agent)')
})
})
describe('attributionHeaders', () => {
it('defaults to the provider-neutral baseline: User-Agent and nothing else', () => {
expect(attributionHeaders()).toEqual({ 'user-agent': userAgent() })
})
it('maps a custom identity onto the User-Agent header only', () => {
expect(attributionHeaders(forkIdentity)).toEqual({
'user-agent': 'fork-agent/9.9.9 (+https://example.com/fork-agent)',
})
})
})

52
packages/llm/llm/tests/properties.spec.ts
View File

@@ -4,13 +4,13 @@
* The assembler is protocol-shaped: arbitrary interleavings of block-start,
* deltas, block-end, usage, and finish — valid and malformed (duplicate
* indices, stragglers after block-end, missing block-start, delta-only). The
* invariants below are the contract the agent loop and LlmService rely on.
* invariants below are the contract the agent loop relies on.
*/
import { describe, expect, it } from 'vitest'
import fc from 'fast-check'
import { BlockAssembler } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm'
import type { StreamChunk } from '@deepseek-ai/dsh-llm'
import { CallId } from '@deepseek-ai/dsh-llm'
// A small pool of indices so collisions (duplicate-index bugs) are common.
@@ -55,38 +55,6 @@ function feed(chunks: StreamChunk[]): BlockAssembler {
}
describe('BlockAssembler properties', () => {
it('flushReady() ++ flushRemaining() === blocks(), in order', () => {
fc.assert(fc.property(streamArb, (chunks) => {
const streaming = new BlockAssembler()
const flushed: ContentBlock[] = []
for (const chunk of chunks) {
streaming.push(chunk)
flushed.push(...streaming.flushReady())
}
flushed.push(...streaming.flushRemaining())
const oneShot = feed(chunks).blocks()
expect(flushed).toEqual(oneShot)
}))
})
it('streamBlocks-style flush never yields a block before an earlier open one', () => {
// flushReady is strict-order: once it stops at an open index, no later
// index may be emitted until that one closes. We assert the flushed prefix
// is always a prefix of the final blocks() order.
fc.assert(fc.property(streamArb, (chunks) => {
const streaming = new BlockAssembler()
const flushed: ContentBlock[] = []
for (const chunk of chunks) {
streaming.push(chunk)
flushed.push(...streaming.flushReady())
}
const finalSoFar = streaming.blocks()
// Everything flushed mid-stream is a prefix of the full ordered blocks.
expect(finalSoFar.slice(0, flushed.length)).toEqual(flushed)
}))
})
it('partials map size never exceeds the number of distinct indices seen', () => {
fc.assert(fc.property(streamArb, (chunks) => {
const distinct = new Set<number>()
@@ -131,20 +99,4 @@ describe('BlockAssembler properties', () => {
}
}))
})
it('streaming and one-shot assembly agree on usage and finish', () => {
fc.assert(fc.property(streamArb, (chunks) => {
// Streaming consumer: push + flush as it goes.
const streaming = new BlockAssembler()
for (const chunk of chunks) {
streaming.push(chunk)
streaming.flushReady()
}
streaming.flushRemaining()
// One-shot consumer: push all, then read.
const oneShot = feed(chunks)
expect(streaming.usage).toEqual(oneShot.usage)
expect(streaming.finish).toEqual(oneShot.finish)
}))
})
})

58
packages/llm/llm/tests/service.spec.ts
View File

@@ -19,24 +19,22 @@ const SCRIPT: StreamChunk[] = [
]
describe('LlmService', () => {
it('routes stream() to the registered adapter and generate() assembles it', async () => {
it('routes stream() to the registered adapter', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], new ScriptedAdapter(SCRIPT))
const chunks: StreamChunk[] = []
for await (const chunk of ctx.llm.stream({ model: 'test-model', messages: [] })) chunks.push(chunk)
expect(chunks).toHaveLength(3)
const result = await ctx.llm.generate({ model: 'test-model', messages: [] })
expect(result.message.content).toEqual([{ type: 'text', text: 'hi' }])
expect(result.finish).toEqual({ kind: 'stop' })
expect(chunks).toEqual(SCRIPT)
})
it('throws NO_ADAPTER for unregistered models', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
await expect(ctx.llm.generate({ model: 'nope', messages: [] })).rejects.toThrow('no adapter registered')
await expect((async () => {
for await (const _ of ctx.llm.stream({ model: 'nope', messages: [] })) { /* drain */ }
})()).rejects.toThrow('no adapter registered')
})
it('unregisters adapters when the owning fiber is disposed (HMR safety)', async () => {
@@ -71,21 +69,6 @@ describe('LlmService', () => {
expect(chunks[0]).toMatchObject({ index: 99 })
})
it('lets llm/generate waterfall listeners intercept and transform the result', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
ctx.llm.registerAdapter(['test-model'], new ScriptedAdapter(SCRIPT))
ctx.on('llm/generate', async function (_options, next) {
const result = await next()
return { ...result, finish: { kind: 'max-tokens' } as const }
})
const result = await ctx.llm.generate({ model: 'test-model', messages: [] })
expect(result.finish).toEqual({ kind: 'max-tokens' })
expect(result.message.content).toEqual([{ type: 'text', text: 'hi' }])
})
it('creates LlmError with a code for programmatic handling', () => {
const err = new LlmError('something went wrong', 'CUSTOM_CODE')
expect(err).toBeInstanceOf(Error)
@@ -116,20 +99,13 @@ describe('LlmService', () => {
expect(isHarnessError('nope')).toBe(false)
})
it('disposes adapter registration on adapter-change event emission', async () => {
it('removes the adapter when the returned disposer is called', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const changes: string[][] = []
ctx.on('llm/adapter-change', () => {
changes.push([...ctx.llm.models()])
})
const dispose = ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))
expect(changes).toEqual([['m1']])
expect(ctx.llm.models()).toEqual(['m1'])
dispose()
expect(changes).toEqual([['m1'], []])
expect(ctx.llm.models()).toEqual([])
})
@@ -147,25 +123,19 @@ describe('LlmService', () => {
}
})
it('rolls back the adapter entry when an adapter-change listener throws (P1-1)', async () => {
it('re-registers a model after its prior registration is disposed', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
// A change listener that throws on the FIRST emit only.
let threw = false
ctx.on('llm/adapter-change', () => {
if (!threw) { threw = true; throw new Error('boom change listener') }
})
// The throwing emit must roll the mutation back, not leak it.
expect(() => ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))).toThrow('boom change listener')
expect(ctx.llm.models()).toEqual([]) // entry rolled back, not leaked
// A subsequent listener-free register of the SAME model succeeds and
// contributes exactly once (the duplicate check is not wedged).
const dispose = ctx.llm.registerAdapter(['m1'], new ScriptedAdapter(SCRIPT))
expect(ctx.llm.models()).toEqual(['m1'])
dispose()
expect(ctx.llm.models()).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'])
disposeAgain()
expect(ctx.llm.models()).toEqual([])
})
})

5
packages/llm/llm/tsconfig.json
View File

@@ -2,7 +2,7 @@
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib"
"outDir": "lib/types"
},
"include": [
"src"
@@ -13,6 +13,9 @@
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../util/brand"
}
]
}