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

fix(agent-loop): latch wakes landing in the cancel-convergence window

Browse Source
This commit is contained in:
_Kerman
2026-08-07 14:54:31 +08:00
committed by Tianyi Cui
parent 22609ea425
commit df30c62e2b
23 changed files with 379 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -126,6 +126,121 @@ describe('Agent.cancel()', () => {
expect(adapter.requests).toHaveLength(3)
})
it('cancel({ keepInbox: true }) latches a waking send landing in the abort-to-idle window', async () => {
const adapter = new MockAdapter(['hang', textResponse('B reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('latch-window'), { provider: 'mock', model: 'mock' })
send(agent, 'active')
await new Promise(resolve => setTimeout(resolve, 30))
// The abort signal is set but the driver has not converged to idle yet:
// the waking send must be latched, not parked until another wake.
agent.cancel({ kind: 'user' }, { keepInbox: true })
send(agent, 'B')
await agent.whenIdle()
expect(userTexts(agent)).toEqual(['active', 'B'])
expect(adapter.requests).toHaveLength(2)
expect(agent.inbox.nextTurn).toHaveLength(0)
expect(agent.session.events.filter(e => e.type === 'turn/end').map(e =>
e.type === 'turn/end' ? e.data.reason : null)).toEqual([
{ kind: 'aborted', reason: { kind: 'user' } },
{ kind: 'completed' },
])
})
it('cancel() without keepInbox clears a latched wake alongside the inbox', async () => {
const adapter = new MockAdapter(['hang', textResponse('C reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('latch-cleared'), { provider: 'mock', model: 'mock' })
send(agent, 'active')
await new Promise(resolve => setTimeout(resolve, 30))
agent.cancel({ kind: 'user' }, { keepInbox: true })
send(agent, 'B') // latched behind the aborted activity
agent.cancel({ kind: 'user' }) // drops the inbox and the latch with it
await agent.whenIdle()
expect(userTexts(agent)).toEqual(['active'])
expect(agent.inbox.nextTurn).toHaveLength(0)
expect(adapter.requests).toHaveLength(1)
send(agent, 'C')
await agent.whenIdle()
expect(userTexts(agent)).toEqual(['active', 'C'])
expect(adapter.requests).toHaveLength(2)
})
it('removing the latched wake before convergence suppresses the replay', async () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('removed-latched-wake'), { provider: 'mock', model: 'mock' })
send(agent, 'active')
await new Promise(resolve => setTimeout(resolve, 30))
agent.cancel({ kind: 'user' }, { keepInbox: true })
const steer = createUserMessage({ content: [{ type: 'text', text: 'steer me' }], source: { kind: 'user' } })
agent.steer(steer) // latched behind the aborted activity
agent.inbox.remove(steer.id) // the wake is retracted before convergence
await agent.whenIdle()
expect(userTexts(agent)).toEqual(['active'])
expect(adapter.requests).toHaveLength(1)
expect(agent.inbox.nextTurn).toHaveLength(0)
expect(agent.status).toBe('idle')
// No replay with nothing to run: the latched message is gone, so no
// empty follow-up turn is recorded.
expect(agent.session.events.filter(e => e.type === 'turn/start')).toHaveLength(1)
})
it('latches a wake arriving deep into a slow abort convergence', async () => {
// The stream notices the abort only after 50ms, so the driver stays in
// the abort-to-idle window long after `cancel()` returned: the wake must
// be latched across the whole window, not just the same-tick case.
const adapter = new MockAdapter(['hang-slow', textResponse('B reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('slow-convergence'), { provider: 'mock', model: 'mock' })
send(agent, 'A')
await new Promise(resolve => setTimeout(resolve, 30))
agent.cancel({ kind: 'user' }, { keepInbox: true })
await new Promise(resolve => setTimeout(resolve, 10))
send(agent, 'B')
await agent.whenIdle()
expect(userTexts(agent)).toEqual(['A', 'B'])
expect(adapter.requests).toHaveLength(2)
expect(agent.inbox.nextTurn).toHaveLength(0)
})
it('does not latch a wake landing after disposal begins', async () => {
const adapter = new MockAdapter(['hang-slow', textResponse('late reply')])
const ctx = await harness(adapter)
const handle = await ctx.agents.create({
sessionId: SessionId('dispose-window-wake'),
agentOptions: { provider: 'mock', model: 'mock' },
})
const agent = handle.agent
send(agent, 'active')
await new Promise(resolve => setTimeout(resolve, 30))
// Dispose cancels with `{ kind: 'disposed' }`; a wake landing in the
// abort-to-idle window must not latch, so `whenIdle()` does not wait on
// a model turn over the session being torn down.
const disposal = handle.dispose()
setTimeout(() => { send(agent, 'late wake') }, 10)
await disposal
expect(adapter.requests).toHaveLength(1)
expect(userTexts(agent)).toEqual(['active'])
})
it('cancel after waking send closes its synchronously opened turn without a step', async () => {
const adapter = new MockAdapter([textResponse('should not run')])
const ctx = await harness(adapter)
@@ -228,7 +343,7 @@ describe('Agent.cancel()', () => {
expect(userTexts(agent)).toEqual(['first', 'later'])
})
it('replacement work queued after idle-listener cancellation waits for another wakeup', async () => {
it('replacement work queued after idle-listener cancellation replays at convergence', async () => {
const adapter = new MockAdapter([
textResponse('first reply'),
textResponse('replacement reply'),
@@ -253,9 +368,11 @@ describe('Agent.cancel()', () => {
if (replacementIdle === undefined) throw new Error('idle listener did not register replacement work')
await replacementIdle
expect(adapter.requests).toHaveLength(1)
expect(userTexts(agent)).toEqual(['first'])
expect(agent.inbox.nextTurn).toHaveLength(1)
// The wake sent after the cancel fired is latched: the surviving
// replacement runs at convergence without a third message.
expect(adapter.requests).toHaveLength(2)
expect(userTexts(agent)).toEqual(['first', 'surviving replacement'])
expect(agent.inbox.nextTurn).toHaveLength(0)
const idle = waitForIdle(ctx, agent)
send(agent, 'wake it')
@@ -479,7 +596,7 @@ describe('Agent.cancel()', () => {
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
})
it('a running-listener cancellation parks replacement work until another wakeup', async () => {
it('a running-listener cancellation replays replacement work at convergence', async () => {
const adapter = new MockAdapter([textResponse('A reply'), textResponse('B reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -497,17 +614,20 @@ describe('Agent.cancel()', () => {
await idle
dispose()
expect(userTexts(agent)).toEqual([])
expect(agent.inbox.nextTurn).toHaveLength(1)
// B's wake was latched behind the cancelled driver: it runs on its own.
expect(userTexts(agent)).toEqual(['B'])
expect(agent.inbox.nextTurn).toHaveLength(0)
expect(adapter.requests).toHaveLength(1)
const replacementIdle = waitForIdle(ctx, agent)
send(agent, 'C')
await replacementIdle
expect(userTexts(agent)).toEqual(['B', 'C'])
expect(adapter.requests).toHaveLength(2)
expect(agent.session.events.filter(event => event.type === 'turn/end')).toHaveLength(2)
})
it('a prompt queued during pre-step cancellation waits for another wakeup', async () => {
it('a prompt queued during pre-step cancellation replays at convergence', async () => {
const adapter = new MockAdapter([textResponse('A reply'), textResponse('B reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -518,13 +638,15 @@ describe('Agent.cancel()', () => {
send(agent, 'B')
await idle
expect(userTexts(agent)).toEqual([])
expect(agent.inbox.nextTurn).toHaveLength(1)
expect(userTexts(agent)).toEqual(['B'])
expect(agent.inbox.nextTurn).toHaveLength(0)
expect(adapter.requests).toHaveLength(1)
const replacementIdle = waitForIdle(ctx, agent)
send(agent, 'C')
await replacementIdle
expect(userTexts(agent)).toEqual(['B', 'C'])
expect(adapter.requests).toHaveLength(2)
expect(agent.session.events.filter(event => event.type === 'turn/end')).toHaveLength(3)
})
@@ -556,7 +678,7 @@ describe('Agent.cancel()', () => {
expect(flat).not.toContain('steer text')
})
it('parks replacement work queued synchronously by an abort observer', async () => {
it('replays replacement work queued synchronously by an abort observer', async () => {
const adapter = new MockAdapter([
'hang',
textResponse('replacement reply'),
@@ -586,13 +708,15 @@ describe('Agent.cancel()', () => {
}),
])
expect(adapter.requests).toHaveLength(1)
expect(userTexts(agent)).toEqual(['original'])
expect(agent.inbox.nextTurn).toHaveLength(1)
// The abort-observer wake was latched: replacement runs at convergence,
// so the original turn is followed by a completed replacement turn.
expect(adapter.requests).toHaveLength(2)
expect(userTexts(agent)).toEqual(['original', 'replacement'])
expect(agent.inbox.nextTurn).toHaveLength(0)
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', reason: { kind: 'user' } }])
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }, { kind: 'completed' }])
const replacementIdle = waitForIdle(ctx, agent)
send(agent, 'wake it')

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

@@ -41,6 +41,14 @@ function send(agent: Agent, text: string) {
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
/** All user-message texts recorded in the log (to assert what actually ran). */
function userTexts(agent: Agent): string[] {
return agent.session.events
.filter(e => e.type === 'user/message')
.flatMap(e => e.type === 'user/message' ? e.data.content : [])
.flatMap(b => b.type === 'text' ? [b.text] : [])
}
describe('agent loop', () => {
it.each([0, -1, 1.5, Number.NaN, Number.MAX_SAFE_INTEGER + 1])(
'rejects invalid AgentOptions.maxTokens %s before publication',
@@ -70,7 +78,7 @@ describe('agent loop', () => {
})
it('cancels queued wakeup work together with an active maintenance task', async () => {
const adapter = new MockAdapter([textResponse('unused')])
const adapter = new MockAdapter([textResponse('park reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('cancel-maintenance-wakeup'), {
provider: 'mock',
@@ -87,15 +95,68 @@ describe('agent loop', () => {
})
await started.promise
send(agent, 'discard this wakeup')
agent.cancel({ kind: 'user' })
send(agent, 'park after cancellation')
send(agent, 'discard this wakeup') // latched behind the live maintenance task
agent.cancel({ kind: 'user' }) // drops the queue and the latch, aborts maintenance
send(agent, 'park after cancellation') // newer intent: re-latched, replays at convergence
await expect(maintenance).rejects.toThrow('maintenance aborted')
await agent.whenIdle()
expect(agent.inbox.nextTurn).toHaveLength(1)
// The pre-cancel wakeup is gone; the post-cancel wake replays at convergence.
expect(userTexts(agent)).toEqual(['park after cancellation'])
expect(agent.inbox.nextTurn).toHaveLength(0)
expect(adapter.requests).toHaveLength(1)
})
it('replays a wake latched behind maintenance at convergence', async () => {
const adapter = new MockAdapter([textResponse('wake reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('maintenance-wake-replay'), {
provider: 'mock',
model: 'mock',
})
const started = Promise.withResolvers<undefined>()
const finish = Promise.withResolvers<undefined>()
const maintenance = agent.runMaintenance(async () => {
started.resolve(undefined)
await finish.promise
})
await started.promise
send(agent, 'wake behind maintenance')
finish.resolve(undefined)
await maintenance
await agent.whenIdle()
expect(userTexts(agent)).toEqual(['wake behind maintenance'])
expect(adapter.requests).toHaveLength(1)
})
it('suppresses the replay when a latched maintenance wake is removed', async () => {
const adapter = new MockAdapter([])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('maintenance-wake-removed'), {
provider: 'mock',
model: 'mock',
})
const started = Promise.withResolvers<undefined>()
const finish = Promise.withResolvers<undefined>()
const maintenance = agent.runMaintenance(async () => {
started.resolve(undefined)
await finish.promise
})
await started.promise
const wake = createUserMessage({ content: [{ type: 'text', text: 'removed wake' }], source: { kind: 'user' } })
agent.followup(wake)
agent.inbox.remove(wake.id)
finish.resolve(undefined)
await maintenance
await agent.whenIdle()
expect(userTexts(agent)).toEqual([])
expect(adapter.requests).toEqual([])
agent.cancel({ kind: 'user' })
expect(agent.session.events.filter(e => e.type === 'turn/start')).toHaveLength(0)
})
it('runs a simple turn: queued message → model → idle, with ordered events', async () => {

18
packages/core/agent-loop/tests/mock-adapter.ts
View File

@@ -58,14 +58,16 @@ export function toolCallResponse(rawCallId: string, name: string, args: object,
/**
* Mock adapter driven by a script: each model call consumes the next entry.
* Records every request it receives for assertions. An entry may be a
* function to compute chunks from the request, or a 'hang' marker that
* streams one chunk then waits until aborted.
* function to compute chunks from the request, a 'hang' marker that
* streams one chunk then waits until aborted, or 'hang-slow' which takes
* 50ms to notice the abort — a stand-in for slow real-world teardown
* (LLM stream cancellation, tool unwinding).
*/
export class MockAdapter extends LlmAdapter {
requests: GenerateOptions[] = []
constructor(
private script: (StreamChunk[] | ((options: GenerateOptions) => StreamChunk[]) | 'hang')[],
private script: (StreamChunk[] | ((options: GenerateOptions) => StreamChunk[]) | 'hang' | 'hang-slow')[],
private readonly reasoning?: LlmModelReasoningInfo,
private readonly defaultMaxTokens?: number,
) {
@@ -98,6 +100,16 @@ export class MockAdapter extends LlmAdapter {
})
return
}
if (entry === 'hang-slow') {
yield { type: 'block-start', index: 0, blockType: 'text' }
yield { type: 'text-delta', index: 0, text: 'partial' }
await new Promise<void>((_resolve, reject) => {
const fail = (): void => { reject(new Error('aborted')) }
if (options.signal?.aborted) { setTimeout(fail, 50); return }
options.signal?.addEventListener('abort', () => { setTimeout(fail, 50) }, { once: true })
})
return
}
const chunks = typeof entry === 'function' ? entry(options) : entry
for (const chunk of chunks) {
if (options.signal?.aborted) throw new Error('aborted')