Add abstract service interface packages

@deepseek-ai/dsh-llm: provider-neutral content-block vocabulary
(merge-extensible maps), raw StreamChunk protocol, ToolSchema,
abstract LlmAdapter, LlmService adapter registry, BlockAssembler.

@deepseek-ai/dsh-session: event-sourced Session (append-only log,
deriveMessages; context/steering render as tagged envelopes),
SessionStore, session/event + awaited session/flush durability seam.

@deepseek-ai/dsh-system-prompt: ordered sections + tool-schema
providers; assemble() through the system-prompt/assemble waterfall.
Tool schemas are part of the assembly by design.

@deepseek-ai/dsh-tools: tool registry feeding schemas into the
assembly; execute() through the tools/execute waterfall (the single
sandbox/permission/hook seam).

@deepseek-ai/dsh-agent: Agent interface (send/steer/inject/abort,
spawn/fork TODO seams), AgentRegistry, and the full agent/* event
taxonomy so plugins never depend on the concrete loop.
This commit is contained in:
Tianyi Cui
2026-06-11 10:54:06 +08:00
parent 72688a3888
commit d5a1d9bb75
25 changed files with 1703 additions and 0 deletions

View File

@@ -0,0 +1,115 @@
import type { ContentBlock, FinishReason, GenerateResult, Message, StreamChunk, TokenUsage } from './types.ts'
/**
* 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.
*/
export class BlockAssembler {
private partials = new Map<number, {
blockType: string
text: string
toolCallId?: string
toolCallName?: string
toolCallArguments: string
block?: ContentBlock
}>()
private order: number[] = []
private _usage: TokenUsage | undefined
private _finish: FinishReason | undefined
/**
* Feed one chunk. Returns the completed block when the chunk closes one
* (either an explicit `block-end` or an implicit close), otherwise undefined.
*/
push(chunk: StreamChunk): ContentBlock | undefined {
switch (chunk.type) {
case 'block-start': {
if (!this.partials.has(chunk.index)) this.order.push(chunk.index)
this.partials.set(chunk.index, {
blockType: chunk.blockType,
text: '',
toolCallArguments: '',
})
return
}
case 'text-delta':
case 'reasoning-delta': {
const partial = this.ensure(chunk.index, chunk.type === 'text-delta' ? 'text' : 'reasoning')
partial.text += chunk.text
return
}
case 'tool-call-delta': {
const partial = this.ensure(chunk.index, 'tool-call')
partial.toolCallId = chunk.id
if (chunk.name) partial.toolCallName = chunk.name
partial.toolCallArguments += chunk.argumentsDelta
return
}
case 'block-end': {
const partial = this.ensure(chunk.index, chunk.block.type)
partial.block = chunk.block
return chunk.block
}
case 'usage': {
this._usage = chunk.usage
return
}
case 'finish': {
this._finish = chunk.reason
return
}
}
}
private ensure(index: number, blockType: string) {
let partial = this.partials.get(index)
if (!partial) {
partial = { blockType, text: '', toolCallArguments: '' }
this.partials.set(index, partial)
this.order.push(index)
}
return partial
}
/** Assemble all blocks seen so far, in stream order. */
blocks(): ContentBlock[] {
return this.order.map((index) => {
const partial = this.partials.get(index)!
if (partial.block) return partial.block
switch (partial.blockType) {
case 'text': return { type: 'text', text: partial.text }
case 'reasoning': return { type: 'reasoning', text: partial.text }
case 'tool-call': return {
type: 'tool-call',
id: partial.toolCallId ?? `call-${index}`,
name: partial.toolCallName ?? '',
arguments: partial.toolCallArguments,
}
default: throw new Error(`cannot assemble incomplete block of type "${partial.blockType}"`)
}
})
}
get usage(): TokenUsage | undefined {
return this._usage
}
get finish(): FinishReason {
return this._finish ?? { kind: 'stop' }
}
/** The assembled assistant message. */
message(): Message {
return { role: 'assistant', content: this.blocks() }
}
/** The assembled non-streaming result. */
result(): GenerateResult {
return { message: this.message(), usage: this._usage, finish: this.finish }
}
}

108
packages/llm/src/index.ts Normal file
View File

@@ -0,0 +1,108 @@
import { Context, Service } from 'cordis'
import type { ContentBlock, GenerateOptions, GenerateResult, StreamChunk } from './types.ts'
import { BlockAssembler } from './assembler.ts'
export * from './types.ts'
export { BlockAssembler } from './assembler.ts'
declare module 'cordis' {
interface Context {
llm: LlmService
}
interface Events {
/** Waterfall around every streaming model call (retry, caching, routing). */
'llm/stream'(this: LlmService, options: GenerateOptions, next: () => AsyncIterable<StreamChunk>): AsyncIterable<StreamChunk>
/** Waterfall around every non-streaming model call. */
'llm/generate'(this: LlmService, options: GenerateOptions, next: () => Promise<GenerateResult>): Promise<GenerateResult>
/** An adapter was registered or unregistered. */
'llm/adapter-change'(): void
}
}
export class LlmError extends Error {
constructor(message: string, public code: string) {
super(message)
this.name = 'LlmError'
}
}
/**
* Base class for LLM provider adapters.
*
* An adapter translates between the harness vocabulary (Message/ContentBlock/
* StreamChunk) and one provider's wire format. Adapters register themselves
* via `ctx.llm.registerAdapter(models, adapter)`.
*
* TODO: the first real adapter (DeepSeek V4) lands in a later phase; until
* then only mock adapters (tests, demo) exist.
*/
export abstract class LlmAdapter {
/** Stream one model call as raw chunks. The only required method. */
abstract stream(options: GenerateOptions): AsyncIterable<StreamChunk>
}
/**
* The abstract `llm` service: an adapter registry plus streaming /
* non-streaming call surfaces, both interceptable via waterfall events.
*/
export class LlmService extends Service {
private adapters = new Map<string, LlmAdapter>()
constructor(ctx: Context) {
super(ctx, 'llm')
}
/** Register an adapter for the given model names. Disposed with the fiber. */
registerAdapter(models: string[], adapter: LlmAdapter): () => void {
return this.ctx.effect(() => {
for (const model of models) this.adapters.set(model, adapter)
this.ctx.emit('llm/adapter-change')
return () => {
for (const model of models) this.adapters.delete(model)
this.ctx.emit('llm/adapter-change')
}
}, 'llm.registerAdapter()')
}
/** Model names with a registered adapter. */
models(): string[] {
return [...this.adapters.keys()]
}
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
}
/** Stream one model call as raw chunks (token-level deltas). */
stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
return this.ctx.waterfall(this, 'llm/stream', options, () => {
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.
*/
async * streamBlocks(options: GenerateOptions): AsyncIterable<ContentBlock> {
const assembler = new BlockAssembler()
for await (const chunk of this.stream(options)) {
const block = assembler.push(chunk)
if (block) yield block
}
}
/** One model call, fully assembled (drains the chunk stream). */
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

178
packages/llm/src/types.ts Normal file
View File

@@ -0,0 +1,178 @@
/**
* Provider-neutral message and streaming vocabulary.
*
* This is the canonical language spoken by the agent loop, session logs, and
* every plugin. Adapters translate it to provider wire formats (DeepSeek V4
* first); nothing outside an adapter should ever see a provider-specific
* shape.
*
* Extensibility: the unions in this file are derived from interfaces
* (`ContentBlockMap`, `MessageSourceMap`, `FinishReasonMap`) so that plugins
* can extend them via declaration merging:
*
* ```ts
* declare module '@deepseek-ai/dsh-llm' {
* interface ContentBlockMap {
* video: { type: 'video'; url: string }
* }
* }
* ```
*/
/** Cache hint attached to a content block (provider-interpreted). */
export type CacheHint = 'ephemeral'
/** Plain text visible to the end user. */
export interface TextBlock {
type: 'text'
text: string
cache?: CacheHint
}
/** Reasoning / thinking content, distinct from visible text. */
export interface ReasoningBlock {
type: 'reasoning'
text: string
}
/** A tool invocation requested by the model. */
export interface ToolCallBlock {
type: 'tool-call'
/** Provider-issued call id; correlates with the matching tool result. */
id: string
name: string
/** Raw JSON string as produced by the model. */
arguments: string
}
/** The result of a tool invocation, sent back to the model. */
export interface ToolResultBlock {
type: 'tool-result'
toolCallId: string
content: ContentBlock[]
isError?: boolean
cache?: CacheHint
}
/** An image, by URL or data URL. */
export interface ImageBlock {
type: 'image'
url: string
mimeType?: string
cache?: CacheHint
}
/**
* All known content block shapes, keyed by their `type` tag.
* Merge-extensible: plugins add new block types via declaration merging.
*/
export interface ContentBlockMap {
'text': TextBlock
'reasoning': ReasoningBlock
'tool-call': ToolCallBlock
'tool-result': ToolResultBlock
'image': ImageBlock
}
export type ContentBlockType = keyof ContentBlockMap
export type ContentBlock = ContentBlockMap[ContentBlockType]
/** A single message in a conversation history. */
export interface Message {
role: 'system' | 'user' | 'assistant'
content: ContentBlock[]
}
/**
* Where a message (or injected content) came from.
* Merge-extensible sum type — plugins add their own `kind`s.
*/
export interface MessageSourceMap {
user: { kind: 'user' }
plugin: { kind: 'plugin'; plugin: string }
agent: { kind: 'agent'; agentId: string }
}
export type MessageSource = MessageSourceMap[keyof MessageSourceMap]
/**
* Why a model response stopped.
* Merge-extensible so adapters can surface provider-specific reasons.
*/
export interface FinishReasonMap {
'stop': { kind: 'stop' }
'tool-calls': { kind: 'tool-calls' }
'max-tokens': { kind: 'max-tokens' }
'aborted': { kind: 'aborted' }
'error': { kind: 'error'; message: string; code?: string }
}
export type FinishReason = FinishReasonMap[keyof FinishReasonMap]
/** Token accounting for one model call (cache fields are optional). */
export interface TokenUsage {
inputTokens: number
outputTokens: number
cacheReadTokens?: number
cacheWriteTokens?: number
reasoningTokens?: number
}
/**
* Raw streaming protocol emitted by adapters.
*
* A streaming response interleaves several typed blocks (text, reasoning,
* multiple tool calls); `index` ties each delta to its block, and `block-end`
* carries the fully-assembled ContentBlock so consumers don't have to
* re-assemble deltas themselves (use {@link BlockAssembler} when they do).
*
* TODO(review): this protocol needs careful review before the first real
* adapter lands (DeepSeek V4 wire format, partial JSON arguments, interleaved
* reasoning signatures, …).
*/
export type StreamChunk =
| { type: 'block-start'; index: number; blockType: ContentBlockType }
| { type: 'text-delta'; index: number; text: string }
| { type: 'reasoning-delta'; index: number; text: string }
| { type: 'tool-call-delta'; index: number; id: string; name?: string; argumentsDelta: string }
| { type: 'block-end'; index: number; block: ContentBlock }
| { type: 'usage'; usage: TokenUsage }
| { type: 'finish'; reason: FinishReason }
/**
* JSON-schema description of a tool, as sent to the model.
*
* Declared here (not in dsh-tools) because it is part of {@link GenerateOptions};
* dsh-tools' ToolDefinition and dsh-system-prompt's PromptAssembly both import
* it from this package.
*/
export interface ToolSchema {
name: string
description: string
/** JSON Schema object for the arguments. */
parameters: Record<string, unknown>
strict?: boolean
}
/** A single model request, fully assembled. */
export interface GenerateOptions {
model: string
messages: Message[]
/** System prompt text (adapters map to the provider's system slot). */
system?: string
/** Tool schemas (adapters map to the provider's `tools` field). */
tools?: ToolSchema[]
/** Assistant prefix continuation (prefill). */
prefill?: ContentBlock[]
temperature?: number
maxTokens?: number
stop?: string[]
signal?: AbortSignal
}
/** Non-streaming result, assembled from the chunk stream. */
export interface GenerateResult {
message: Message
usage?: TokenUsage
finish: FinishReason
}