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

simplify(llm): drop unconsumed adapter-change event and assembled call surfaces

Browse Source 
The LLM service exposed three call surfaces (stream/streamBlocks/generate) but
the only production consumer — the agent loop — uses stream() exclusively,
feeding raw chunks through its own BlockAssembler for replay fidelity. Drop the
speculative convenience surfaces and the registry-change event that no listener
consumed, leaving stream() as the single model-call contract for both
production and tests.

- Remove LlmService.streamBlocks() and generate(), the llm/generate waterfall,
  and GenerateResult.
- Remove the llm/adapter-change event (declaration + emits) and the
  listener-throw rollback ordering that existed only to protect it; keep the
  HMR rollback disposer.
- Remove BlockAssembler.flushReady()/flushRemaining()/result() and the flushed
  cursor — the streaming-flush slice existed only for streamBlocks().
- Adapter tests drive a stream()+BlockAssembler helper (tests/assemble.ts)
  instead of generate(), exercising the same path production uses.
- Land the AGENTS.md "RFCs are proposals, not golden truth" principle and move
  both RFCs proposed -> implemented.

Implements:
- docs/rfc/implemented/simplification/2026-06-20-drop-unconsumed-llm-adapter-change-event.md
- docs/rfc/implemented/simplification/2026-06-20-drop-unconsumed-llm-assembled-surfaces.md
This commit is contained in:
Tianyi Cui
2026-06-21 01:27:41 +08:00
parent 18f1c010ca
commit 30cd67b8a1
26 changed files with 164 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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)', () => {

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

@@ -10,7 +10,7 @@
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>()
@@ -134,13 +102,9 @@ 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.
// Streaming consumer: push as it goes.
const streaming = new BlockAssembler()
for (const chunk of chunks) {
streaming.push(chunk)
streaming.flushReady()
}
streaming.flushRemaining()
for (const chunk of chunks) streaming.push(chunk)
// One-shot consumer: push all, then read.
const oneShot = feed(chunks)
expect(streaming.usage).toEqual(oneShot.usage)

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([])
})
})