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

refactor(agent-loop): simplify observable state machine

Browse Source
This commit is contained in:
_Kerman
2026-07-24 21:18:48 +08:00
parent fb0ef82aa6
commit b73eb7663c
131 changed files with 2011 additions and 4292 deletions
Download Patch File Download Diff File Expand all files Collapse all files

602
packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -1,17 +1,17 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId, ContentBlock, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
import LlmService, { CallId, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool, TOOL_ABORTED, TOOL_ABORTED_BEFORE_DISPATCH, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
import ToolRegistry, { defineContentToolFixture, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { prepareReactLoopAgent } from '../src/agent.ts'
import { ReactLoopAgent } from '../src/agent.ts'
import InvariantService from '@deepseek-ai/dsh-invariants'
import * as SessionInvariant from '@deepseek-ai/dsh-session/invariant'
import * as AgentInvariant from '@deepseek-ai/dsh-agent/invariant'
import * as AgentLoopInvariant from '@deepseek-ai/dsh-agent-loop/invariant'
import { maxTokensResponse, MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
async function mountInvariants(ctx: Context): Promise<void> {
await ctx.plugin(InvariantService)
@@ -53,59 +53,8 @@ function send(agent: Agent, text: string) {
agent.followup([{ type: 'text', text }])
}
describe('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 original = textResponse('original')
original[original.length - 1] = { type: 'finish', reason: { kind: 'stop' }, replayState: { private: 'original-state' } }
const adapter = new MockAdapter([original, textResponse('done')])
const ctx = await harness(adapter)
const executed: string[] = []
ctx.tools.register(defineTool({
name: 'injected-tool',
description: '',
parameters: {},
async execute() {
executed.push('injected-tool')
return [{ type: 'text', text: 'ran' }]
},
}))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', 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, _signal, 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: CallId('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')
expect(recorded.type === 'assistant/message' && recorded.data.provenance.replayState).toBeUndefined()
// tool/call + tool/result correlate with the injected call id
const callEvent = agent.session.events.find(e => e.type === 'tool/call')!
if (callEvent.type !== 'tool/call') throw new Error('wrong event type')
expect(callEvent.data.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')
})
it('records adapter replay state when step-result preserves the assembled content', async () => {
describe('assistant replay provenance', () => {
it('records adapter replay state with the assembled assistant content', async () => {
const response = textResponse('unchanged')
const replayState = { private: 'state' }
response[response.length - 1] = { type: 'finish', reason: { kind: 'stop' }, replayState }
@@ -116,7 +65,7 @@ describe('session log records what agent/step-result actually produced', () => {
send(agent, 'go')
await waitForIdle(ctx, agent)
const recorded = agent.session.events.find(e => e.type === 'assistant/message')
const recorded = agent.session.events.find(event => event.type === 'assistant/message')
expect(recorded?.type === 'assistant/message' && recorded.data.provenance).toEqual({
provider: 'mock', model: 'next-model', replayState,
})
@@ -124,215 +73,14 @@ describe('session log records what agent/step-result actually produced', () => {
provider: 'mock', model: 'next-model', replayState,
})
})
it('drops adapter replay state when step-result mutates assembled content in place', async () => {
const response = textResponse('original')
response[response.length - 1] = { type: 'finish', reason: { kind: 'stop' }, replayState: { private: 'state' } }
const adapter = new MockAdapter([response])
const ctx = await harness(adapter)
ctx.on('agent/step-result', async (_agent, _turn, _step, message) => {
const block = message.content[0]
if (block?.type === 'text') block.text = 'mutated'
return message
})
const agent = ctx.agentLoop.create(SessionId('mutated-replay-state'), { provider: 'mock', model: 'next-model' })
send(agent, 'go')
await waitForIdle(ctx, agent)
const recorded = agent.session.events.find(event => event.type === 'assistant/message')
expect(recorded?.type === 'assistant/message' && recorded.data.content).toEqual([{ type: 'text', text: 'mutated' }])
expect(recorded?.type === 'assistant/message' && recorded.data.provenance.replayState).toBeUndefined()
})
})
describe('successful provider completion survives agent/step-result failure', () => {
async function expectContentlessCompletionAnchor(
response: StreamChunk[],
id: string,
providerText: string,
): Promise<void> {
const adapter = new MockAdapter([response])
const ctx = await harness(adapter)
await mountInvariants(ctx)
const agent = ctx.agentLoop.create(SessionId(id), { provider: 'mock', model: 'mock' })
const failure = new Error(`${id} result processing failed`)
const reported: Error[] = []
ctx.on('agent/step-result', async () => {
throw failure
})
ctx.on('agent/error', (subject, _turn, _step, error) => {
if (subject === agent) reported.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
const events = [...agent.session.events]
const chunks = events.filter(event => event.type === 'assistant/chunk')
const completions = events.filter(event => event.type === 'assistant/message')
expect(completions).toHaveLength(1)
expect(completions[0]?.type === 'assistant/message' && completions[0].data).toEqual({
turn: 1,
step: 1,
content: [],
provenance: { provider: 'mock', model: 'mock' },
usage: { inputTokens: 10, outputTokens: providerText.length },
})
expect(completions[0]?.sourceEventSeqs).toEqual(chunks.map(event => event.seq))
expect(agent.session.deriveMessages()).toEqual([
{ role: 'user', content: [{ type: 'text', text: 'go' }] },
])
expect(reported).toHaveLength(1)
expect(reported[0]).toBe(failure)
const turnEnd = events.findLast(event => event.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({
kind: 'error',
step: 1,
message: failure.message,
})
}
it('records one content-less anchor when ordinary stop result processing rejects', async () => {
const providerText = 'ordinary provider output'
await expectContentlessCompletionAnchor(
textResponse(providerText),
'a-step-result-stop-failure',
providerText,
)
})
it('records one content-less anchor when max-token result processing rejects', async () => {
const providerText = 'truncated provider output'
await expectContentlessCompletionAnchor(
maxTokensResponse(providerText),
'a-step-result-max-token-failure',
providerText,
)
})
})
describe('abort during tool execution ends the turn', () => {
it('balances a cancelled tool batch through context and post-step before closing', 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: CallId('c1'), name: 'aborter', arguments: '{}' } },
{ type: 'block-start', index: 1, blockType: 'tool-call' },
{ type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('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(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'aborter',
description: '',
parameters: {},
async execute(_args, exec) {
executed.push('aborter')
exec.agent?.steer(
[{ type: 'text', text: 'steering before abort' }],
{ source: { kind: 'plugin', plugin: 'abort-test' } },
)
agent.cancel({ kind: 'user' })
return [{ type: 'text', text: 'done' }]
},
}))
ctx.on('tools/post-execute', async exec => ({
kind: 'accept',
additionalContexts: [{
content: [{ type: 'text', text: `context for ${exec.callId}` }],
source: { kind: 'plugin', plugin: 'abort-test' },
}],
}))
ctx.tools.register(defineTool({
name: 'second',
description: '',
parameters: {},
async execute() {
executed.push('second')
return [{ type: 'text', text: 'done' }]
},
}))
const reasons: TurnEndReason[] = []
const order: string[] = []
ctx.on('session/event', (session, event) => {
if (session !== agent.session) return
switch (event.type) {
case 'assistant/message': order.push('assistant/message'); break
case 'tool/call': order.push(`tool/call:${event.data.callId}`); break
case 'tool/result': {
const outcome = event.data.error?.code === TOOL_ABORTED
|| event.data.error?.code === TOOL_ABORTED_BEFORE_DISPATCH
? 'aborted'
: 'completed'
order.push(`tool/result:${event.data.callId}:${outcome}`)
break
}
case 'context/message': order.push('context/message'); break
case 'steering/message': order.push('steering/message'); break
case 'step/end': order.push('step/end'); break
case 'turn/end': {
reasons.push(event.data.reason)
order.push(`turn/end:${event.data.reason.kind}`)
break
}
}
})
let postSteps = 0
ctx.on('agent/post-step', (subject, turn, step, signal) => {
if (subject !== agent) return
postSteps += 1
expect({ turn, step, aborted: signal.aborted }).toEqual({ turn: 1, step: 1, aborted: true })
order.push('agent/post-step')
})
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(executed).toEqual(['aborter'])
expect(adapter.requests).toHaveLength(1)
expect(postSteps).toBe(1)
expect(order).toEqual([
'assistant/message',
'tool/call:c1',
'tool/result:c1:aborted',
'tool/call:c2',
'tool/result:c2:aborted',
'context/message',
'agent/post-step',
'step/end',
'turn/end:aborted',
])
expect(reasons).toEqual([{ kind: 'aborted' }])
const calls = agent.session.events.filter(event => event.type === 'tool/call')
const results = agent.session.events.filter(event => event.type === 'tool/result')
expect(calls.map(event => event.data.callId)).toEqual([CallId('c1'), CallId('c2')])
expect(results).toHaveLength(2)
expect(results[0]!.data).toMatchObject({
callId: CallId('c1'),
content: [{ type: 'text', text: 'Error: tool call aborted' }],
isError: true,
error: { name: 'AbortError', code: TOOL_ABORTED },
})
expect(results[1]!.data).toMatchObject({
callId: CallId('c2'),
isError: true,
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
})
})
it('records context accepted before a tool-step abort in the same turn', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'aborter', {})])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a-abort-injection'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'aborter',
description: '',
parameters: {},
@@ -355,15 +103,16 @@ describe('abort during tool execution ends the turn', () => {
const events = [...agent.session.events]
expect(events
.filter(event => event.type === 'tool/result' || event.type === 'context/message'
.filter(event => event.type === 'tool/result'
|| (event.type === 'user/message' && event.data.source.kind === 'plugin')
|| event.type === 'step/end' || event.type === 'turn/end')
.map(event => event.type))
.toEqual(['tool/result', 'context/message', 'context/message', 'step/end', 'turn/end'])
.toEqual(['tool/result', 'user/message', 'step/end', 'turn/end'])
expect(events
.filter(event => event.type === 'context/message')
.map(event => event.data.content))
.flatMap(event => event.type === 'user/message' && event.data.source.kind === 'plugin'
? [event.data.content]
: []))
.toEqual([
[{ type: 'text', text: 'accepted before abort' }],
[{ type: 'text', text: 'accepted result context after abort' }],
])
})
@@ -378,7 +127,7 @@ describe('abort during tool execution ends the turn', () => {
] satisfies StreamChunk[]])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a-later-abort-context'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'first',
description: '',
parameters: {},
@@ -386,7 +135,7 @@ describe('abort during tool execution ends the turn', () => {
return [{ type: 'text', text: 'first done' }]
},
}))
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'aborter',
description: '',
parameters: {},
@@ -411,15 +160,19 @@ describe('abort during tool execution ends the turn', () => {
const events = [...agent.session.events]
expect(events
.filter(event => event.type === 'tool/result' || event.type === 'context/message'
.filter(event => event.type === 'tool/result'
|| (event.type === 'user/message' && event.data.source.kind === 'plugin')
|| event.type === 'step/end' || event.type === 'turn/end')
.map(event => event.type))
.toEqual(['tool/result', 'tool/result', 'context/message', 'step/end', 'turn/end'])
expect(events.find(event => event.type === 'context/message')?.data.content)
.toEqual([{ type: 'text', text: 'accepted after first result' }])
.toEqual(['tool/result', 'tool/result', 'step/end', 'turn/end'])
expect(events.flatMap(event =>
event.type === 'user/message' && event.data.source.kind === 'plugin'
? [event.data.content]
: [])[0])
.toBeUndefined()
})
it('drains deferred context before disposal reaches quiescence', async () => {
it('records result context finalized after disposal cancellation', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'waiter', {})])
const ctx = await harness(adapter)
const started = Promise.withResolvers<undefined>()
@@ -427,7 +180,7 @@ describe('abort during tool execution ends the turn', () => {
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(SessionId('a-dispose-injection'), { provider: 'mock', model: 'mock' })
}, { inject: ['agentLoop'] }))
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'waiter',
description: '',
parameters: {},
@@ -456,10 +209,10 @@ describe('abort during tool execution ends the turn', () => {
await fiber.dispose()
expect(agent.session.events
.filter(event => event.type === 'context/message')
.map(event => event.data.content))
.flatMap(event => event.type === 'user/message' && event.data.source.kind === 'plugin'
? [event.data.content]
: []))
.toEqual([
[{ type: 'text', text: 'accepted before disposal' }],
[{ type: 'text', text: 'accepted result context during disposal' }],
])
expect(agent.session.events.find(event => event.type === 'turn/end')?.data.reason)
@@ -479,7 +232,7 @@ describe('abort during tool execution ends the turn', () => {
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a-historical-tool-pair'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'aborter',
description: '',
parameters: {},
@@ -488,7 +241,7 @@ describe('abort during tool execution ends the turn', () => {
return [{ type: 'text', text: 'done' }]
},
}))
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'second',
description: '',
parameters: {},
@@ -499,7 +252,7 @@ describe('abort during tool execution ends the turn', () => {
send(agent, 'leave an unmatched historical call')
await waitForIdle(ctx, agent)
ctx.on('agent/pre-step', (subject, turn) => {
ctx.on('agent/step', (subject, turn) => {
if (subject === agent && turn === 2) {
agent.inject([{ type: 'text', text: 'new turn context' }], { source: { kind: 'plugin', plugin: 'test' } })
}
@@ -507,14 +260,17 @@ describe('abort during tool execution ends the turn', () => {
send(agent, 'start a text-only turn')
await waitForIdle(ctx, agent)
expect(agent.session.events.find(event => event.type === 'context/message')?.data.content)
expect(agent.session.events.flatMap(event =>
event.type === 'user/message' && event.data.source.kind === 'plugin'
? [event.data.content]
: [])[0])
.toEqual([{ type: 'text', text: 'new turn context' }])
expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('new turn context')
})
})
describe('steering from late extension points is never stranded', () => {
it('steer() from an agent/turn-continuation listener overrides a stop decision', async () => {
it('steer() from an agent/stopping listener continues the same turn', async () => {
const adapter = new MockAdapter([
textResponse('no tools, would stop here'),
textResponse('continued because of steering'),
@@ -523,12 +279,11 @@ describe('steering from late extension points is never stranded', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let steeredOnce = false
ctx.on('agent/turn-continuation', async (_agent, _turn, _decision, _signal, next) => {
ctx.on('agent/stopping', () => {
if (!steeredOnce) {
steeredOnce = true
agent.steer([{ type: 'text', text: 'one more thing' }])
}
return next()
})
send(agent, 'go')
@@ -600,22 +355,23 @@ describe('steering from late extension points is never stranded', () => {
})
describe('plugin exceptions are contained', () => {
it('a throwing agent/turn-continuation listener ends the turn with an error, loop survives', async () => {
it('a throwing agent/stopping 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(SessionId('a1'), { provider: 'mock', model: 'mock' })
let threwOnce = false
ctx.on('agent/turn-continuation', async (): Promise<ContinuationDecision> => {
ctx.on('agent/stopping', async () => {
if (!threwOnce) {
threwOnce = true
throw new Error('broken continuation plugin')
}
return { action: 'stop' }
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
ctx.on('agent/error', (_agent, _turn, _step, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'first')
await waitForIdle(ctx, agent)
@@ -628,45 +384,9 @@ describe('plugin exceptions are contained', () => {
expect(agent.status).toBe('idle')
})
it('a rejecting first-turn flush settles before the queued tail starts', async () => {
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const firstFlush = Promise.withResolvers<undefined>()
const releaseFirstFlush = Promise.withResolvers<undefined>()
let flushes = 0
ctx.on('session/flush', async (session) => {
if (session !== agent.session) return
flushes += 1
if (flushes === 1) {
firstFlush.resolve(undefined)
await releaseFirstFlush.promise
throw new Error('disk full')
}
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
const idle = waitForIdle(ctx, agent)
send(agent, 'first')
send(agent, 'second')
await firstFlush.promise
expect(adapter.requests).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
releaseFirstFlush.resolve(undefined)
await idle
expect(errors.map(e => e.message)).toEqual(['disk full'])
expect(adapter.requests).toHaveLength(2)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
})
})
describe('disposed status is part of the agent/status contract', () => {
describe('disposal leaves the two-state status contract balanced', () => {
it('disposing the fiber ends the active turn and never starts its queued tail', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
@@ -687,7 +407,7 @@ describe('disposed status is part of the agent/status contract', () => {
await fiber.dispose()
await driverDone(agent)
expect(statuses).toEqual(['running', 'disposed'])
expect(statuses).toEqual(['running', 'idle'])
expect(reasons).toEqual([{ kind: 'disposed' }])
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
const messages = agent.session.events
@@ -708,7 +428,7 @@ describe('disposed status is part of the agent/status contract', () => {
}, { inject: ['agentLoop'] }))
ctx.on('agent/status', (_agent, status) => {
if (status === 'disposed') throw new Error('broken status listener')
if (status === 'idle') throw new Error('broken status listener')
})
send(agent, 'go')
@@ -716,8 +436,7 @@ describe('disposed status is part of the agent/status contract', () => {
await fiber.dispose()
await driverDone(agent) // must not hang
expect(agent.status).toBe('disposed')
expect(ctx.agents.get(SessionId('scoped'))).toBeUndefined() // unregistered despite the throw
await expect.poll(() => ctx.agents.get(SessionId('scoped')) === undefined).toBe(true)
})
})
@@ -739,7 +458,9 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), {}) // no model
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
ctx.on('agent/error', (_agent, _turn, _step, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -753,8 +474,8 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), {}) // no model — router plugin decides
ctx.on('agent/request', async (_agent, _turn, _step, config, _signal) => {
return { ...config, provider: 'mock', model: 'mock' }
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
return { ...await next(), provider: 'mock', model: 'mock' }
})
send(agent, 'go')
@@ -763,11 +484,11 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
expect(agent.session.deriveMessages().at(-1)?.content).toEqual([{ type: 'text', text: 'routed' }])
})
it('agent/queued carries the resolved source; steering/message records its source', async () => {
it('agent/inbox/enqueue carries the exact message; steering/message records its source', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'noop', {}), textResponse('done')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'noop',
description: '',
parameters: {},
@@ -777,107 +498,30 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
},
}))
const queuedSources: { source: MessageSource; steering: boolean }[] = []
ctx.on('agent/queued', (_agent, _content, info) => void queuedSources.push(info))
const queuedSources: MessageSource[] = []
const queuedShapes: string[][] = []
ctx.on('agent/inbox/enqueue', (_agent, message) => {
queuedSources.push(message.source)
queuedShapes.push(Object.keys(message).sort())
})
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(queuedSources).toEqual([
{ kind: 'user' },
{ kind: 'plugin', plugin: 'goal' },
])
expect(queuedShapes).toEqual([
['content', 'id', 'source'],
['content', 'id', 'source'],
])
// The drain appends the durable steering/message with the caller's source
// intact — the log, not a transient emit, is where consumers read it.
const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.source] : [])
expect(steeringSources).toEqual([{ kind: 'plugin', plugin: 'goal' }])
})
it('send() owns content and source before notification and delivery', async () => {
const adapter = new MockAdapter([textResponse('done')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('owned-send'), { provider: 'mock', model: 'mock' })
const content = [{ type: 'text' as const, text: 'accepted-send' }]
const source = { kind: 'plugin' as const, plugin: 'accepted-source' }
let notifiedContent: ContentBlock[] | undefined
let notifiedSource: MessageSource | undefined
ctx.on('agent/queued', (subject, acceptedContent, info) => {
if (subject !== agent || info.steering) return
// Retain the exact notification references: cloning here would test the
// listener's copy rather than the event/inbox ownership boundary.
notifiedContent = acceptedContent
notifiedSource = info.source
})
agent.followup(content, { source })
content[0]!.text = 'caller-mutated-send'
source.plugin = 'caller-mutated-source'
await waitForIdle(ctx, agent)
expect(notifiedContent).toEqual([{ type: 'text', text: 'accepted-send' }])
expect(notifiedSource).toEqual({ kind: 'plugin', plugin: 'accepted-source' })
expect(Object.isFrozen(notifiedContent)).toBe(true)
expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
expect(Object.isFrozen(notifiedSource)).toBe(true)
const recorded = agent.session.events.flatMap(event => event.type === 'user/message' ? [event.data] : [])
expect(recorded).toContainEqual({
content: [{ type: 'text', text: 'accepted-send' }],
source: { kind: 'plugin', plugin: 'accepted-source' },
})
const request = JSON.stringify(adapter.requests[0]!.messages)
expect(request).toContain('accepted-send')
expect(request).not.toContain('caller-mutated-send')
})
it('running steer() owns content and source before notification and delivery', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'gate', {}), textResponse('done')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('owned-steer'), { provider: 'mock', model: 'mock' })
const entered = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
ctx.tools.register(defineTool({
name: 'gate',
description: '',
parameters: {},
async execute() {
entered.resolve(undefined)
await release.promise
return [{ type: 'text', text: 'tool done' }]
},
}))
let notifiedContent: ContentBlock[] | undefined
let notifiedSource: MessageSource | undefined
ctx.on('agent/queued', (subject, acceptedContent, info) => {
if (subject !== agent || !info.steering) return
notifiedContent = acceptedContent
notifiedSource = info.source
})
agent.followup([{ type: 'text', text: 'start' }])
await entered.promise
expect(agent.status).toBe('running')
const content = [{ type: 'text' as const, text: 'accepted-steer' }]
const source = { kind: 'plugin' as const, plugin: 'accepted-source' }
agent.steer(content, { source })
content[0]!.text = 'caller-mutated-steer'
source.plugin = 'caller-mutated-source'
const idle = waitForIdle(ctx, agent)
release.resolve(undefined)
await idle
expect(notifiedContent).toEqual([{ type: 'text', text: 'accepted-steer' }])
expect(notifiedSource).toEqual({ kind: 'plugin', plugin: 'accepted-source' })
expect(Object.isFrozen(notifiedContent)).toBe(true)
expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
expect(Object.isFrozen(notifiedSource)).toBe(true)
const recorded = agent.session.events.flatMap(event => event.type === 'steering/message' ? [event.data] : [])
expect(recorded).toContainEqual({
turn: 1,
content: [{ type: 'text', text: 'accepted-steer' }],
source: { kind: 'plugin', plugin: 'accepted-source' },
})
const request = JSON.stringify(adapter.requests[1]!.messages)
expect(request).toContain('accepted-steer')
expect(request).not.toContain('caller-mutated-steer')
})
})
describe('turn numbering continues across seeded sessions', () => {
@@ -900,12 +544,9 @@ describe('turn numbering continues across seeded sessions', () => {
ctx2.llm.registerAdapter(['mock'], second)
const seeded = ctx2.sessions.create(SessionId('forked'), { seed: [...agent.session.events] })
const prepared = prepareReactLoopAgent(
const forked = new ReactLoopAgent(
ctx2, SessionId('forked-agent'), { provider: 'mock', model: 'mock' }, seeded,
)
const forked = prepared.agent
prepared.markPublished()
ctx2.effect(() => { prepared.start(); return prepared.dispose })
const turns: number[] = []
ctx2.on('session/event', (_s, event) => { if (event.type === 'turn/start') turns.push(event.data.turn) })
@@ -1076,7 +717,9 @@ describe('turn and step boundary recovery', () => {
if (event.type === 'step/start' && !threw) { threw = true; throw new Error('boom step-start') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
ctx.on('agent/error', (_a, _t, _s, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -1107,7 +750,9 @@ describe('turn and step boundary recovery', () => {
}
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) })
ctx.on('agent/error', (_agent, _turn, _step, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -1123,41 +768,6 @@ describe('turn and step boundary recovery', () => {
expect(errors.map(error => error.message)).toEqual(['reject step-start before commit'])
})
it('a one-shot turn/end validation failure preserves the earlier turn error on retry', async () => {
const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', failure: { message: 'provider failed', code: 'UNKNOWN' } } }]
const adapter = new MockAdapter([errorStream])
const ctx = await balancedHarness(adapter)
const agent = ctx.agentLoop.create(SessionId('a-turnend-veto'), { provider: 'mock', model: 'mock' })
let rejected = false
ctx.on('internal/dispatch', (_mode, name, args) => {
if (name !== 'session/event') return
const event = args[1] as SessionEvent
if (event.type === 'turn/end' && !rejected) {
rejected = true
throw new Error('reject first turn-end')
}
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) })
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(errors.map(error => error.message)).toEqual(['provider failed'])
expect(boundaryCounts(agent)).toMatchObject({
turnStart: 1,
turnEnd: 1,
stepStart: 1,
stepEnd: 1,
errors: 1,
})
const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toMatchObject({
kind: 'error',
failure: { message: 'provider failed', code: 'UNKNOWN' },
})
})
it('a one-shot step/end validation failure keeps the step open until retry succeeds', async () => {
const adapter = new MockAdapter([textResponse('completed before close validation')])
const ctx = await balancedHarness(adapter)
@@ -1172,7 +782,9 @@ describe('turn and step boundary recovery', () => {
}
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) })
ctx.on('agent/error', (_agent, _turn, _step, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -1261,14 +873,16 @@ describe('turn and step boundary recovery', () => {
}, { inject: ['agentLoop'] }))
let threw = false
ctx.on('agent/pre-step', () => {
ctx.on('agent/step', () => {
if (threw) return
threw = true
void fiber.dispose()
throw new Error('boom pre-step during disposal')
})
const errorEmits: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => void errorEmits.push(error))
ctx.on('agent/error', (_a, _t, _s, error) => {
if (error instanceof Error) errorEmits.push(error)
})
send(agent, 'go')
await driverDone(agent)
@@ -1295,7 +909,9 @@ describe('turn and step boundary recovery', () => {
if (!threw && event.type === 'turn/start') { threw = true; throw new Error('boom turn/start append') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
ctx.on('agent/error', (_a, _t, _s, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -1326,7 +942,9 @@ describe('turn and step boundary recovery', () => {
if (event.type === 'step/end' && !threw) { threw = true; throw new Error('boom step-end') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
ctx.on('agent/error', (_a, _t, _s, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -1365,7 +983,9 @@ describe('turn and step boundary recovery', () => {
if (!threw && event.type === 'step/end') { threw = true; throw new Error('boom step/end listener') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
ctx.on('agent/error', (_a, _t, _s, error) => {
if (error instanceof Error) errors.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -1419,7 +1039,7 @@ describe('tool result call identity', () => {
textResponse('done'),
])
const ctx = await harness(adapter)
ctx.tools.register(defineTool({
ctx.tools.register(defineContentToolFixture({
name: 'echo',
description: 'echo',
parameters: { x: { type: 'number' } },
@@ -1458,34 +1078,6 @@ describe('tool result call identity', () => {
})
})
describe('surface: assistant/message records exact empty provenance when no chunks streamed', () => {
it('a step-result listener injecting content over an empty stream records sourceEventSeqs []', async () => {
// The explicit empty source set distinguishes a known empty provider
// stream from legacy events whose provenance was not recorded.
const adapter = new MockAdapter([[]])
const ctx = await harness(adapter)
await mountInvariants(ctx)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/step-result', async (_agent, _turn, _step, _message, _signal) => ({
role: 'assistant' as const,
content: [{ type: 'text' as const, text: 'injected' }],
}))
send(agent, 'go')
await waitForIdle(ctx, agent)
const recorded = agent.session.events.find(e => e.type === 'assistant/message')!
expect(recorded.type).toBe('assistant/message')
expect(recorded.surfaceOp).toBe('append')
expect(recorded.sourceEventSeqs).toEqual([])
// The injected content reaches derived history.
expect(JSON.stringify(agent.session.deriveMessages())).toContain('injected')
})
})
describe('disposal and cancellation during pre-step assembly', () => {
it('disposal during system-prompt assembly drops the about-to-start step as disposed', { timeout: 30000 }, async () => {
// Start disposal, then release assembly. Do not await disposal first: it
@@ -1590,7 +1182,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
expect(reasons).toEqual([{ kind: 'aborted' }])
})
it('disposal during agent/pre-step seam ends the turn disposed', { timeout: 15000 }, async () => {
it('disposal during agent/step seam ends the turn disposed', { timeout: 15000 }, async () => {
// Start disposal, then release pre-step; awaiting disposal first would
// deadlock on the blocked driver.
const adapter = new MockAdapter(['hang'])
@@ -1607,7 +1199,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
await mountInvariants(ctx)
ctx.llm.registerAdapter(['mock'], adapter)
ctx.on('agent/pre-step', async () => {
ctx.on('agent/step', async () => {
await blocker
})
@@ -1642,7 +1234,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
// (turn boundaries have no agent/* mirror).
})
it('cancel during agent/pre-step seam ends the turn aborted', { timeout: 15000 }, async () => {
it('cancel during agent/step seam ends the turn aborted', { timeout: 15000 }, async () => {
// Release pre-step after cancellation to exercise the post-seam check.
const adapter = new MockAdapter(['hang'])
let releasePreStep!: () => void
@@ -1658,7 +1250,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
await mountInvariants(ctx)
ctx.llm.registerAdapter(['mock'], adapter)
ctx.on('agent/pre-step', async () => {
ctx.on('agent/step', async () => {
await blocker
})