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

Fix architecture-review findings in the loop and service packages

Browse Source 
High (loop pipeline): agent/step-result now runs before the
assistant/message append so the session log records what tool dispatch
actually uses; abort is honored between tool calls, not just
mid-stream; steering drains at step start, pending steering overrides
a negative turn-continuation decision (/goal pattern), and leftover
steering is re-enqueued as queued messages so it is never stranded;
exceptions from turn-continuation listeners and session/flush are
contained to the turn (error event + agent/error) instead of killing
the driver loop.

Medium: disposal emits agent/status('disposed') and mid-turn disposal
records reason 'disposed'; duplicate LLM adapter registration throws
(all-or-nothing); SessionEvent is a real discriminated union (casts
removed); model-less agents fail with a clear actionable error unless
agent/request supplies a model.

Low: agent/queued and agent/steering carry the resolved MessageSource;
streamBlocks() yields strictly in stream order and flushes delta-only
blocks (matches generate()); BlockAssembler freezes blocks on
block-end and ignores stragglers from malformed streams; turn
numbering is a counter seeded from the log (fork-safe); LoopAgent's
stop disposer is infallible (a throwing status listener cannot skip
registry cleanup); AgentLoop.create uses a generator effect so stop
and unregister are independent disposables; SessionStore wires
onAppend inside its effect.

21 regression tests added (review-fixes.spec.ts), organized by
finding. Docs updated: loop pseudocode (status emissions, ordering,
error containment, steering guarantees) and waterfall composition
caveat in docs/architecture.md; AGENTS.md notes that excessive tests
are welcome.
This commit is contained in:
Tianyi Cui
2026-06-11 12:18:52 +08:00
parent cacfae3cae
commit 217b8ec0e2
12 changed files with 887 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -5,3 +5,4 @@ lib/
.yarn/ .yarn/
yarn-error.log yarn-error.log
examples/*/*.jsonl examples/*/*.jsonl
.claude/

5
AGENTS.md
View File

@@ -73,7 +73,10 @@ only needed for publishing/consumption outside the repo.
docs/architecture.md). Changing `agent-loop` requires updating that doc. docs/architecture.md). Changing `agent-loop` requires updating that doc.
- **Tests**: vitest, colocated under `packages/<name>/tests/*.spec.ts`. Every - **Tests**: vitest, colocated under `packages/<name>/tests/*.spec.ts`. Every
registry needs an HMR-safety test (dispose the contributing fiber, assert registry needs an HMR-safety test (dispose the contributing fiber, assert
cleanup). cleanup). **Excessive tests are welcome** — when in doubt, write the test;
err on the side of covering edge cases, error paths, event ordering, and
concurrency races even if they seem unlikely. Review findings get regression
tests (see `packages/agent-loop/tests/review-fixes.spec.ts`).
## Vendoring Policy ## Vendoring Policy

31
docs/architecture.md
View File

@@ -154,28 +154,40 @@ deferred.
``` ```
forever: forever:
wait for queued messages (idle) wait for queued messages (idle)
TURN: drain queued → session('user/message'…) → 'turn/start' → emit agent/turn-start emit agent/status(running)
TURN (error-contained — a throwing plugin ends the turn, never the loop):
drain queued → session('user/message'…) → 'turn/start' → emit agent/turn-start
STEP loop: STEP loop:
drain steering (late steering from previous step's listeners)
emit agent/step-start emit agent/step-start
assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble
req = {model, system, tools, messages: session.deriveMessages(), signal} req = {model, system, tools, messages: session.deriveMessages(), signal}
req = waterfall agent/request ⟵ hooks, compaction, model switch req = waterfall agent/request ⟵ hooks, compaction, model switch
stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks) stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
session('assistant/chunk'); emit agent/stream-chunk session('assistant/chunk'); emit agent/stream-chunk
session('assistant/message', 'usage') msg = waterfall agent/step-result ⟵ runs BEFORE the log append, so the
msg = waterfall agent/step-result ⟵ post-process before tool dispatch session('assistant/message', 'usage') log records what tool dispatch uses
each tool-call (sequential): each tool-call (sequential, abort-checked between calls):
session('tool/call'); ctx.tools.execute() ⟵ waterfall tools/execute session('tool/call'); ctx.tools.execute() ⟵ waterfall tools/execute
session('tool/result') session('tool/result')
drain steering → session('steering/message'); emit agent/steering drain steering → session('steering/message'); emit agent/steering
emit agent/step-end emit agent/step-end
cont = waterfall agent/turn-continuation(default = hadToolCalls || steered) cont = waterfall agent/turn-continuation(default = hadToolCalls || steered)
steering pending from step-end/continuation listeners forces cont = true
if !cont: break if !cont: break
session('turn/end'); emit agent/turn-end session('turn/end'); emit agent/turn-end
await ctx.parallel('session/flush', session) ⟵ durability checkpoint await ctx.parallel('session/flush', session) ⟵ durability checkpoint (failure
idle (emit agent/status) unless more queued reported via agent/error, not fatal)
leftover steering re-enqueued as queued messages ⟵ steering is never stranded
emit agent/status(idle) unless more queued
``` ```
Error containment: a throwing `agent/turn-continuation` listener or a
rejecting `session/flush` ends the **turn** with an `error` event — never the
driver loop. `abort()` is honored mid-stream **and** between tool calls;
disposal mid-turn ends the turn with reason `disposed` and emits
`agent/status('disposed')`.
### Event taxonomy ### Event taxonomy
Declared in `@deepseek-ai/dsh-agent` (so nothing depends on the loop package). Declared in `@deepseek-ai/dsh-agent` (so nothing depends on the loop package).
@@ -206,8 +218,11 @@ receives `(...args, next)`:
- return a value **without** calling `next()` to short-circuit (veto); - return a value **without** calling `next()` to short-circuit (veto);
- listeners run in registration order; `prepend: true` jumps the queue. - listeners run in registration order; `prepend: true` jumps the queue.
Mutate the passed-in object (e.g. `options.model = '…'`) or return a Composition caveat: values propagate through `next()`'s **return value**.
replacement value — both compose. Mutating the passed-in object works when later listeners receive the same
reference, but a listener that returns a *new* object makes earlier mutations
invisible downstream. Prefer mutate-then-`next()` for cooperative middleware;
return a replacement only when you mean to take over the result.
## Plugin sanity checklist ## Plugin sanity checklist

29
packages/agent-loop/src/agent.ts
View File

@@ -1,7 +1,7 @@
import type { Context } from 'cordis' import type { Context } from 'cordis'
import type { AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent' import type { AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent' import type { Agent } from '@deepseek-ai/dsh-agent'
import type { ContentBlock } from '@deepseek-ai/dsh-llm' import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import type { Session } from '@deepseek-ai/dsh-session' import type { Session } from '@deepseek-ai/dsh-session'
import { Inbox } from './inbox.ts' import { Inbox } from './inbox.ts'
import { runLoop } from './loop.ts' import { runLoop } from './loop.ts'
@@ -44,25 +44,28 @@ export class LoopAgent implements Agent {
this.ctx.emit('agent/status', this, status) this.ctx.emit('agent/status', this, status)
} }
private resolveSource(options?: SendOptions): MessageSource {
return options?.source ?? { kind: 'user' }
}
send(content: ContentBlock[], options?: SendOptions): void { send(content: ContentBlock[], options?: SendOptions): void {
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`) if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
const source = options?.source ?? { kind: 'user' as const } const source = this.resolveSource(options)
this.inbox.enqueue({ content, source }) this.inbox.enqueue({ content, source })
this.ctx.emit('agent/queued', this, content, { ...options, steering: false }) this.ctx.emit('agent/queued', this, content, { source, steering: false })
} }
steer(content: ContentBlock[], options?: SendOptions): void { steer(content: ContentBlock[], options?: SendOptions): void {
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`) if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
if (this._status !== 'running') return this.send(content, options) if (this._status !== 'running') return this.send(content, options)
const source = options?.source ?? { kind: 'user' as const } const source = this.resolveSource(options)
this.inbox.steer({ content, source }) this.inbox.steer({ content, source })
this.ctx.emit('agent/queued', this, content, { ...options, steering: true }) this.ctx.emit('agent/queued', this, content, { source, steering: true })
} }
inject(content: ContentBlock[], options?: SendOptions): void { inject(content: ContentBlock[], options?: SendOptions): void {
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`) if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
const source = options?.source ?? { kind: 'user' as const } this.session.append('context/message', { content, source: this.resolveSource(options) })
this.session.append('context/message', { content, source })
} }
abort(reason?: string): void { abort(reason?: string): void {
@@ -77,10 +80,22 @@ export class LoopAgent implements Agent {
disposed: this.disposed, disposed: this.disposed,
isDisposed: () => this._status === 'disposed', isDisposed: () => this._status === 'disposed',
}) })
// The disposer must be infallible: it runs inside the fiber's LIFO
// disposal chain, where a throw would skip later disposers (e.g. the
// registry unregistration) and leave `done` pending forever.
return () => { return () => {
if (this._status === 'disposed') return
this._status = 'disposed' this._status = 'disposed'
this.resolveDisposed() this.resolveDisposed()
this.currentAbort?.abort('disposed') this.currentAbort?.abort('disposed')
// setStatus refuses transitions out of 'disposed', so emit directly —
// 'disposed' is part of the agent/status contract. Guarded: a throwing
// listener must not break the disposal chain.
try {
this.ctx.emit('agent/status', this, 'disposed')
} catch {
// listener error during disposal — nothing safe left to do with it
}
} }
} }
} }

14
packages/agent-loop/src/index.ts
View File

@@ -59,14 +59,12 @@ export class AgentLoop extends Service {
create(id: string, options: AgentOptions = {}): LoopAgent { create(id: string, options: AgentOptions = {}): LoopAgent {
const session = this.ctx.sessions.create(`${id}-session`) const session = this.ctx.sessions.create(`${id}-session`)
const agent = new LoopAgent(this.ctx, id, options, session) const agent = new LoopAgent(this.ctx, id, options, session)
this.ctx.effect(() => { // Generator effect: stop and unregister are independent disposables
const stop = agent.start() // (LIFO), so a throwing stop() cannot leak the registry entry.
const unregister = this.ctx.agents.register(agent) this.ctx.effect(function* (this: AgentLoop) {
return () => { yield this.ctx.agents.register(agent)
stop() yield agent.start()
unregister() }.bind(this), 'agentLoop.create()')
}
}, 'agentLoop.create()')
return agent return agent
} }
} }

249
packages/agent-loop/src/loop.ts
View File

@@ -1,7 +1,7 @@
import type { Context } from 'cordis' import type { Context } from 'cordis'
import type { GenerateOptions, Message } from '@deepseek-ai/dsh-llm' import type { GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
import { BlockAssembler } from '@deepseek-ai/dsh-llm' import { BlockAssembler } from '@deepseek-ai/dsh-llm'
import type { TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session' import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt' import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools' import type {} from '@deepseek-ai/dsh-tools'
import type { LoopAgent } from './agent.ts' import type { LoopAgent } from './agent.ts'
@@ -20,110 +20,178 @@ export interface LoopHandle {
* ``` * ```
* forever: * forever:
* wait for queued messages (idle) * wait for queued messages (idle)
* TURN: drain queued → session('turn/start') → emit agent/turn-start * TURN (error-contained — a throwing plugin ends the turn, never the loop):
* drain queued → session('user/message'…) → 'turn/start' → emit agent/turn-start
* STEP loop: * STEP loop:
* drain steering → session('steering/message') ⟵ catches late steering
* emit agent/step-start * emit agent/step-start
* assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble * assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble
* req = {model, system, tools, messages: session.deriveMessages(), signal} * req = {model, system, tools, messages: session.deriveMessages(), signal}
* req = waterfall agent/request ⟵ hooks/compaction/model-switch * req = waterfall agent/request ⟵ hooks/compaction/model-switch
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks) * stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
* session('assistant/chunk'); emit agent/stream-chunk; assembler.push * session('assistant/chunk'); emit agent/stream-chunk
* session('assistant/message','usage') * msg = waterfall agent/step-result ⟵ BEFORE the log append, so the
* msg = waterfall agent/step-result ⟵ post-process before tool dispatch * session('assistant/message','usage') session records what actually ran
* each tool-call (sequential): * each tool-call in msg (sequential, abort-checked):
* session('tool/call'); ctx.tools.execute() ⟵ waterfall tools/execute * session('tool/call'); ctx.tools.execute() ⟵ waterfall tools/execute
* session('tool/result') * session('tool/result')
* drain steering → session('steering/message'); emit agent/steering * drain steering → session('steering/message'); emit agent/steering
* emit agent/step-end * emit agent/step-end
* cont = waterfall agent/turn-continuation(default = hadToolCalls || steered) * cont = waterfall agent/turn-continuation(default = hadToolCalls || steered)
* if !cont && steering arrived from step-end/continuation listeners: cont = true
* if !cont: break
* session('turn/end'); emit agent/turn-end * session('turn/end'); emit agent/turn-end
* await ctx.parallel('session/flush', session) ⟵ durability checkpoint * await ctx.parallel('session/flush', session) ⟵ durability checkpoint
* re-enqueue leftover steering as queued ⟵ steering is never stranded
* idle (emit agent/status) unless more queued
* ``` * ```
*/ */
export async function runLoop(ctx: Context, agent: LoopAgent, handle: LoopHandle): Promise<void> { export async function runLoop(ctx: Context, agent: LoopAgent, handle: LoopHandle): Promise<void> {
const { session } = agent const { session } = agent
let turn = lastTurnNumber(session) // seeded/forked sessions continue numbering
while (!handle.isDisposed()) { while (!handle.isDisposed()) {
await agent.inbox.waitForQueued(handle.disposed) await agent.inbox.waitForQueued(handle.disposed)
if (handle.isDisposed()) break if (handle.isDisposed()) break
handle.setStatus('running') handle.setStatus('running')
const turn = nextTurnNumber(session) turn += 1
try {
// Drain queued messages into the session — they trigger this turn. await runTurn(ctx, agent, handle, turn)
const queued = agent.inbox.drainQueued() } catch (error: any) {
const trigger: TurnTrigger = { kind: 'message', source: queued[0]!.source } // Backstop: a throwing emit listener (turn boundaries) or a broken
for (const message of queued) { // finalizer must not kill the driver. Record what we can and move on.
session.append('user/message', { content: message.content, source: message.source })
}
session.append('turn/start', { turn, trigger })
ctx.emit('agent/turn-start', agent, turn)
let reason: TurnEndReason = { kind: 'completed' }
let step = 0
while (true) {
step += 1
ctx.emit('agent/step-start', agent, turn, step)
session.append('step/start', { turn, step })
const abort = new AbortController()
handle.setAbort(abort)
let stepOutcome: { hadToolCalls: boolean } | { error: Error }
try { try {
stepOutcome = await runStep(ctx, agent, turn, step, abort.signal) const err = error instanceof Error ? error : new Error(String(error))
} catch (error: any) { session.append('error', { turn, step: 0, message: err.message, code: (err as any).code })
stepOutcome = { error: error instanceof Error ? error : new Error(String(error)) } ctx.emit('agent/error', agent, turn, 0, err)
} finally { } catch { /* the error path itself is broken; nothing left to do */ }
handle.setAbort(undefined)
}
// Steering arrives between steps: drain before deciding continuation
// so the decision (and the next request) sees it.
const steered = agent.inbox.drainSteering()
for (const message of steered) {
session.append('steering/message', { turn, content: message.content, source: message.source })
ctx.emit('agent/steering', agent, turn, message.content)
}
session.append('step/end', { turn, step })
ctx.emit('agent/step-end', agent, turn, step)
if ('error' in stepOutcome) {
const { error } = stepOutcome
if (abort.signal.aborted || handle.isDisposed()) {
reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
} else {
session.append('error', { turn, step, message: error.message, code: (error as any).code })
ctx.emit('agent/error', agent, turn, step, error)
reason = { kind: 'error', message: error.message, code: (error as any).code }
}
break
}
const defaultDecision = stepOutcome.hadToolCalls || steered.length > 0
const shouldContinue = await ctx.waterfall(
'agent/turn-continuation', agent, turn, defaultDecision,
async () => defaultDecision,
)
if (!shouldContinue || handle.isDisposed()) break
} }
if (handle.isDisposed() && reason.kind === 'completed') { // Steering that arrived too late to join this turn (turn-end listeners,
reason = { kind: 'disposed' } // flush) becomes a queued message — it must never be stranded.
for (const message of agent.inbox.drainSteering()) {
agent.inbox.enqueue(message)
} }
session.append('turn/end', { turn, reason })
ctx.emit('agent/turn-end', agent, turn, reason)
// Durability checkpoint: persistence plugins drain write-behind buffers.
await ctx.parallel('session/flush', session)
if (!agent.inbox.hasQueued) handle.setStatus('idle') if (!agent.inbox.hasQueued) handle.setStatus('idle')
} }
} }
async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: number): Promise<void> {
const { session } = agent
// Drain queued messages into the session — they trigger this turn.
const queued = agent.inbox.drainQueued()
const trigger: TurnTrigger = { kind: 'message', source: queued[0]!.source }
for (const message of queued) {
session.append('user/message', { content: message.content, source: message.source })
}
session.append('turn/start', { turn, trigger })
ctx.emit('agent/turn-start', agent, turn)
let reason: TurnEndReason = { kind: 'completed' }
let step = 0
while (true) {
step += 1
// Steering from the previous round's step-end/continuation listeners
// (or turn-start listeners on the first step) joins before the request.
drainSteering(ctx, agent, turn)
ctx.emit('agent/step-start', agent, turn, step)
session.append('step/start', { turn, step })
const abort = new AbortController()
handle.setAbort(abort)
let stepOutcome: { hadToolCalls: boolean } | { error: Error }
try {
stepOutcome = await runStep(ctx, agent, turn, step, abort.signal)
} catch (error: any) {
stepOutcome = { error: error instanceof Error ? error : new Error(String(error)) }
} finally {
handle.setAbort(undefined)
}
if ('error' in stepOutcome) {
// Steering that arrived during the failed step stays in the inbox —
// runLoop re-enqueues it as a queued message, so an abort-then-steer
// starts a fresh turn instead of being silently consumed.
session.append('step/end', { turn, step })
ctx.emit('agent/step-end', agent, turn, step)
const { error } = stepOutcome
if (handle.isDisposed()) {
reason = { kind: 'disposed' }
} else if (abort.signal.aborted) {
reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
} else {
session.append('error', { turn, step, message: error.message, code: (error as any).code })
ctx.emit('agent/error', agent, turn, step, error)
reason = { kind: 'error', message: error.message, code: (error as any).code }
}
break
}
// Steering that arrived during streaming/tool execution.
const steered = drainSteering(ctx, agent, turn)
session.append('step/end', { turn, step })
ctx.emit('agent/step-end', agent, turn, step)
const defaultDecision = stepOutcome.hadToolCalls || steered
let shouldContinue: boolean
try {
shouldContinue = await ctx.waterfall(
'agent/turn-continuation', agent, turn, defaultDecision,
async () => defaultDecision,
)
} catch (error: any) {
// A broken continuation plugin ends the turn, not the loop.
const err = error instanceof Error ? error : new Error(String(error))
session.append('error', { turn, step, message: err.message, code: (err as any).code })
ctx.emit('agent/error', agent, turn, step, err)
reason = { kind: 'error', message: err.message, code: (err as any).code }
break
}
// Steering from step-end/continuation listeners (the /goal pattern)
// demands the model see it — it overrides a negative decision; the
// next iteration's drain records it.
if (!shouldContinue && agent.inbox.hasSteering) shouldContinue = true
if (!shouldContinue || handle.isDisposed()) {
if (handle.isDisposed()) reason = { kind: 'disposed' }
break
}
}
session.append('turn/end', { turn, reason })
ctx.emit('agent/turn-end', agent, turn, reason)
// Durability checkpoint: persistence plugins drain write-behind buffers.
// A failing persistence plugin is reported but doesn't kill the agent.
try {
await ctx.parallel('session/flush', session)
} catch (error: any) {
const err = error instanceof Error ? error : new Error(String(error))
session.append('error', { turn, step, message: err.message, code: (err as any).code })
ctx.emit('agent/error', agent, turn, step, err)
}
}
/** Drain the steering queue into the session. Returns whether any arrived. */
function drainSteering(ctx: Context, agent: LoopAgent, turn: number): boolean {
const messages = agent.inbox.drainSteering()
for (const message of messages) {
agent.session.append('steering/message', { turn, content: message.content, source: message.source })
ctx.emit('agent/steering', agent, turn, message.content, message.source)
}
return messages.length > 0
}
/** One step: assemble request → stream model → record → execute tools. */ /** One step: assemble request → stream model → record → execute tools. */
async function runStep( async function runStep(
ctx: Context, ctx: Context,
@@ -141,13 +209,16 @@ async function runStep(
.join('\n\n') .join('\n\n')
let request: GenerateOptions = { let request: GenerateOptions = {
model: options.model ?? 'default', model: options.model ?? '',
messages: session.deriveMessages(), messages: session.deriveMessages(),
system: system || undefined, system: system || undefined,
tools: assembly.tools.length > 0 ? assembly.tools : undefined, tools: assembly.tools.length > 0 ? assembly.tools : undefined,
signal, signal,
} }
request = await ctx.waterfall('agent/request', agent, turn, step, request, async () => request) request = await ctx.waterfall('agent/request', agent, turn, step, request, async () => request)
if (!request.model) {
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
}
// --- Model call (streaming-first; raw chunks are the replay record) --- // --- Model call (streaming-first; raw chunks are the replay record) ---
const assembler = new BlockAssembler() const assembler = new BlockAssembler()
@@ -158,17 +229,23 @@ async function runStep(
assembler.push(chunk) assembler.push(chunk)
} }
// The step-result waterfall runs BEFORE the session append so the log (the
// source of truth for derived history and replay) records the message that
// tool dispatch actually uses.
let message: Message = assembler.message() let message: Message = assembler.message()
message = await ctx.waterfall('agent/step-result', agent, turn, step, message, async () => message)
session.append('assistant/message', { turn, step, content: message.content }) session.append('assistant/message', { turn, step, content: message.content })
if (assembler.usage) { if (assembler.usage) {
session.append('usage', { turn, step, usage: assembler.usage }) session.append('usage', { turn, step, usage: assembler.usage })
} }
message = await ctx.waterfall('agent/step-result', agent, turn, step, message, async () => message)
// --- Tool execution (sequential; parallel execution is a TODO) --- // --- Tool execution (sequential; parallel execution is a TODO) ---
// ToolRegistry.execute converts tool failures (including aborts) into
// isError results, so abort is re-checked around every call here.
const toolCalls = message.content.filter(block => block.type === 'tool-call') const toolCalls = message.content.filter(block => block.type === 'tool-call')
for (const call of toolCalls) { for (const call of toolCalls) {
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments }) session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
let parsedArguments: unknown let parsedArguments: unknown
try { try {
@@ -189,17 +266,17 @@ async function runStep(
content: result.content, content: result.content,
isError: result.isError, isError: result.isError,
}) })
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
} }
return { hadToolCalls: toolCalls.length > 0 } return { hadToolCalls: toolCalls.length > 0 }
} }
function nextTurnNumber(session: LoopAgent['session']): number { /** The last turn number in a (possibly seeded) session log, or 0. */
function lastTurnNumber(session: Session): number {
for (let index = session.events.length - 1; index >= 0; index--) { for (let index = session.events.length - 1; index >= 0; index--) {
const event = session.events[index] const event = session.events[index]
if (event.type === 'turn/start') { if (event.type === 'turn/start') return event.data.turn
return (event.data as { turn: number }).turn + 1
}
} }
return 1 return 0
} }

547
packages/agent-loop/tests/review-fixes.spec.ts Normal file
View File

@@ -0,0 +1,547 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { Session, SessionEvent } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import AgentLoop, { LoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
/**
* Regression tests for the findings of the first architecture review
* (Codex + sub-agent, post phase-1). Each describe block names the finding.
*/
async function harness(adapter: MockAdapter) {
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(AgentLoop, { agents: [] })
ctx.llm.registerAdapter(['mock'], adapter)
return ctx
}
function waitForIdle(ctx: Context, agent: LoopAgent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
}
})
})
}
function send(agent: LoopAgent, text: string) {
agent.send([{ type: 'text', text }])
}
describe('HIGH: session log records what agent/step-result actually produced', () => {
it('a step-result rewrite is what the log, derived history, and tool dispatch all see', async () => {
const adapter = new MockAdapter([textResponse('original'), textResponse('done')])
const ctx = await harness(adapter)
const executed: string[] = []
ctx.tools.register({
name: 'injected-tool',
description: '',
parameters: { type: 'object' },
async execute() {
executed.push('injected-tool')
return [{ type: 'text', text: 'ran' }]
},
})
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
// Plugin rewrites the message: replaces the text AND adds a tool call.
let rewritten = false
ctx.on('agent/step-result', async (_agent, _turn, _step, message, next) => {
if (rewritten) return next()
rewritten = true
return {
role: 'assistant' as const,
content: [
{ type: 'text' as const, text: 'rewritten' },
{ type: 'tool-call' as const, id: 'c-injected', name: 'injected-tool', arguments: '{}' },
],
}
})
send(agent, 'go')
await waitForIdle(ctx, agent)
// the injected tool call was dispatched…
expect(executed).toEqual(['injected-tool'])
// …and the session log recorded the REWRITTEN message, not the original
const recorded = agent.session.events.find(e => e.type === 'assistant/message')!
expect(JSON.stringify(recorded.data)).toContain('rewritten')
expect(JSON.stringify(recorded.data)).not.toContain('original')
// tool/call + tool/result correlate with the injected call id
const callEvent = agent.session.events.find(e => e.type === 'tool/call')!
expect((callEvent.data as any).callId).toBe('c-injected')
// derived history shows the rewritten message (replay-correct)
const derived = agent.session.deriveMessages()
expect(JSON.stringify(derived)).toContain('rewritten')
expect(JSON.stringify(derived)).not.toContain('original')
})
})
describe('HIGH: abort during tool execution ends the turn', () => {
it('abort() inside a tool prevents both remaining tools and the next model step', async () => {
const adapter = new MockAdapter([
// model asks for two tool calls in one step
[
{ type: 'block-start', index: 0, blockType: 'tool-call' },
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: 'c1', name: 'aborter', arguments: '{}' } },
{ type: 'block-start', index: 1, blockType: 'tool-call' },
{ type: 'block-end', index: 1, block: { type: 'tool-call', id: 'c2', name: 'second', arguments: '{}' } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
] satisfies StreamChunk[],
textResponse('should never be requested'),
])
const ctx = await harness(adapter)
const executed: string[] = []
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
ctx.tools.register({
name: 'aborter',
description: '',
parameters: { type: 'object' },
async execute() {
executed.push('aborter')
agent.abort('user interrupt')
return [{ type: 'text', text: 'done' }]
},
})
ctx.tools.register({
name: 'second',
description: '',
parameters: { type: 'object' },
async execute() {
executed.push('second')
return [{ type: 'text', text: 'done' }]
},
})
const reasons: any[] = []
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(executed).toEqual(['aborter']) // second tool never ran
expect(adapter.requests).toHaveLength(1) // no follow-up model call
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
})
})
describe('HIGH: steering from late extension points is never stranded', () => {
it('steer() from an agent/step-end listener reaches the next request (/goal pattern)', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'echo', { text: 'x' }),
textResponse('after steering'),
])
const ctx = await harness(adapter)
ctx.tools.register({
name: 'echo',
description: '',
parameters: { type: 'object' },
async execute(args: any) {
return [{ type: 'text', text: String(args.text) }]
},
})
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
let steeredOnce = false
ctx.on('agent/step-end', () => {
if (steeredOnce) return
steeredOnce = true
agent.steer([{ type: 'text', text: 'goal reminder from step-end' }])
})
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[1].messages)).toContain('goal reminder from step-end')
})
it('steer() from an agent/turn-continuation listener overrides a stop decision', async () => {
const adapter = new MockAdapter([
textResponse('no tools, would stop here'),
textResponse('continued because of steering'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
let steeredOnce = false
ctx.on('agent/turn-continuation', async (_agent, _turn, _decision, next) => {
if (!steeredOnce) {
steeredOnce = true
agent.steer([{ type: 'text', text: 'one more thing' }])
}
return next()
})
send(agent, 'go')
await waitForIdle(ctx, agent)
// the default decision was false (no tools), but steering forced step 2
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[1].messages)).toContain('one more thing')
})
it('steer() from an agent/turn-end listener becomes a queued message for the next turn', async () => {
const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
let steeredOnce = false
ctx.on('agent/turn-end', () => {
if (steeredOnce) return
steeredOnce = true
agent.steer([{ type: 'text', text: 'too late for this turn' }])
})
const turns: number[] = []
ctx.on('agent/turn-start', (_agent, turn) => void turns.push(turn))
send(agent, 'go')
await waitForIdle(ctx, agent)
// the loop chains directly into turn 2 (status never returns to idle in
// between), so the first idle transition means both turns are complete
expect(turns).toEqual([1, 2])
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[1].messages)).toContain('too late for this turn')
})
it('steering queued during an aborted step is re-delivered, not silently consumed', async () => {
const adapter = new MockAdapter(['hang', textResponse('recovered')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
agent.steer([{ type: 'text', text: 'redirect' }])
agent.abort('user interrupt')
await waitForIdle(ctx, agent)
// a new turn ran with the steering content delivered as a message
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[1].messages)).toContain('redirect')
})
})
describe('HIGH: plugin exceptions are contained', () => {
it('a throwing agent/turn-continuation listener ends the turn with an error, loop survives', async () => {
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
let threwOnce = false
ctx.on('agent/turn-continuation', async (): Promise<boolean> => {
if (!threwOnce) {
threwOnce = true
throw new Error('broken continuation plugin')
}
return false
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
send(agent, 'first')
await waitForIdle(ctx, agent)
expect(errors.map(e => e.message)).toEqual(['broken continuation plugin'])
// the loop is still alive: a second send works normally
send(agent, 'second')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(2)
expect(agent.status).toBe('idle')
})
it('a rejecting session/flush listener is reported but does not kill the agent', async () => {
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
let rejectedOnce = false
ctx.on('session/flush', async () => {
if (!rejectedOnce) {
rejectedOnce = true
throw new Error('disk full')
}
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
send(agent, 'first')
await waitForIdle(ctx, agent)
expect(errors.map(e => e.message)).toEqual(['disk full'])
send(agent, 'second')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(2)
})
})
describe('MEDIUM: disposed status is part of the agent/status contract', () => {
it('disposing the fiber emits agent/status(disposed) and ends the turn with reason disposed', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
let agent!: LoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create('scoped', { model: 'mock' })
}, { inject: ['agentLoop'] }))
const statuses: string[] = []
const reasons: any[] = []
ctx.on('agent/status', (_agent, status) => void statuses.push(status))
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
await fiber.dispose()
await agent.done
expect(statuses).toEqual(['running', 'disposed'])
expect(reasons).toEqual([{ kind: 'disposed' }])
})
it('a throwing agent/status listener cannot break disposal or leak the registry entry', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
let agent!: LoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create('scoped', { model: 'mock' })
}, { inject: ['agentLoop'] }))
ctx.on('agent/status', (_agent, status) => {
if (status === 'disposed') throw new Error('broken status listener')
})
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
await fiber.dispose()
await agent.done // must not hang
expect(agent.status).toBe('disposed')
expect(ctx.agents.get('scoped')).toBeUndefined() // unregistered despite the throw
})
})
describe('MEDIUM: misc registry and config fixes', () => {
it('duplicate adapter registration is rejected', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const adapter = new MockAdapter([])
ctx.llm.registerAdapter(['m1'], adapter)
expect(() => ctx.llm.registerAdapter(['m1'], new MockAdapter([])))
.toThrow('already registered')
// the original registration survives the failed attempt
expect(ctx.llm.models()).toEqual(['m1'])
})
it('an agent without a model fails the step with a clear error (not NO_ADAPTER for "default")', async () => {
const adapter = new MockAdapter([textResponse('never')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', {}) // no model
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(errors).toHaveLength(1)
expect(errors[0].message).toContain('has no model')
expect(errors[0].message).toContain('agent/request')
})
it('the agent/request waterfall can supply the model for a model-less agent', async () => {
const adapter = new MockAdapter([textResponse('routed')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', {}) // no model — router plugin decides
ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
options.model = 'mock'
return next()
})
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(1)
expect(agent.session.deriveMessages().at(-1)?.content).toEqual([{ type: 'text', text: 'routed' }])
})
it('agent/queued carries the resolved source; agent/steering carries its source', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'noop', {}), textResponse('done')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
ctx.tools.register({
name: 'noop',
description: '',
parameters: { type: 'object' },
async execute() {
agent.steer([{ type: 'text', text: 's' }], { source: { kind: 'plugin', plugin: 'goal' } })
return []
},
})
const queuedSources: any[] = []
const steeringSources: any[] = []
ctx.on('agent/queued', (_agent, _content, info) => void queuedSources.push(info))
ctx.on('agent/steering', (_agent, _turn, _content, source) => void steeringSources.push(source))
send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
await waitForIdle(ctx, agent)
expect(queuedSources[0]).toEqual({ source: { kind: 'user' }, steering: false })
expect(queuedSources[1]).toEqual({ source: { kind: 'plugin', plugin: 'goal' }, steering: true })
expect(steeringSources).toEqual([{ kind: 'plugin', plugin: 'goal' }])
})
})
describe('MEDIUM: turn numbering continues across seeded (forked) sessions', () => {
it('a forked agent continues turn numbers after the seed log', async () => {
const first = new MockAdapter([textResponse('turn one')])
const ctx = await harness(first)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
send(agent, 'first')
await waitForIdle(ctx, agent)
// fork: seed a second context's agent with the first session's log
const second = new MockAdapter([textResponse('turn two')])
const ctx2 = new Context()
await ctx2.plugin(LlmService)
await ctx2.plugin(SessionStore)
await ctx2.plugin(SystemPrompt)
await ctx2.plugin(ToolRegistry)
await ctx2.plugin(AgentRegistry)
await ctx2.plugin(AgentLoop, { agents: [] })
ctx2.llm.registerAdapter(['mock'], second)
const seeded = ctx2.sessions.create('forked', [...agent.session.events])
const forked = new LoopAgent(ctx2, 'forked-agent', { model: 'mock' }, seeded)
ctx2.effect(() => forked.start())
const turns: number[] = []
ctx2.on('agent/turn-start', (_agent, turn) => void turns.push(turn))
forked.send([{ type: 'text', text: 'continue' }])
await new Promise<void>((resolve) => {
ctx2.on('agent/status', (subject, status) => {
if (subject === forked && status === 'idle') resolve()
})
})
expect(turns).toEqual([2])
})
})
describe('LOW: BlockAssembler and streamBlocks edge cases', () => {
it('ignores deltas arriving after block-end for the same index (malformed stream)', async () => {
const { BlockAssembler } = await import('@deepseek-ai/dsh-llm')
const assembler = new BlockAssembler()
assembler.push({ type: 'block-start', index: 0, blockType: 'text' })
assembler.push({ type: 'text-delta', index: 0, text: 'good' })
assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'good' } })
assembler.push({ type: 'text-delta', index: 0, text: ' straggler' })
expect(assembler.blocks()).toEqual([{ type: 'text', text: 'good' }])
})
it('assembles tool-call blocks from deltas without block-end', async () => {
const { BlockAssembler } = await import('@deepseek-ai/dsh-llm')
const assembler = new BlockAssembler()
assembler.push({ type: 'tool-call-delta', index: 0, id: 'c9', name: 'echo', argumentsDelta: '{"a"' })
assembler.push({ type: 'tool-call-delta', index: 0, id: 'c9', argumentsDelta: ':1}' })
expect(assembler.blocks()).toEqual([
{ type: 'tool-call', id: 'c9', name: 'echo', arguments: '{"a":1}' },
])
})
it('streamBlocks flushes delta-only blocks at end of stream (matches generate())', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const deltaOnly: StreamChunk[] = [
{ type: 'text-delta', index: 0, text: 'no ' },
{ type: 'text-delta', index: 0, text: 'block-end' },
{ type: 'finish', reason: { kind: 'stop' } },
]
ctx.llm.registerAdapter(['m'], new MockAdapter([deltaOnly, deltaOnly]))
const blocks: ContentBlock[] = []
for await (const block of ctx.llm.streamBlocks({ model: 'm', messages: [] })) blocks.push(block)
expect(blocks).toEqual([{ type: 'text', text: 'no block-end' }])
const generated = await ctx.llm.generate({ model: 'm', messages: [] })
expect(generated.message.content).toEqual(blocks)
})
it('streamBlocks preserves stream order when an open block precedes a closed one', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
// index 0 never gets block-end (delta-only); index 1 closes mid-stream.
const interleaved: StreamChunk[] = [
{ type: 'text-delta', index: 0, text: 'first, open' },
{ type: 'block-start', index: 1, blockType: 'text' },
{ type: 'text-delta', index: 1, text: 'second, closed' },
{ type: 'block-end', index: 1, block: { type: 'text', text: 'second, closed' } },
{ type: 'finish', reason: { kind: 'stop' } },
]
ctx.llm.registerAdapter(['m'], new MockAdapter([interleaved, interleaved]))
const blocks: ContentBlock[] = []
for await (const block of ctx.llm.streamBlocks({ model: 'm', messages: [] })) blocks.push(block)
expect(blocks).toEqual([
{ type: 'text', text: 'first, open' },
{ type: 'text', text: 'second, closed' },
])
// identical to generate()'s assembled order
const generated = await ctx.llm.generate({ model: 'm', messages: [] })
expect(generated.message.content).toEqual(blocks)
})
it('streamBlocks yields closed blocks incrementally once preceding blocks close', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
const script: StreamChunk[] = [
{ type: 'block-start', index: 0, blockType: 'text' },
{ type: 'text-delta', index: 0, text: 'a' },
{ type: 'block-end', index: 0, block: { type: 'text', text: 'a' } },
{ type: 'block-start', index: 1, blockType: 'text' },
{ type: 'text-delta', index: 1, text: 'b' },
{ type: 'block-end', index: 1, block: { type: 'text', text: 'b' } },
{ type: 'finish', reason: { kind: 'stop' } },
]
ctx.llm.registerAdapter(['m'], new MockAdapter([script]))
const blocks: ContentBlock[] = []
for await (const block of ctx.llm.streamBlocks({ model: 'm', messages: [] })) blocks.push(block)
expect(blocks).toEqual([{ type: 'text', text: 'a' }, { type: 'text', text: 'b' }])
})
})
describe('LOW: discriminated SessionEvent narrows without casts', () => {
it('narrows event.data from event.type', () => {
const session = new Session('s')
const appended: SessionEvent = session.append('tool/call', {
turn: 1, step: 1, callId: 'c1', name: 'echo', arguments: '{}',
})
// compile-time: this switch narrows; runtime: values flow through
switch (appended.type) {
case 'tool/call': {
expect(appended.data.callId).toBe('c1')
expect(appended.data.name).toBe('echo')
break
}
default: throw new Error('wrong narrow')
}
})
})

9
packages/agent/src/types.ts
View File

@@ -69,8 +69,11 @@ declare module 'cordis' {
'agent/disposed'(agent: Agent): void 'agent/disposed'(agent: Agent): void
/** Agent status changed (idle/running/disposed). */ /** Agent status changed (idle/running/disposed). */
'agent/status'(agent: Agent, status: AgentStatus): void 'agent/status'(agent: Agent, status: AgentStatus): void
/** A message entered the agent's inbox (queued or steering). */ /**
'agent/queued'(agent: Agent, content: ContentBlock[], options: SendOptions & { steering: boolean }): void * A message entered the agent's inbox (queued or steering). `source` is
* the resolved source (defaults applied), not the caller's raw options.
*/
'agent/queued'(agent: Agent, content: ContentBlock[], info: { source: MessageSource; steering: boolean }): void
// ---- turn/step boundaries (emit) ---- // ---- turn/step boundaries (emit) ----
'agent/turn-start'(agent: Agent, turn: number): void 'agent/turn-start'(agent: Agent, turn: number): void
@@ -100,7 +103,7 @@ declare module 'cordis' {
/** A raw stream chunk arrived (token-level UI/log feed). */ /** A raw stream chunk arrived (token-level UI/log feed). */
'agent/stream-chunk'(agent: Agent, turn: number, step: number, chunk: StreamChunk): void 'agent/stream-chunk'(agent: Agent, turn: number, step: number, chunk: StreamChunk): void
/** Steering content was injected into a running turn. */ /** Steering content was injected into a running turn. */
'agent/steering'(agent: Agent, turn: number, content: ContentBlock[]): void 'agent/steering'(agent: Agent, turn: number, content: ContentBlock[], source: MessageSource): void
/** A step or turn errored. */ /** A step or turn errored. */
'agent/error'(agent: Agent, turn: number, step: number, error: Error): void 'agent/error'(agent: Agent, turn: number, step: number, error: Error): void
} }

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

@@ -1,5 +1,15 @@
import type { ContentBlock, FinishReason, GenerateResult, Message, StreamChunk, TokenUsage } from './types.ts' import type { ContentBlock, FinishReason, GenerateResult, Message, StreamChunk, TokenUsage } from './types.ts'
interface PartialBlock {
blockType: string
text: string
toolCallId?: string
toolCallName?: string
toolCallArguments: string
/** Set by `block-end` — authoritative, and freezes the partial. */
block?: ContentBlock
}
/** /**
* Incrementally assembles raw {@link StreamChunk}s into complete * Incrementally assembles raw {@link StreamChunk}s into complete
* {@link ContentBlock}s and a final assistant {@link Message}. * {@link ContentBlock}s and a final assistant {@link Message}.
@@ -7,44 +17,45 @@ import type { ContentBlock, FinishReason, GenerateResult, Message, StreamChunk,
* This is the single shared assembly implementation: the agent loop feeds it * This is the single shared assembly implementation: the agent loop feeds it
* while logging raw chunks for replay fidelity, and `LlmService.generate()` / * while logging raw chunks for replay fidelity, and `LlmService.generate()` /
* `streamBlocks()` use it to offer assembled views of the same stream. * `streamBlocks()` use it to offer assembled views of the same stream.
*
* 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
* misbehaving adapter cannot grow memory or corrupt a completed block.
*/ */
export class BlockAssembler { export class BlockAssembler {
private partials = new Map<number, { private partials = new Map<number, PartialBlock>()
blockType: string
text: string
toolCallId?: string
toolCallName?: string
toolCallArguments: string
block?: ContentBlock
}>()
private order: number[] = [] private order: number[] = []
private flushed = 0
private _usage: TokenUsage | undefined private _usage: TokenUsage | undefined
private _finish: FinishReason | undefined private _finish: FinishReason | undefined
/** /**
* Feed one chunk. Returns the completed block when the chunk closes one * Feed one chunk. Returns the completed block when the chunk closes one
* (either an explicit `block-end` or an implicit close), otherwise undefined. * (an explicit `block-end`), otherwise undefined.
*/ */
push(chunk: StreamChunk): ContentBlock | undefined { push(chunk: StreamChunk): ContentBlock | undefined {
switch (chunk.type) { switch (chunk.type) {
case 'block-start': { case 'block-start': {
if (!this.partials.has(chunk.index)) this.order.push(chunk.index) if (!this.partials.has(chunk.index)) {
this.partials.set(chunk.index, { this.order.push(chunk.index)
blockType: chunk.blockType, this.partials.set(chunk.index, {
text: '', blockType: chunk.blockType,
toolCallArguments: '', text: '',
}) toolCallArguments: '',
})
}
return return
} }
case 'text-delta': case 'text-delta':
case 'reasoning-delta': { case 'reasoning-delta': {
const partial = this.ensure(chunk.index, chunk.type === 'text-delta' ? 'text' : 'reasoning') const partial = this.ensure(chunk.index, chunk.type === 'text-delta' ? 'text' : 'reasoning')
if (partial.block) return // closed by block-end; ignore stragglers
partial.text += chunk.text partial.text += chunk.text
return return
} }
case 'tool-call-delta': { case 'tool-call-delta': {
const partial = this.ensure(chunk.index, 'tool-call') const partial = this.ensure(chunk.index, 'tool-call')
if (partial.block) return // closed by block-end; ignore stragglers
partial.toolCallId = chunk.id partial.toolCallId = chunk.id
if (chunk.name) partial.toolCallName = chunk.name if (chunk.name) partial.toolCallName = chunk.name
partial.toolCallArguments += chunk.argumentsDelta partial.toolCallArguments += chunk.argumentsDelta
@@ -66,7 +77,7 @@ export class BlockAssembler {
} }
} }
private ensure(index: number, blockType: string) { private ensure(index: number, blockType: string): PartialBlock {
let partial = this.partials.get(index) let partial = this.partials.get(index)
if (!partial) { if (!partial) {
partial = { blockType, text: '', toolCallArguments: '' } partial = { blockType, text: '', toolCallArguments: '' }
@@ -76,23 +87,57 @@ export class BlockAssembler {
return partial return partial
} }
private assemble(partial: PartialBlock, index: number): ContentBlock {
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}"`)
}
}
/** Assemble all blocks seen so far, in stream order. */ /** Assemble all blocks seen so far, in stream order. */
blocks(): ContentBlock[] { blocks(): ContentBlock[] {
return this.order.map((index) => { return this.order.map(index => this.assemble(this.partials.get(index)!, index))
const partial = this.partials.get(index)! }
if (partial.block) return partial.block
switch (partial.blockType) { /**
case 'text': return { type: 'text', text: partial.text } * Streaming flush: returns (once) every block that is complete AND has no
case 'reasoning': return { type: 'reasoning', text: partial.text } * incomplete block before it in stream order. Call after each `push()`;
case 'tool-call': return { * blocks come out strictly in stream order, so a streaming consumer sees
type: 'tool-call', * exactly the sequence `blocks()` would produce.
id: partial.toolCallId ?? `call-${index}`, */
name: partial.toolCallName ?? '', flushReady(): ContentBlock[] {
arguments: partial.toolCallArguments, const ready: ContentBlock[] = []
} while (this.flushed < this.order.length) {
default: throw new Error(`cannot assemble incomplete block of type "${partial.blockType}"`) const partial = this.partials.get(this.order[this.flushed])!
} 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]
remaining.push(this.assemble(this.partials.get(index)!, index))
this.flushed += 1
}
return remaining
} }
get usage(): TokenUsage | undefined { get usage(): TokenUsage | undefined {

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

@@ -56,6 +56,11 @@ export class LlmService extends Service {
/** Register an adapter for the given model names. Disposed with the fiber. */ /** Register an adapter for the given model names. Disposed with the fiber. */
registerAdapter(models: string[], adapter: LlmAdapter): () => void { registerAdapter(models: string[], adapter: LlmAdapter): () => void {
return this.ctx.effect(() => { return this.ctx.effect(() => {
for (const model of models) {
if (this.adapters.has(model)) {
throw new LlmError(`an adapter for model "${model}" is already registered`, 'DUPLICATE_ADAPTER')
}
}
for (const model of models) this.adapters.set(model, adapter) for (const model of models) this.adapters.set(model, adapter)
this.ctx.emit('llm/adapter-change') this.ctx.emit('llm/adapter-change')
return () => { return () => {
@@ -85,14 +90,19 @@ export class LlmService extends Service {
/** /**
* Stream one model call as completed content blocks — a convenience view * Stream one model call as completed content blocks — a convenience view
* for consumers that don't care about token-level deltas. * 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> { async * streamBlocks(options: GenerateOptions): AsyncIterable<ContentBlock> {
const assembler = new BlockAssembler() const assembler = new BlockAssembler()
for await (const chunk of this.stream(options)) { for await (const chunk of this.stream(options)) {
const block = assembler.push(chunk) assembler.push(chunk)
if (block) yield block yield * assembler.flushReady()
} }
yield * assembler.flushRemaining()
} }
/** One model call, fully assembled (drains the chunk stream). */ /** One model call, fully assembled (drains the chunk stream). */

20
packages/session/src/index.ts
View File

@@ -66,9 +66,9 @@ export class Session {
/** Append one event. Synchronous — the hot path never blocks on I/O. */ /** Append one event. Synchronous — the hot path never blocks on I/O. */
append<T extends SessionEventType>(type: T, data: SessionEventMap[T]): SessionEvent<T> { append<T extends SessionEventType>(type: T, data: SessionEventMap[T]): SessionEvent<T> {
const event: SessionEvent<T> = { type, seq: this.log.length, time: Date.now(), data } const event = { type, seq: this.log.length, time: Date.now(), data } as SessionEvent<T>
this.log.push(event as SessionEvent) this.log.push(event)
this.onAppend?.(event as SessionEvent) this.onAppend?.(event)
return event return event
} }
@@ -87,17 +87,15 @@ export class Session {
for (const event of this.log) { for (const event of this.log) {
switch (event.type) { switch (event.type) {
case 'user/message': { case 'user/message': {
const { content } = event.data as SessionEventMap['user/message'] messages.push({ role: 'user', content: event.data.content })
messages.push({ role: 'user', content })
break break
} }
case 'assistant/message': { case 'assistant/message': {
const { content } = event.data as SessionEventMap['assistant/message'] messages.push({ role: 'assistant', content: event.data.content })
messages.push({ role: 'assistant', content })
break break
} }
case 'tool/result': { case 'tool/result': {
const { callId, content, isError } = event.data as SessionEventMap['tool/result'] const { callId, content, isError } = event.data
messages.push({ messages.push({
role: 'user', role: 'user',
content: [{ type: 'tool-result', toolCallId: callId, content, isError }], content: [{ type: 'tool-result', toolCallId: callId, content, isError }],
@@ -105,12 +103,12 @@ export class Session {
break break
} }
case 'context/message': { case 'context/message': {
const { content, source } = event.data as SessionEventMap['context/message'] const { content, source } = event.data
messages.push({ role: 'user', content: renderTagged('context', content, source) }) messages.push({ role: 'user', content: renderTagged('context', content, source) })
break break
} }
case 'steering/message': { case 'steering/message': {
const { content, source } = event.data as SessionEventMap['steering/message'] const { content, source } = event.data
messages.push({ role: 'user', content: renderTagged('steering', content, source) }) messages.push({ role: 'user', content: renderTagged('steering', content, source) })
break break
} }
@@ -139,8 +137,8 @@ export class SessionStore extends Service {
id ??= `session-${++this.counter}` id ??= `session-${++this.counter}`
if (this.store.has(id)) throw new Error(`session "${id}" already exists`) if (this.store.has(id)) throw new Error(`session "${id}" already exists`)
const session = new Session(id, seed) const session = new Session(id, seed)
session.onAppend = (event) => this.ctx.emit('session/event', session, event)
this.ctx.effect(() => { this.ctx.effect(() => {
session.onAppend = (event) => this.ctx.emit('session/event', session, event)
this.store.set(id, session) this.store.set(id, session)
this.ctx.emit('session/created', session) this.ctx.emit('session/created', session)
return () => { return () => {

25
packages/session/src/types.ts
View File

@@ -63,12 +63,19 @@ export interface SessionEventMap {
export type SessionEventType = keyof SessionEventMap export type SessionEventType = keyof SessionEventMap
/** One immutable entry in the session log. */ /**
export interface SessionEvent<T extends SessionEventType = SessionEventType> { * One immutable entry in the session log.
type: T *
/** Monotonic sequence number within the session. */ * A proper discriminated union over `type` (not independent `type`/`data`
seq: number * unions), so `switch (event.type)` narrows `event.data` without casts.
/** Unix epoch milliseconds. */ */
time: number export type SessionEvent<T extends SessionEventType = SessionEventType> = {
data: SessionEventMap[T] [K in SessionEventType]: {
} type: K
/** Monotonic sequence number within the session. */
seq: number
/** Unix epoch milliseconds. */
time: number
data: SessionEventMap[K]
}
}[T]