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

test: migrate consumers to inbox and owned-run APIs

Browse Source
This commit is contained in:
_Kerman
2026-07-30 17:28:03 +08:00
parent a6baddaaac
commit 5a0d26a0e4
72 changed files with 716 additions and 1381 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
packages/core/agent-loop/tests/cancel.spec.ts
View File

@@ -235,7 +235,7 @@ describe('Agent.cancel()', () => {
agent.cancel({ kind: 'user' })
await waitForIdle(ctx, agent)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
expect(userTexts(agent)).toEqual(['go'])
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
expect(adapter.requests).toHaveLength(1)
@@ -272,7 +272,7 @@ describe('Agent.cancel()', () => {
dispose()
expect(executions).toBe(0)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
const call = agent.session.events.find(event => event.type === 'tool/call')
const result = agent.session.events.find(event => event.type === 'tool/result')
expect(call?.type === 'tool/call' ? call.data.callId : undefined).toBe('c1')
@@ -291,7 +291,7 @@ describe('Agent.cancel()', () => {
.find(block => block.type === 'tool-result')
expect(replayedResult).toMatchObject({ toolCallId: 'c1', isError: true })
expect(reasons).toEqual([
{ kind: 'aborted' },
{ kind: 'aborted', reason: { kind: 'user' } },
{ kind: 'completed' },
])
})
@@ -342,7 +342,7 @@ describe('Agent.cancel()', () => {
// the caller's cause — the marker carries `cancel(cause)` through even
// though no AbortController observed it in this window.
expect(streamed).toBe(false)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
})
it('cancel from a synchronous step/start session-event listener drops the step (post-step-start window)', async () => {
@@ -370,12 +370,12 @@ describe('Agent.cancel()', () => {
// No step streamed, the turn ended with the coarse aborted outcome, and the
// log is balanced (the open step was closed by the cancel branch).
expect(streamed).toBe(false)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
const types = agent.session.events.map(e => e.type)
expect(types.filter(t => t === 'step/start').length).toBe(types.filter(t => t === 'step/end').length)
})
it('disposal from a synchronous step/start session-event listener closes the open step as disposed', async () => {
it('disposal from a synchronous step/start session-event listener stops before adapter dispatch', async () => {
const adapter = new MockAdapter([textResponse('should not stream')])
const ctx = new Context()
await ctx.plugin(LlmService)
@@ -405,8 +405,7 @@ describe('Agent.cancel()', () => {
expect(streamed).toBe(false)
expect(adapter.requests).toHaveLength(0)
const turnEnd = agent.session.events.findLast(e => e.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(false)
const types = agent.session.events.map(e => e.type)
expect(types.filter(t => t === 'step/start').length).toBe(types.filter(t => t === 'step/end').length)
})
@@ -437,7 +436,7 @@ describe('Agent.cancel()', () => {
// Only ONE step ran (the second was cancelled in the stopping window),
// and the shared turn signal classified the durable outcome as aborted.
expect(steps).toBe(1)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
})
it('cancel from a synchronous agent/status(running) listener drops the turn (window 2)', async () => {
@@ -569,10 +568,10 @@ describe('Agent.cancel()', () => {
const reasons = agent.session.events
.filter(event => event.type === 'turn/end')
.map(event => event.type === 'turn/end' ? event.data.reason : undefined)
expect(reasons).toEqual([{ kind: 'aborted' }, { kind: 'completed' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }, { kind: 'completed' }])
})
it('keeps the first typed cause for an active turn and detaches the runtime reason', async () => {
it('keeps the first typed cause for an active turn', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('typed-first-wins'), { provider: 'mock', model: 'mock' })
@@ -581,16 +580,17 @@ describe('Agent.cancel()', () => {
send(agent, 'go')
await expect.poll(() => adapter.requests.length).toBe(1)
agent.cancel(supplied)
supplied.kind = 'user'
agent.cancel({ kind: 'user' })
await waitForIdle(ctx, agent)
const runtimeReason: unknown = adapter.requests[0]?.signal?.reason
expect(runtimeReason).toEqual({ kind: 'parent' })
expect(runtimeReason).not.toBe(supplied)
expect(Object.isFrozen(runtimeReason)).toBe(true)
expect(runtimeReason).toBe(supplied)
const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({
kind: 'aborted',
reason: { kind: 'parent' },
})
})
it('preserves the first user cancellation when lifecycle teardown races it', async () => {
@@ -608,7 +608,7 @@ describe('Agent.cancel()', () => {
await handle.dispose()
const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } })
})
it.each([
@@ -688,7 +688,7 @@ describe('Agent.cancel()', () => {
if (stage === 'prompt-submit') {
expect(turnEnd).toBeUndefined()
} else {
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } })
}
await ctx.fiber.dispose()
})

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

@@ -1,7 +1,7 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { createUserMessage, freezeMessage, CallId, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason, type UserMessage } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineContentToolFixture, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
@@ -268,7 +268,7 @@ describe('abort during tool execution ends the turn', () => {
[{ type: 'text', text: 'accepted result context during disposal' }],
])
expect(agent.session.events.find(event => event.type === 'turn/end')?.data.reason)
.toEqual({ kind: 'disposed' })
.toEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
})
it('limits injection deferral to the current tool batch', async () => {
@@ -390,7 +390,6 @@ describe('steering from late extension points is never stranded', () => {
if (event.type === 'turn/start') turns.push(event.data.turn)
if (event.type === 'turn/end' && !steeredOnce) {
steeredOnce = true
expect(agent.acceptsNextStep).toBe(false)
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'too late for this turn' }], source: { kind: 'user' } }))
}
})
@@ -461,7 +460,7 @@ describe('disposal leaves the two-state status contract balanced', () => {
await driverDone(agent)
expect(statuses).toEqual(['running', 'idle'])
expect(reasons).toEqual([{ kind: 'disposed' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'disposed' } }])
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
const messages = agent.session.events
.filter(event => event.type === 'user/message')
@@ -956,7 +955,7 @@ describe('turn and step boundary recovery', () => {
const turnEnds = e.filter(x => x.type === 'turn/end').length
expect(turnStarts).toBe(1)
expect(turnEnds).toBe(1) // balanced — the turn was closed despite disposal
expect(reasons).toEqual([{ kind: 'disposed' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'disposed' } }])
// no error reason: disposal is not a failure.
expect(e.some(x => x.type === 'turn/end' && x.data.reason.kind === 'error')).toBe(false)
})
@@ -989,7 +988,7 @@ describe('turn and step boundary recovery', () => {
// Balanced: one turn/start, one turn/end carrying disposed (NOT error).
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
expect(e.some(x => x.type === 'turn/end' && x.data.reason.kind === 'error')).toBe(false)
// No step opened (the throw was before step/start) and disposal is not a
// failure, so no agent/error for the contained throw.
@@ -1225,7 +1224,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
})
@@ -1272,12 +1271,12 @@ describe('disposal and cancellation during pre-step assembly', () => {
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } })
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(e.some(x => x.type === 'assistant/message')).toBe(false)
expect(adapter.requests).toHaveLength(0)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
})
it('disposal during agent/step listeners ends the turn disposed', { timeout: 15000 }, async () => {
@@ -1324,7 +1323,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
// Disposal wins the post-listener check — reason is `disposed`.
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
// The durable turn/end record is the authoritative turn-boundary signal
@@ -1372,10 +1371,10 @@ describe('disposal and cancellation during pre-step assembly', () => {
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } })
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
})
it('disposal during assembly does not leak an LLM call or append assistant/chunk', { timeout: 15000 }, async () => {

24
packages/core/agent-loop/tests/coverage-edges.spec.ts
View File

@@ -165,8 +165,8 @@ describe('thrown-value propagation', () => {
expect(errors[0]).toEqual({ code: 500 })
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error'
&& ('failure' in turnEnd.data.reason ? turnEnd.data.reason.failure.code : turnEnd.data.reason.code))
.toBeUndefined()
? turnEnd.data.reason.error
: undefined).toEqual({ code: 500 })
})
})
@@ -197,8 +197,7 @@ describe('coded error data emission', () => {
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
expect(turnEnd).toBeDefined()
if (turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error') {
expect('failure' in turnEnd.data.reason ? turnEnd.data.reason.failure.code : turnEnd.data.reason.code)
.toBe('RATE_LIMIT')
expect(turnEnd.data.reason.error).toMatchObject({ code: 'RATE_LIMIT' })
}
})
})
@@ -221,7 +220,7 @@ describe('disposed vs aborted branching', () => {
await driverDone(agent)
// Disposal wins abort classification because the error path checks it first.
expect(reasons).toContainEqual({ kind: 'disposed' })
expect(reasons).toContainEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
})
})
@@ -285,9 +284,7 @@ describe('request-error action edges', () => {
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('retry-raced'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request-error', async (
subject, _turn, _step, _error, _failure, _priorFailures, _retryPolicy, signal, next,
) => {
ctx.on('agent/request-error', async (subject, _context, signal, next) => {
await next()
subject.cancel({ kind: 'user' })
expect(signal.aborted).toBe(true)
@@ -480,7 +477,7 @@ describe('unrenderable failure settlement', () => {
if (end?.type === 'turn/end' && end.data.reason.kind === 'error') {
// The durable failure keeps the adapter facts' message, not the
// unrenderable chain.
expect(end.data.reason.failure?.message).not.toBe('<unrenderable value>')
expect(errorChain(end.data.reason.error)).not.toBe('<unrenderable value>')
}
})
})
@@ -493,10 +490,11 @@ describe('driver bookkeeping edges', () => {
provider: 'mock',
model: 'mock',
})
ctx.on('agent/inbox/enqueue', (subject) => {
if (subject !== agent) return
subject.cancel({ kind: 'user' })
const mutable = subject as Agent & { done: Promise<void> }
ctx.on('session/event', (session, event) => {
if (session !== agent.session || event.type !== 'agent/inbox/spliced'
|| event.data.target !== 'next-turn' || event.data.inserted.length === 0) return
agent.cancel({ kind: 'user' })
const mutable = agent as Agent & { done: Promise<void> }
mutable.done = Promise.reject(new Error('replacement rejected'))
})

84
packages/core/agent-loop/tests/interception.spec.ts
View File

@@ -11,7 +11,6 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineContentToolFixture, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, {
type Agent,
type InboxPlacement,
type PromptDecision,
type SessionStartSource,
} from '@deepseek-ai/dsh-agent'
@@ -66,8 +65,8 @@ describe('agent/prompt-submit', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const seen: string[] = []
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next) => {
seen.push(message.content.map(b => (b.type === 'text' ? b.text : '')).join(''))
ctx.on('agent/prompt-submit', async (_agent, messages, _signal, next) => {
seen.push(messages[0]!.content.map(b => (b.type === 'text' ? b.text : '')).join(''))
return next()
})
@@ -86,9 +85,9 @@ describe('agent/prompt-submit', () => {
const entered = Promise.withResolvers<undefined>()
const decision = Promise.withResolvers<PromptDecision>()
const observed: UserMessage[] = []
ctx.on('agent/inbox/enqueue', (subject, item) => {
if (subject !== agent) return
const message = item.message
ctx.on('agent/prompt-submit', async (subject, messages) => {
if (subject !== agent) return { kind: 'allow', messages }
const message = messages[0]!
expect(Object.isFrozen(message)).toBe(true)
expect(Object.isFrozen(message.content)).toBe(true)
expect(Object.isFrozen(message.content[0])).toBe(true)
@@ -97,11 +96,7 @@ describe('agent/prompt-submit', () => {
const block = message.content[0]
if (block?.type === 'text') block.text = 'listener mutation'
}).toThrow()
})
ctx.on('agent/inbox/enqueue', (subject, item) => {
if (subject === agent) observed.push(item.message)
})
ctx.on('agent/prompt-submit', async () => {
observed.push(message)
entered.resolve(undefined)
return decision.promise
})
@@ -120,7 +115,7 @@ describe('agent/prompt-submit', () => {
expect(() => {
if (input.source.kind === 'plugin') input.source.plugin = 'caller mutation'
}).toThrow(TypeError)
decision.resolve({ kind: 'allow' })
decision.resolve({ kind: 'allow', messages: [input] })
await idle
expect(observed).toHaveLength(1)
@@ -138,8 +133,11 @@ describe('agent/prompt-submit', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
({ kind: 'allow', content: [{ type: 'text', text: 'REWRITTEN' }] }))
ctx.on('agent/prompt-submit', async (_agent, messages): Promise<PromptDecision> =>
({
kind: 'allow',
messages: [{ ...messages[0]!, content: [{ type: 'text', text: 'REWRITTEN' }] }],
}))
send(agent, 'original')
await waitForIdle(ctx, agent)
@@ -156,10 +154,10 @@ describe('agent/prompt-submit', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
ctx.on('agent/prompt-submit', async (_agent, messages): Promise<PromptDecision> =>
({
kind: 'allow',
additionalContexts: [createUserMessage({
messages: [...messages, createUserMessage({
content: [{ type: 'text', text: '<system-reminder>extra ctx</system-reminder>' }],
source: { kind: 'plugin', plugin: 'test' },
})],
@@ -183,11 +181,13 @@ describe('agent/prompt-submit', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
ctx.on('agent/prompt-submit', async (_agent, messages): Promise<PromptDecision> =>
({
kind: 'allow',
content: [{ type: 'text', text: 'REWRITTEN prompt' }],
additionalContexts: [createUserMessage({
messages: [{
...messages[0]!,
content: [{ type: 'text', text: 'REWRITTEN prompt' }],
}, createUserMessage({
content: [{ type: 'text', text: 'injected ctx' }], source: { kind: 'plugin', plugin: 'test' },
})],
}))
@@ -236,20 +236,17 @@ describe('agent/prompt-submit', () => {
const agent = ctx.agentLoop.create(SessionId('admission-outbox'), { provider: 'mock', model: 'mock' })
const entered = Promise.withResolvers<undefined>()
const decision = Promise.withResolvers<PromptDecision>()
const placements: InboxPlacement[] = []
ctx.on('agent/prompt-submit', async () => {
let claimed: UserMessage[] = []
ctx.on('agent/prompt-submit', async (_agent, messages) => {
claimed = messages
entered.resolve(undefined)
return decision.promise
})
ctx.on('agent/inbox/enqueue', (subject, item) => {
if (subject === agent) placements.push(item.placement)
})
const idle = waitForIdle(ctx, agent)
send(agent, 'admitted prompt')
await entered.promise
expect(agent.status).toBe('running')
expect(agent.acceptsNextStep).toBe(true)
expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
agent.inject(createUserMessage({
@@ -258,11 +255,15 @@ describe('agent/prompt-submit', () => {
}))
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'admission steering' }], source: { kind: 'user' } }))
expect(events(agent).some(event => event.type === 'user/message')).toBe(false)
expect(placements).toEqual(['queued', 'steering'])
expect(agent.inbox.nextStep.map(message => message.content[0]))
.toEqual([
{ type: 'text', text: 'attached context' },
{ type: 'text', text: 'admission steering' },
])
decision.resolve({ kind: 'allow' })
decision.resolve({ kind: 'allow', messages: claimed })
await idle
expect(agent.acceptsNextStep).toBe(false)
expect(agent.inbox.hasPending).toBe(false)
const staged = events(agent).filter(event =>
event.type === 'turn/start' || event.type === 'user/message' || event.type === 'steering/message')
@@ -298,7 +299,6 @@ describe('agent/prompt-submit', () => {
const blockedIdle = waitForIdle(ctx, agent)
send(agent, 'blocked prompt')
await entered.promise
expect(agent.acceptsNextStep).toBe(true)
agent.inject(createUserMessage({
content: [{ type: 'text', text: 'staged context' }],
source: { kind: 'plugin', plugin: 'test' },
@@ -307,7 +307,7 @@ describe('agent/prompt-submit', () => {
decision.resolve({ kind: 'block', reason: 'policy' })
await blockedIdle
expect(agent.acceptsNextStep).toBe(false)
expect(agent.inbox.nextStep).toHaveLength(2)
expect(events(agent)).toEqual([])
expect(adapter.requests).toEqual([])
@@ -334,14 +334,16 @@ describe('agent/prompt-submit', () => {
provider: 'mock',
model: 'mock',
})
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next) => {
ctx.on('agent/prompt-submit', async (_agent, messages, _signal, next) => {
const decision = await next()
return message.content.some(block => block.type === 'text' && block.text === 'blocked prompt')
return messages.some(message =>
message.content.some(block => block.type === 'text' && block.text === 'blocked prompt'))
? { kind: 'block', reason: 'policy' }
: decision
})
ctx.on('agent/prompt-submit', async (subject, message, _signal, next) => {
if (message.content.some(block => block.type === 'text' && block.text === 'blocked prompt')) {
ctx.on('agent/prompt-submit', async (subject, messages, _signal, next) => {
if (messages.some(message =>
message.content.some(block => block.type === 'text' && block.text === 'blocked prompt'))) {
subject.inject(createUserMessage({
content: [{ type: 'text', text: 'earlier state change' }],
source: { kind: 'plugin', plugin: 'test' },
@@ -446,8 +448,9 @@ describe('agent/prompt-submit', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next): Promise<PromptDecision> => {
const text = message.content.map(b => (b.type === 'text' ? b.text : '')).join('')
ctx.on('agent/prompt-submit', async (_agent, messages, _signal, next): Promise<PromptDecision> => {
const text = messages.flatMap(message => message.content)
.map(b => (b.type === 'text' ? b.text : '')).join('')
return text === 'secret' ? { kind: 'block', reason: 'policy: no secrets' } : next()
})
@@ -475,9 +478,9 @@ describe('agent/prompt-submit', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let threw = false
ctx.on('agent/prompt-submit', async () => {
ctx.on('agent/prompt-submit', async (_agent, messages) => {
if (!threw) { threw = true; throw new Error('prompt hook broke') }
return { kind: 'allow' as const }
return { kind: 'allow' as const, messages }
})
const errors: Error[] = []
const reasons: TurnEndReason[] = []
@@ -680,8 +683,9 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
agent.inject(createUserMessage({ content: [{ type: 'text', text: `policy active (started: ${source})` }], source: { kind: 'plugin', plugin: 'native-guard' } }))
})
// 2. PromptSubmit: block a forbidden prompt, annotate the rest.
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next): Promise<PromptDecision> => {
const text = message.content.map(b => (b.type === 'text' ? b.text : '')).join('')
ctx.on('agent/prompt-submit', async (_agent, messages, _signal, next): Promise<PromptDecision> => {
const text = messages.flatMap(message => message.content)
.map(b => (b.type === 'text' ? b.text : '')).join('')
if (text.includes('rm -rf')) return { kind: 'block', reason: 'destructive prompt blocked' }
return next()
})

134
packages/core/agent-loop/tests/loop.spec.ts
View File

@@ -94,11 +94,10 @@ describe('agent loop', () => {
expect(order).toEqual(['turn/start', 'step/start', 'step/end', 'turn/end'])
const types = agent.session.events.map(e => e.type)
// turn/start opens the turn, THEN the queued user message is recorded inside
// it (every event is turn-enclosed), then the assembled message (carrying the
// step's usage).
expect(types[0]).toBe('turn/start')
expect(types[1]).toBe('user/message')
// Durable inbox receipt and admission bracket the turn-owned transcript.
expect(types[0]).toBe('agent/inbox/spliced')
expect(types).toContain('turn/start')
expect(types).toContain('user/message')
expect(types).toContain('assistant/message')
const assistantMessage = agent.session.events.find(e => e.type === 'assistant/message')
expect(assistantMessage?.type === 'assistant/message' && assistantMessage.data.usage).toEqual({ inputTokens: 10, outputTokens: 'hello there'.length })
@@ -201,9 +200,12 @@ describe('agent loop', () => {
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(0) // the request was never sent
expect(errors.some(e => e.message.includes('no value for this assembly'))).toBe(true)
expect(errors).toEqual([])
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('error')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error'
? turnEnd.data.reason.error
: '').toContain('no value for this assembly')
// The loop survived: a waterfall listener rescues {{cwd}} and the SAME
// agent completes a real model turn.
@@ -307,9 +309,11 @@ describe('agent loop', () => {
await waitForIdle(ctx, agent)
const types = agent.session.events.map(e => e.type)
expect(types).toContain('steering/message')
// steering recorded before the second step's request derived its history
const steeringSeq = agent.session.events.find(e => e.type === 'steering/message')!.seq
const steering = agent.session.events.find(e =>
e.type === 'user/message' && JSON.stringify(e.data.content).includes('change of plans'))
expect(steering).toBeDefined()
// Steering is admitted before the second step's request derives history.
const steeringSeq = steering!.seq
const secondStepStart = agent.session.events.filter(e => e.type === 'step/start')[1]
expect(secondStepStart).toBeDefined()
expect(steeringSeq).toBeLessThan(secondStepStart!.seq)
@@ -320,8 +324,8 @@ describe('agent loop', () => {
expect(flat).toContain('change of plans')
})
it('same-tick idle steering preserves one turn per send', async () => {
const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
it('coalesces same-tick idle steering into one turn', async () => {
const adapter = new MockAdapter([textResponse('first')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -330,7 +334,7 @@ describe('agent loop', () => {
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'second idle steer' }], source: { kind: 'user' } }))
await idle
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
expect(agent.session.events
.filter(event => event.type === 'user/message')
.map(event => event.data.content)).toEqual([
@@ -338,13 +342,12 @@ describe('agent loop', () => {
[{ type: 'text', text: 'second idle steer' }],
])
expect(agent.session.events.filter(event => event.type === 'steering/message')).toEqual([])
expect(adapter.requests).toHaveLength(2)
expect(adapter.requests).toHaveLength(1)
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('first idle steer')
expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('second idle steer')
expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('second idle steer')
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('second idle steer')
})
it('keeps steering staged after a failed step until the next admitted turn', async () => {
it('contains a throwing step observer and carries steering into a replacement turn', async () => {
const adapter = new MockAdapter([textResponse('recovered')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('failed-steering'), { provider: 'mock', model: 'mock' })
@@ -359,20 +362,13 @@ describe('agent loop', () => {
send(agent, 'prompt')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(false)
send(agent, 'resume')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(true)
expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(false)
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('pending steering')
})
it('inject() while idle appends context without opening a turn', async () => {
it('inject() while idle durably stages context without opening a turn', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -382,11 +378,14 @@ describe('agent loop', () => {
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(0)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'user/message',
type: 'agent/inbox/spliced',
data: {
role: 'user',
content: [{ type: 'text', text: 'file changed: a.ts' }],
source: { kind: 'plugin', plugin: 'watcher' },
target: 'next-step',
inserted: [{
role: 'user',
content: [{ type: 'text', text: 'file changed: a.ts' }],
source: { kind: 'plugin', plugin: 'watcher' },
}],
},
})
@@ -540,7 +539,7 @@ describe('agent loop', () => {
expect(agent.session.events.some(e => e.type === 'tool/result')).toBe(true)
})
it('a concluding tool result beats steering that arrived during the same step', async () => {
it('continues for steering that arrived during a concluding tool step', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'finalize', {}),
textResponse('next turn reply'),
@@ -562,17 +561,10 @@ describe('agent loop', () => {
send(agent, 'go')
await waitForIdle(ctx, agent)
// The terminal result stands: no extra request reopens the concluded turn.
expect(adapter.requests).toHaveLength(1)
expect(adapter.requests).toHaveLength(2)
const events = agent.session.events.map(event => event.type)
expect(events.filter(type => type === 'turn/end')).toHaveLength(1)
// The steering is durable inside the concluded turn and feeds the NEXT
// turn's request instead of being dropped or re-queued.
expect(events).toContain('steering/message')
send(agent, 'follow up')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('late steering')
const texts = adapter.requests[1]!.messages
.flatMap(message => message.content)
.filter(block => block.type === 'text')
@@ -630,38 +622,21 @@ describe('agent loop', () => {
expect(fires.every(({ signal }) => signal instanceof AbortSignal)).toBe(true)
})
it('agent/step fires BEFORE the step it precedes opens (events land outside the step)', async () => {
// The append lands before step/start, yet derive happens afterwards and the
// same step's request must include it.
it('agent/step fires after its step boundary opens and before the request', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let injected = false
let boundaryOpen = false
ctx.on('agent/step', (subject) => {
if (subject === agent && !injected) {
injected = true
subject.session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'INJECTED-IN-PRE-STEP' }],
source: { kind: 'plugin', plugin: 'test' },
}), { surfaceOp: 'append' })
}
if (subject === agent) boundaryOpen = subject.session.events.at(-1)?.type === 'step/start'
})
send(agent, 'go')
await waitForIdle(ctx, agent)
// The adapter's request includes the node injected during pre-step (derive
// reflects it).
const text = JSON.stringify(adapter.requests[0]!.messages)
expect(text).toContain('INJECTED-IN-PRE-STEP')
// And the injected event sits BEFORE the first step/start in the log —
// the seam fired outside the step.
const events = agent.session.events
const injectedSeq = events.find(e => e.type === 'user/message' && e.data.source.kind === 'plugin')!.seq
const firstStepStartSeq = events.find(e => e.type === 'step/start')!.seq
expect(injectedSeq).toBeLessThan(firstStepStartSeq)
expect(boundaryOpen).toBe(true)
expect(adapter.requests).toHaveLength(1)
})
it('a throwing agent/step listener ends the turn (error), not the loop', async () => {
@@ -683,13 +658,11 @@ describe('agent loop', () => {
send(agent, 'first')
await waitForIdle(ctx, agent)
// The first turn failed at step 1 (no model call happened), surfaced via
// agent/error, with the durable failure on turn/end.reason.
expect(errors).toHaveLength(1)
expect(errors[0]!.message).toContain('boom in pre-step')
// The first turn failed at step 1 before a model call.
expect(errors).toEqual([])
expect(adapter.requests.length).toBe(0)
const firstTurnEnd = agent.session.events.find(e => e.type === 'turn/end')
expect(firstTurnEnd?.type === 'turn/end' && firstTurnEnd.data.reason).toMatchObject({ kind: 'error', step: 1 })
expect(firstTurnEnd?.type === 'turn/end' && firstTurnEnd.data.reason).toMatchObject({ kind: 'error' })
// The step opened-and-closed count stays balanced even though it never ran.
const types = agent.session.events.map(e => e.type)
expect(types.filter(t => t === 'step/start').length).toBe(types.filter(t => t === 'step/end').length)
@@ -717,7 +690,7 @@ describe('agent loop', () => {
agent.cancel({ kind: 'user' })
await waitForIdle(ctx, agent)
expect(reasons).toEqual([{ kind: 'aborted' }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
})
it('surfaces max-tokens as the turn-end reason when the last step is cut off', async () => {
@@ -788,7 +761,7 @@ describe('agent loop', () => {
source: { kind: 'plugin', plugin: 'max-tokens-test' },
},
])
expect(reasons).toEqual([{ kind: 'max-tokens' }])
expect(reasons).toEqual([{ kind: 'completed' }])
})
it('a completed step after no max-tokens keeps the turn completed (max-tokens does not leak across turns)', async () => {
@@ -1005,14 +978,15 @@ describe('agent loop', () => {
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('completed')
})
it('keeps a reentrant agent/inbox/enqueue send as the next independent turn', async () => {
const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
it('contains a reentrant send attempted during durable inbox publication', async () => {
const adapter = new MockAdapter([textResponse('first')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let nested = false
ctx.on('agent/inbox/enqueue', (subject) => {
if (subject !== agent || nested) return
ctx.on('session/event', (session, event) => {
if (session !== agent.session || event.type !== 'agent/inbox/spliced'
|| event.data.inserted.length === 0 || nested) return
nested = true
send(agent, 'queued listener message')
})
@@ -1025,11 +999,8 @@ describe('agent loop', () => {
const messages = agent.session.events
.filter(event => event.type === 'user/message')
.map(event => event.data.content)
expect(turns).toHaveLength(2)
expect(messages).toEqual([
[{ type: 'text', text: 'outer message' }],
[{ type: 'text', text: 'queued listener message' }],
])
expect(turns).toHaveLength(1)
expect(messages).toEqual([[{ type: 'text', text: 'outer message' }]])
})
it('preserves independent turn sources across an adjacent microtask send', async () => {
@@ -1068,7 +1039,7 @@ describe('agent loop', () => {
ctx.on('session/event', (_s, event) => {
if (event.type === 'assistant/chunk' && !queued) {
queued = true
send(agent, 'second message')
queueMicrotask(() => { send(agent, 'second message') })
}
})
@@ -1110,7 +1081,7 @@ describe('agent loop', () => {
])
})
it('errors from the model surface as agent/error and end the turn', async () => {
it('records normalized model errors on the turn boundary', async () => {
const adapter = new MockAdapter([]) // script exhausted → throws
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -1125,13 +1096,12 @@ describe('agent loop', () => {
send(agent, 'hi')
await waitForIdle(ctx, agent)
expect(errors).toHaveLength(1)
expect(errors[0]!.message).toContain('script exhausted')
expect(errors).toEqual([])
expect(reasons[0]).toMatchObject({ kind: 'error' })
// The durable failure lives entirely on turn/end.reason (with the failing
// step), not a standalone error event.
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toMatchObject({ kind: 'error', step: 1 })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toMatchObject({ kind: 'error' })
})
it('disposing the loop fiber mid-turn stops the loop (HMR safety)', async () => {

13
packages/core/agent-loop/tests/request-error.spec.ts
View File

@@ -59,22 +59,19 @@ describe('agent/request-error', () => {
turn: number
step: number
failure: LlmFailure
priorFailures: readonly LlmFailure[]
retryPolicy: ResolvedRetryPolicy | undefined
}[] = []
const statuses: string[] = []
ctx.on('agent/status', (subject, status) => {
if (subject === agent) statuses.push(status)
})
ctx.on('agent/request-error', async (
subject, turn, step, _error, failure, priorFailures, retryPolicy,
) => {
ctx.on('agent/request-error', async (subject, context) => {
expect(subject).toBe(agent)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'step/end',
data: { turn, step },
data: { turn: context.turn, step: context.step },
})
seen.push({ turn, step, failure, priorFailures, retryPolicy })
seen.push(context)
return { kind: 'retry' }
})
@@ -98,8 +95,6 @@ describe('agent/request-error', () => {
},
])
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
expect(seen.map(item => item.priorFailures.map(failure => failure.code)))
.toEqual([[], ['RATE_LIMIT']])
expect(seen.map(item => item.retryPolicy)).toEqual([
expect.objectContaining({ mode: 'normal' }),
expect.objectContaining({ mode: 'normal' }),
@@ -123,7 +118,7 @@ describe('agent/request-error', () => {
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
expect(agent.session.events.find(event => event.type === 'turn/end')).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'aborted' } },
data: { reason: { kind: 'aborted', reason: { kind: 'user' } } },
})
})