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

fix(agent-loop): open turns before pre-step

Browse Source
This commit is contained in:
_Kerman
2026-08-04 13:49:35 +08:00
parent 0ac95437b4
commit d4fa26023d
154 changed files with 1104 additions and 1099 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -56,11 +56,18 @@ describe('Agent', () => {
const inserted: unknown[] = []
const claimed: unknown[] = []
const discarded: unknown[] = []
const lifecycle: string[] = []
ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'turn/start') lifecycle.push('turn/start')
})
ctx.on('agent/inbox/inserted', (subject, event) => {
if (subject === agent) inserted.push(event)
})
ctx.on('agent/inbox/claimed', (subject, event) => {
if (subject === agent) claimed.push(event)
if (subject === agent) {
lifecycle.push('agent/inbox/claimed')
claimed.push(event)
}
})
ctx.on('agent/inbox/discarded', (subject, event) => {
if (subject === agent) discarded.push(event)
@@ -78,6 +85,7 @@ describe('Agent', () => {
expect(inserted).toEqual([{ message: context }, { message: prompt }])
expect(discarded).toEqual([{ message: context }])
expect(claimed).toEqual([{ message: prompt, turn: 1 }])
expect(lifecycle).toEqual(['turn/start', 'agent/inbox/claimed'])
})
it('idle inject() rejects invalid input before enqueue', async () => {

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

@@ -361,33 +361,6 @@ describe('Agent.cancel()', () => {
expect(reasons.length).toBe(2)
})
it('cancel from a synchronous turn/start session-event listener drops the step (step-start window)', async () => {
const adapter = new MockAdapter([textResponse('should not stream')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
// A turn/start listener fires before a step controller exists, so the
// turn-scoped marker—not step abort—must drop the pending step.
let streamed = false
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'turn/start') agent.cancel({ kind: 'user' })
})
const reasons: TurnEndReason[] = []
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
send(agent, 'go')
await waitForIdle(ctx, agent)
dispose()
// No step streamed (the model never ran), and the turn ended aborted with
// 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', reason: { kind: 'user' } }])
})
it('cancel from a synchronous step/start session-event listener drops the step (post-step-start window)', async () => {
const adapter = new MockAdapter([textResponse('should not stream')])
const ctx = await harness(adapter)
@@ -487,8 +460,8 @@ describe('Agent.cancel()', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
// `agent/status` is synchronous, so cancellation can land after the first
// pre-step check; the second check must drop the now-empty turn.
// `agent/status` is synchronous, so cancellation can land before the
// durable turn-start commit and must drop the reserved work.
let streamed = false
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('agent/status', (subject, status) => {
@@ -735,11 +708,8 @@ describe('Agent.cancel()', () => {
agent.cancel({ kind: 'user' })
await idle
const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
if (stage === 'pre-step' || stage === 'system-prompt') {
expect(turnEnd).toBeUndefined()
} else {
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } })
}
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason)
.toEqual({ kind: 'aborted', reason: { kind: 'user' } })
await ctx.fiber.dispose()
})
})

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

@@ -187,7 +187,7 @@ describe('abort during tool execution ends the turn', () => {
.toBeUndefined()
})
it('parks an empty admitted batch instead of opening a turn', async () => {
it('closes an empty admitted batch as a turn without a step', async () => {
const adapter = new MockAdapter([textResponse('must not run')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a-empty-batch'), { provider: 'mock', model: 'mock' })
@@ -197,7 +197,11 @@ describe('abort during tool execution ends the turn', () => {
agent.inbox.remove(agent.inbox.nextTurn[0]!.id)
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.some(event => event.type === 'turn/start')).toBe(false)
expect(agent.session.events.filter(event => event.type === 'turn/start'
|| event.type === 'step/start' || event.type === 'turn/end').map(event => event.type))
.toEqual(['turn/start', 'turn/end'])
expect(agent.session.events.find(event => event.type === 'turn/end')?.data)
.toEqual({ turn: 1, step: 0, reason: { kind: 'completed' } })
expect(agent.inbox.nextTurn).toHaveLength(0)
})
@@ -747,6 +751,7 @@ describe('turn and step boundary recovery', () => {
expect(agent.session.events.some(event => event.type === 'turn/start'
|| event.type === 'user/message')).toBe(false)
expect(agent.inbox.nextTurn).toHaveLength(1)
expect(errors.map(error => error.message)).toEqual(['reject turn-start before commit'])
expect(adapter.requests).toHaveLength(0)
})
@@ -878,7 +883,7 @@ describe('turn and step boundary recovery', () => {
expect(e.some(x => x.type === 'turn/end' && x.data.reason.kind === 'error')).toBe(false)
})
it('contains a pre-step throw after disposal without opening a turn', async () => {
it('contains a pre-step throw after disposal inside a balanced no-step turn', async () => {
const adapter = new MockAdapter([textResponse('never reached')])
const ctx = await balancedHarness(adapter)
let agent!: Agent
@@ -902,8 +907,10 @@ describe('turn and step boundary recovery', () => {
await agent.whenIdle()
const e = [...agent.session.events]
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.filter(x => x.type === 'turn/start' || x.type === 'turn/end').map(x => x.type))
.toEqual(['turn/start', 'turn/end'])
expect(e.find(x => x.type === 'turn/end')?.data.reason)
.toEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(errorEmits).toHaveLength(0)
})
@@ -1090,7 +1097,7 @@ describe('tool result call identity', () => {
})
describe('disposal and cancellation during pre-step assembly', () => {
it('disposal during system-prompt assembly prevents the turn from opening', { timeout: 30000 }, async () => {
it('disposal during system-prompt assembly closes a no-step turn', { timeout: 30000 }, async () => {
// Start disposal, then release assembly. Do not await disposal first: it
// waits for the blocked driver to exit.
const adapter = new MockAdapter(['hang'])
@@ -1134,14 +1141,15 @@ describe('disposal and cancellation during pre-step assembly', () => {
unlisten()
const e = [...agent.session.events]
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.filter(x => x.type === 'turn/start' || x.type === 'turn/end').map(x => x.type))
.toEqual(['turn/start', 'turn/end'])
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'step/end')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'disposed' } }])
})
it('cancel during system-prompt assembly prevents the turn from opening', { timeout: 30000 }, async () => {
it('cancel during system-prompt assembly closes a no-step turn', { timeout: 30000 }, async () => {
const adapter = new MockAdapter([textResponse('should not appear')])
let releaseAssemble!: () => void
const blocker = new Promise<void>(r => void (releaseAssemble = r))
@@ -1180,17 +1188,17 @@ describe('disposal and cancellation during pre-step assembly', () => {
unlisten()
const e = [...agent.session.events]
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.filter(x => x.type === 'turn/start' || x.type === 'turn/end').map(x => x.type))
.toEqual(['turn/start', 'turn/end'])
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'step/end')).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([])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
})
it('disposal during pre-step prevents the turn from opening', { timeout: 15000 }, async () => {
it('disposal during pre-step closes a no-step turn', { timeout: 15000 }, async () => {
// Start disposal, then release pre-step; awaiting disposal first would deadlock on the blocked driver.
const adapter = new MockAdapter(['hang'])
let releasePreStep!: () => void
@@ -1227,16 +1235,16 @@ describe('disposal and cancellation during pre-step assembly', () => {
await disposalDone
await driverDone(agent)
// The post-listener cancellation check catches disposal before any turn or LLM call.
// The post-listener cancellation check catches disposal before any step or LLM call.
const e = [...agent.session.events]
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.filter(x => x.type === 'turn/start' || x.type === 'turn/end').map(x => x.type))
.toEqual(['turn/start', 'turn/end'])
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(reasons).toEqual([])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'disposed' } }])
})
it('cancel during pre-step prevents the turn from opening', { timeout: 15000 }, async () => {
it('cancel during pre-step closes a no-step turn', { timeout: 15000 }, async () => {
// Release pre-step after cancellation to exercise the post-listener check.
const adapter = new MockAdapter(['hang'])
let releasePreStep!: () => void
@@ -1275,11 +1283,11 @@ describe('disposal and cancellation during pre-step assembly', () => {
await driverDone(agent)
const e = [...agent.session.events]
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.filter(x => x.type === 'turn/start' || x.type === 'turn/end').map(x => x.type))
.toEqual(['turn/start', 'turn/end'])
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(reasons).toEqual([])
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 () => {
@@ -1319,8 +1327,10 @@ describe('disposal and cancellation during pre-step assembly', () => {
await driverDone(agent)
const e = [...agent.session.events]
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.filter(x => x.type === 'turn/start' || x.type === 'turn/end').map(x => x.type))
.toEqual(['turn/start', 'turn/end'])
expect(e.find(x => x.type === 'turn/end')?.data.reason)
.toEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
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)

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

@@ -231,7 +231,7 @@ describe('agent/pre-step', () => {
expect(events(agent).filter(event => event.type === 'step/start')).toHaveLength(1)
})
it('reject drops the claimed prompt before any turn or model call', async () => {
it('reject closes the claimed prompt turn without a step or model call', async () => {
const adapter = new MockAdapter([textResponse('should not run')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -247,11 +247,11 @@ describe('agent/pre-step', () => {
// the model was never called
expect(adapter.requests).toHaveLength(0)
const log = events(agent)
expect(log.some(e => e.type === 'turn/start')).toBe(false)
expect(log.some(e => e.type === 'turn/end')).toBe(false)
expect(log.filter(e => e.type === 'turn/start' || e.type === 'turn/end').map(e => e.type))
.toEqual(['turn/start', 'turn/end'])
expect(log.some(e => e.type === 'user/message')).toBe(false)
expect(log.some(e => e.type === 'step/start')).toBe(false)
expect(reasons).toEqual([])
expect(reasons).toEqual([{ kind: 'blocked' }])
})
it('stages inject and steer during pre-step for the entered turn', async () => {
@@ -274,7 +274,7 @@ describe('agent/pre-step', () => {
send(agent, 'entered prompt')
await entered.promise
expect(agent.status).toBe('running')
expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
expect(events(agent).some(event => event.type === 'turn/start')).toBe(true)
agent.inject(createUserMessage({
content: [{ type: 'text', text: 'attached context' }],
@@ -342,7 +342,8 @@ describe('agent/pre-step', () => {
{ type: 'text', text: 'staged context' },
{ type: 'text', text: 'staged steering' },
])
expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
expect(events(agent).filter(event => event.type === 'turn/start' || event.type === 'turn/end')
.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
expect(adapter.requests).toEqual([])
disposeBlock()
@@ -398,7 +399,8 @@ describe('agent/pre-step', () => {
send(agent, 'later prompt')
await idle
expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
expect(events(agent).filter(event => event.type === 'turn/start' || event.type === 'turn/end')
.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
expect(agent.inbox.nextStep.map(message => message.content[0]))
.toEqual([
{ type: 'text', text: 'earlier state change' },
@@ -498,8 +500,9 @@ describe('agent/pre-step', () => {
const log = events(agent)
expect(log.filter(e => e.type === 'user/message')).toHaveLength(0)
expect(adapter.requests).toHaveLength(0)
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(0)
expect(reasons).toEqual([])
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
expect(log.filter(e => e.type === 'turn/end')).toHaveLength(1)
expect(reasons).toEqual([{ kind: 'blocked' }])
expect(agent.inbox.nextTurn.map(message => message.content[0]))
.toEqual([{ type: 'text', text: 'safe' }])
@@ -537,9 +540,12 @@ describe('agent/pre-step', () => {
await idle
expect(errors).toEqual([expect.objectContaining({ message: 'prompt hook broke' })])
const log = events(agent)
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(0)
expect(log.filter(e => e.type === 'turn/end')).toHaveLength(0)
expect(reasons).toEqual([])
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
expect(log.filter(e => e.type === 'turn/end')).toHaveLength(1)
expect(reasons).toEqual([{
kind: 'error',
error: { message: 'prompt hook broke', code: 'UNKNOWN' },
}])
expect(statuses).toEqual(['running', 'idle'])
expect(adapter.requests).toHaveLength(0)
expect(agent.inbox.nextTurn.map(message => message.content[0]))
@@ -775,7 +781,7 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
expect(log.some(e => e.type.startsWith('hook/'))).toBe(false)
})
it('the same plugin blocks a destructive prompt before a turn or model call', async () => {
it('the same plugin blocks a destructive prompt inside a no-step turn', async () => {
const adapter = new MockAdapter([textResponse('should not run')])
const ctx = await harness(adapter)
await ctx.plugin(NativeGuard)
@@ -788,7 +794,7 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
await agent.whenIdle()
expect(adapter.requests).toHaveLength(0)
expect(reasons).toEqual([])
expect(reasons).toEqual([{ kind: 'blocked' }])
})
it('HMR-safety: disposing the plugin fiber removes all four listeners', async () => {