|
|
@@ -1,9 +1,8 @@
|
|
|
/**
|
|
|
* Tests for the queue-aware `Agent.cancel()` primitive. `cancel()` is the broad verb — it
|
|
|
- * clears queued + steering work, aborts an in-flight step, and drops a turn about to start —
|
|
|
- * whereas a bare step abort (the loop's private `AbortController`) kills only the current step
|
|
|
- * and leaves the queue intact. The suite covers every landing window plus marker
|
|
|
- * reset and `whenIdle()` quiescence.
|
|
|
+ * clears queued + steering work, aborts the active turn, and drops work not yet claimed by the
|
|
|
+ * driver without leaking cancellation into a replacement prompt. The suite covers every landing
|
|
|
+ * window plus signal reset and `whenIdle()` quiescence.
|
|
|
* @module dsh-agent-loop/tests/cancel
|
|
|
*/
|
|
|
|
|
|
@@ -61,22 +60,22 @@ describe('Agent.cancel()', () => {
|
|
|
const agent = ctx.agentLoop.create(SessionId('cancel-event'), { provider: 'mock', model: 'mock' })
|
|
|
const warned = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
|
|
const seen: string[] = []
|
|
|
- ctx.on('agent/cancel-requested', (subject, reason) => {
|
|
|
+ ctx.on('agent/cancel-requested', (subject, cause) => {
|
|
|
if (subject !== agent) return
|
|
|
- seen.push(`first:${reason}`)
|
|
|
+ seen.push(`first:${cause.kind}`)
|
|
|
subject.send([{ type: 'text', text: 'queued by cancel observer' }])
|
|
|
throw new Error('observer failed')
|
|
|
})
|
|
|
- ctx.on('agent/cancel-requested', (subject, reason) => {
|
|
|
- if (subject === agent) seen.push(`second:${reason}`)
|
|
|
+ ctx.on('agent/cancel-requested', (subject, cause) => {
|
|
|
+ if (subject === agent) seen.push(`second:${cause.kind}`)
|
|
|
})
|
|
|
|
|
|
send(agent, 'drop me')
|
|
|
agent.cancel()
|
|
|
await new Promise(resolve => setTimeout(resolve, 30))
|
|
|
- agent.cancel('idle no-op')
|
|
|
+ agent.cancel({ kind: 'parent' })
|
|
|
|
|
|
- expect(seen).toEqual(['first:cancelled', 'second:cancelled'])
|
|
|
+ expect(seen).toEqual(['first:user', 'second:user'])
|
|
|
expect(userTexts(agent)).toEqual([])
|
|
|
expect(adapter.requests).toHaveLength(0)
|
|
|
expect(warned).toHaveBeenCalledWith(expect.stringContaining('agent/cancel-requested'))
|
|
|
@@ -89,7 +88,7 @@ describe('Agent.cancel()', () => {
|
|
|
|
|
|
// The loop is parked at the idle wait with nothing queued. A cancel here must
|
|
|
// NOT arm the marker — otherwise the next legitimate prompt would be dropped.
|
|
|
- agent.cancel('nothing to cancel')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
|
|
|
send(agent, 'real prompt')
|
|
|
await waitForIdle(ctx, agent)
|
|
|
@@ -108,7 +107,7 @@ describe('Agent.cancel()', () => {
|
|
|
// resumed). Cancel in that pre-step window: the queued turn must not run.
|
|
|
send(agent, 'drop me first')
|
|
|
send(agent, 'drop me second')
|
|
|
- agent.cancel('pre-step')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
|
|
|
// Give the loop a chance to wake and process the cancel.
|
|
|
await new Promise(r => setTimeout(r, 30))
|
|
|
@@ -157,7 +156,7 @@ describe('Agent.cancel()', () => {
|
|
|
// drops the turn before it runs; the skip path must settle it directly.
|
|
|
send(agent, 'q')
|
|
|
const idle = agent.whenIdle()
|
|
|
- agent.cancel('pre-step')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
|
|
|
// Must resolve (not hang). A timeout makes the failure a clear test failure.
|
|
|
await Promise.race([
|
|
|
@@ -186,7 +185,7 @@ describe('Agent.cancel()', () => {
|
|
|
// before its resolved waitForQueued continuation checks cancellation.
|
|
|
queueMicrotask(() => {
|
|
|
queueMicrotask(() => {
|
|
|
- agent.cancel('between turns')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
cancelled.resolve(undefined)
|
|
|
})
|
|
|
})
|
|
|
@@ -236,7 +235,7 @@ describe('Agent.cancel()', () => {
|
|
|
ctx.on('agent/error', (subject, _turn, _step, error) => {
|
|
|
if (subject !== agent || error.message !== 'first flush failed') return
|
|
|
queueMicrotask(() => {
|
|
|
- queueMicrotask(() => { agent.cancel('between turns') })
|
|
|
+ queueMicrotask(() => { agent.cancel({ kind: 'user' }) })
|
|
|
})
|
|
|
})
|
|
|
|
|
|
@@ -277,7 +276,7 @@ describe('Agent.cancel()', () => {
|
|
|
requests: adapter.requests.length,
|
|
|
turns: agent.session.events.filter(event => event.type === 'turn/start').length,
|
|
|
}))
|
|
|
- agent.cancel('idle listener')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
replacementRegistered.resolve(undefined)
|
|
|
})
|
|
|
|
|
|
@@ -307,7 +306,7 @@ describe('Agent.cancel()', () => {
|
|
|
ctx.on('agent/status', (subject, status) => {
|
|
|
if (subject !== agent || status !== 'idle' || replacementIdle !== undefined) return
|
|
|
send(agent, 'cancelled replacement')
|
|
|
- agent.cancel('idle listener')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
send(agent, 'surviving replacement')
|
|
|
replacementIdle = agent.whenIdle()
|
|
|
replacementRegistered.resolve(undefined)
|
|
|
@@ -334,16 +333,16 @@ describe('Agent.cancel()', () => {
|
|
|
await new Promise(r => setTimeout(r, 30))
|
|
|
expect(agent.status).toBe('running')
|
|
|
send(agent, 'queued tail')
|
|
|
- agent.cancel('mid-step')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
await waitForIdle(ctx, agent)
|
|
|
|
|
|
- expect(reasons).toEqual([{ kind: 'aborted', reason: 'mid-step' }])
|
|
|
+ expect(reasons).toEqual([{ kind: 'aborted' }])
|
|
|
expect(userTexts(agent)).toEqual(['go'])
|
|
|
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
|
|
|
expect(adapter.requests).toHaveLength(1)
|
|
|
})
|
|
|
|
|
|
- it('cancel() with no reason defaults to "cancelled" when aborting an in-flight step', async () => {
|
|
|
+ it('cancel() with no cause defaults to user when aborting an active turn', async () => {
|
|
|
const adapter = new MockAdapter(['hang'])
|
|
|
const ctx = await harness(adapter)
|
|
|
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
|
|
@@ -353,10 +352,10 @@ describe('Agent.cancel()', () => {
|
|
|
|
|
|
send(agent, 'go')
|
|
|
await new Promise(r => setTimeout(r, 30))
|
|
|
- agent.cancel() // no reason → default 'cancelled'
|
|
|
+ agent.cancel()
|
|
|
await waitForIdle(ctx, agent)
|
|
|
|
|
|
- expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled' }])
|
|
|
+ expect(reasons).toEqual([{ kind: 'aborted' }])
|
|
|
})
|
|
|
|
|
|
it('cancel from an assistant/message observer skips execution but balances replay', async () => {
|
|
|
@@ -378,7 +377,7 @@ describe('Agent.cancel()', () => {
|
|
|
const agent = ctx.agentLoop.create(SessionId('cancel-after-assistant-message'), { provider: 'mock', model: 'mock' })
|
|
|
const dispose = ctx.on('session/event', (session, event) => {
|
|
|
if (session === agent.session && event.type === 'assistant/message') {
|
|
|
- agent.cancel('cancelled after assistant message')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
}
|
|
|
})
|
|
|
|
|
|
@@ -390,7 +389,7 @@ describe('Agent.cancel()', () => {
|
|
|
dispose()
|
|
|
|
|
|
expect(executions).toBe(0)
|
|
|
- expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled after assistant message' }])
|
|
|
+ expect(reasons).toEqual([{ kind: 'aborted' }])
|
|
|
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')
|
|
|
@@ -407,7 +406,7 @@ describe('Agent.cancel()', () => {
|
|
|
.find(block => block.type === 'tool-result')
|
|
|
expect(replayedResult).toMatchObject({ toolCallId: 'c1', isError: true })
|
|
|
expect(reasons).toEqual([
|
|
|
- { kind: 'aborted', reason: 'cancelled after assistant message' },
|
|
|
+ { kind: 'aborted' },
|
|
|
{ kind: 'completed' },
|
|
|
])
|
|
|
})
|
|
|
@@ -420,7 +419,7 @@ describe('Agent.cancel()', () => {
|
|
|
// First turn hangs; cancel it mid-step.
|
|
|
send(agent, 'first')
|
|
|
await new Promise(r => setTimeout(r, 30))
|
|
|
- agent.cancel('cancel first')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
await waitForIdle(ctx, agent)
|
|
|
|
|
|
// The marker must have been reset after the cancelled turn — a fresh prompt
|
|
|
@@ -445,7 +444,7 @@ describe('Agent.cancel()', () => {
|
|
|
let streamed = false
|
|
|
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
|
|
|
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next) => {
|
|
|
- agent.cancel('from prefix composition')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
return next()
|
|
|
})
|
|
|
|
|
|
@@ -456,7 +455,7 @@ describe('Agent.cancel()', () => {
|
|
|
await waitForIdle(ctx, agent)
|
|
|
|
|
|
expect(streamed).toBe(false)
|
|
|
- expect(reasons).toEqual([{ kind: 'aborted', reason: 'from prefix composition' }])
|
|
|
+ expect(reasons).toEqual([{ kind: 'aborted' }])
|
|
|
})
|
|
|
|
|
|
it('disposal from inside the agent/session-prefix waterfall ends the turn disposed (prefix-composition window)', async () => {
|
|
|
@@ -508,7 +507,7 @@ describe('Agent.cancel()', () => {
|
|
|
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
|
|
|
compositions += 1
|
|
|
if (compositions === 1) {
|
|
|
- agent.cancel('mid-composition')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
return next()
|
|
|
}
|
|
|
return [opener, ...await next()]
|
|
|
@@ -536,7 +535,7 @@ describe('Agent.cancel()', () => {
|
|
|
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('from turn-start')
|
|
|
+ if (session === agent.session && event.type === 'turn/start') agent.cancel({ kind: 'user' })
|
|
|
})
|
|
|
|
|
|
const reasons: TurnEndReason[] = []
|
|
|
@@ -547,10 +546,10 @@ describe('Agent.cancel()', () => {
|
|
|
dispose()
|
|
|
|
|
|
// No step streamed (the model never ran), and the turn ended aborted with
|
|
|
- // the CALLER's reason — the marker carries `cancel(reason)` through even
|
|
|
+ // 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: 'from turn-start' }])
|
|
|
+ expect(reasons).toEqual([{ kind: 'aborted' }])
|
|
|
})
|
|
|
|
|
|
it('cancel from a synchronous step/start session-event listener drops the step (post-step-start window)', async () => {
|
|
|
@@ -565,7 +564,7 @@ describe('Agent.cancel()', () => {
|
|
|
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 === 'step/start') agent.cancel('from step-start')
|
|
|
+ if (session === agent.session && event.type === 'step/start') agent.cancel({ kind: 'user' })
|
|
|
})
|
|
|
|
|
|
const reasons: TurnEndReason[] = []
|
|
|
@@ -575,10 +574,10 @@ describe('Agent.cancel()', () => {
|
|
|
await waitForIdle(ctx, agent)
|
|
|
dispose()
|
|
|
|
|
|
- // No step streamed, the turn ended aborted with the caller's reason, and the
|
|
|
+ // 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', reason: 'from step-start' }])
|
|
|
+ expect(reasons).toEqual([{ kind: 'aborted' }])
|
|
|
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)
|
|
|
})
|
|
|
@@ -636,11 +635,11 @@ describe('Agent.cancel()', () => {
|
|
|
})
|
|
|
|
|
|
let continued = false
|
|
|
- ctx.on('agent/turn-continuation', async (subject, _turn, _default, next) => {
|
|
|
+ ctx.on('agent/turn-continuation', async (subject, _turn, _default, _signal, next) => {
|
|
|
if (subject === agent && !continued) {
|
|
|
continued = true
|
|
|
- agent.cancel('from continuation')
|
|
|
- return { action: 'continue' as const } // vote to continue — the post-waterfall marker check must override
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
+ return { action: 'continue' as const }
|
|
|
}
|
|
|
return next()
|
|
|
})
|
|
|
@@ -649,10 +648,9 @@ describe('Agent.cancel()', () => {
|
|
|
await waitForIdle(ctx, agent)
|
|
|
|
|
|
// Only ONE step ran (the second was cancelled in the continuation window),
|
|
|
- // and the turn ended aborted with the CALLER's reason (carried by the
|
|
|
- // marker, since the finished step's AbortController was already cleared).
|
|
|
+ // and the shared turn signal classified the durable outcome as aborted.
|
|
|
expect(steps).toBe(1)
|
|
|
- expect(reasons).toEqual([{ kind: 'aborted', reason: 'from continuation' }])
|
|
|
+ expect(reasons).toEqual([{ kind: 'aborted' }])
|
|
|
})
|
|
|
|
|
|
it('cancel from a synchronous agent/status(running) listener drops the turn (window 2)', async () => {
|
|
|
@@ -665,7 +663,7 @@ describe('Agent.cancel()', () => {
|
|
|
let streamed = false
|
|
|
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
|
|
|
const dispose = ctx.on('agent/status', (subject, status) => {
|
|
|
- if (subject === agent && status === 'running') agent.cancel('from running listener')
|
|
|
+ if (subject === agent && status === 'running') agent.cancel({ kind: 'user' })
|
|
|
})
|
|
|
|
|
|
send(agent, 'go')
|
|
|
@@ -688,7 +686,7 @@ describe('Agent.cancel()', () => {
|
|
|
const dispose = ctx.on('agent/status', (subject, status) => {
|
|
|
if (subject !== agent || status !== 'running' || replaced) return
|
|
|
replaced = true
|
|
|
- agent.cancel('drop A')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
send(agent, 'B')
|
|
|
})
|
|
|
|
|
|
@@ -713,7 +711,7 @@ describe('Agent.cancel()', () => {
|
|
|
|
|
|
send(agent, 'A') // queues A (status still idle, loop microtask pending)
|
|
|
const idle = agent.whenIdle() // registers a waiter (idle + hasQueued → no fast path)
|
|
|
- agent.cancel('drop A') // arms marker, clears A
|
|
|
+ agent.cancel({ kind: 'user' }) // arms marker, clears A
|
|
|
send(agent, 'B') // B races in before the loop resumes
|
|
|
|
|
|
// whenIdle() must resolve only after B's turn fully ran — by which point B's user message
|
|
|
@@ -736,7 +734,7 @@ describe('Agent.cancel()', () => {
|
|
|
// Steer (joins the running turn's steering FIFO), then cancel: the steering
|
|
|
// must be dropped, NOT re-enqueued as a new queued turn.
|
|
|
agent.steer([{ type: 'text', text: 'steer text' }])
|
|
|
- agent.cancel('cancel with steering')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
await waitForIdle(ctx, agent)
|
|
|
|
|
|
// After the cancelled turn settles, the agent is idle with NO follow-up turn
|
|
|
@@ -752,4 +750,228 @@ describe('Agent.cancel()', () => {
|
|
|
.flatMap(b => b.type === 'text' ? [b.text] : [])
|
|
|
expect(flat).not.toContain('steer text')
|
|
|
})
|
|
|
+
|
|
|
+ it('keeps replacement work queued synchronously by an abort observer', async () => {
|
|
|
+ const adapter = new MockAdapter(['hang', textResponse('replacement reply')])
|
|
|
+ const ctx = await harness(adapter)
|
|
|
+ const agent = ctx.agentLoop.create(SessionId('abort-observer-replacement'), { provider: 'mock', model: 'mock' })
|
|
|
+
|
|
|
+ send(agent, 'original')
|
|
|
+ await expect.poll(() => adapter.requests.length).toBe(1)
|
|
|
+ const signal = adapter.requests[0]?.signal
|
|
|
+ if (signal === undefined) throw new Error('model request omitted its turn signal')
|
|
|
+ signal.addEventListener('abort', () => { send(agent, 'replacement') }, { once: true })
|
|
|
+ const idle = waitForIdle(ctx, agent)
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
+ await Promise.race([
|
|
|
+ idle,
|
|
|
+ new Promise((_resolve, reject) => {
|
|
|
+ setTimeout(() => {
|
|
|
+ reject(new Error(`replacement did not settle: ${JSON.stringify({
|
|
|
+ status: agent.status,
|
|
|
+ requests: adapter.requests.length,
|
|
|
+ users: userTexts(agent),
|
|
|
+ events: agent.session.events.map(event => event.type),
|
|
|
+ })}`))
|
|
|
+ }, 1000)
|
|
|
+ }),
|
|
|
+ ])
|
|
|
+
|
|
|
+ expect(adapter.requests).toHaveLength(2)
|
|
|
+ expect(userTexts(agent)).toEqual(['original', 'replacement'])
|
|
|
+ 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' }])
|
|
|
+ })
|
|
|
+
|
|
|
+ it('keeps the first typed cause for an active turn and detaches the runtime reason', async () => {
|
|
|
+ const adapter = new MockAdapter(['hang'])
|
|
|
+ const ctx = await harness(adapter)
|
|
|
+ const agent = ctx.agentLoop.create(SessionId('typed-first-wins'), { provider: 'mock', model: 'mock' })
|
|
|
+ const supplied: { kind: 'parent' | 'user' } = { kind: 'parent' }
|
|
|
+
|
|
|
+ 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)
|
|
|
+ const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
|
|
|
+ expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
|
|
|
+ })
|
|
|
+
|
|
|
+ it('retires turn cancellation before terminal publication and a blocked durability flush', async () => {
|
|
|
+ const adapter = new MockAdapter([textResponse('done')])
|
|
|
+ const ctx = await harness(adapter)
|
|
|
+ const agent = ctx.agentLoop.create(SessionId('terminal-cancellation-authority'), { provider: 'mock', model: 'mock' })
|
|
|
+ const flushStarted = Promise.withResolvers<undefined>()
|
|
|
+ const releaseFlush = Promise.withResolvers<undefined>()
|
|
|
+ let abortedDuringTurnEnd: boolean | undefined
|
|
|
+ let cancelNotifications = 0
|
|
|
+
|
|
|
+ ctx.on('agent/cancel-requested', (subject) => {
|
|
|
+ if (subject === agent) cancelNotifications += 1
|
|
|
+ })
|
|
|
+ ctx.on('session/event', (session, event) => {
|
|
|
+ if (session !== agent.session || event.type !== 'turn/end') return
|
|
|
+ const signal = adapter.requests[0]?.signal
|
|
|
+ if (signal === undefined) throw new Error('model request omitted its turn signal')
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
+ abortedDuringTurnEnd = signal.aborted
|
|
|
+ })
|
|
|
+ ctx.on('session/flush', async (session) => {
|
|
|
+ if (session !== agent.session) return
|
|
|
+ flushStarted.resolve(undefined)
|
|
|
+ await releaseFlush.promise
|
|
|
+ })
|
|
|
+
|
|
|
+ send(agent, 'finish before persistence drains')
|
|
|
+ await flushStarted.promise
|
|
|
+ const signal = adapter.requests[0]?.signal
|
|
|
+ if (signal === undefined) throw new Error('model request omitted its turn signal')
|
|
|
+ const idle = agent.whenIdle()
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
+
|
|
|
+ expect(abortedDuringTurnEnd).toBe(false)
|
|
|
+ expect(signal.aborted).toBe(false)
|
|
|
+ expect(cancelNotifications).toBe(0)
|
|
|
+ expect(agent.session.events.findLast(event => event.type === 'turn/end')).toMatchObject({
|
|
|
+ data: { reason: { kind: 'completed' } },
|
|
|
+ })
|
|
|
+
|
|
|
+ releaseFlush.resolve(undefined)
|
|
|
+ await idle
|
|
|
+ expect(agent.status).toBe('idle')
|
|
|
+ })
|
|
|
+
|
|
|
+ it('records disposed when lifecycle teardown races an already-requested cancel', async () => {
|
|
|
+ const adapter = new MockAdapter(['hang'])
|
|
|
+ const ctx = await harness(adapter)
|
|
|
+ const handle = await ctx.agents.create({
|
|
|
+ sessionId: SessionId('cancel-dispose-race'),
|
|
|
+ agentOptions: { provider: 'mock', model: 'mock' },
|
|
|
+ })
|
|
|
+ const { agent } = handle
|
|
|
+
|
|
|
+ send(agent, 'go')
|
|
|
+ await expect.poll(() => adapter.requests.length).toBe(1)
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
+ await handle.dispose()
|
|
|
+
|
|
|
+ const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
|
|
|
+ expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
|
|
|
+ })
|
|
|
+
|
|
|
+ it.each([
|
|
|
+ 'prompt-submit',
|
|
|
+ 'system-prompt',
|
|
|
+ 'session-prefix',
|
|
|
+ 'pre-step',
|
|
|
+ 'request',
|
|
|
+ 'step-result',
|
|
|
+ 'post-step',
|
|
|
+ 'turn-continuation',
|
|
|
+ 'turn-stop',
|
|
|
+ 'tool',
|
|
|
+ ] as const)('lets a cooperative %s boundary settle from the explicit turn signal', async (stage) => {
|
|
|
+ const adapter = new MockAdapter(stage === 'tool'
|
|
|
+ ? [toolCallResponse('blocked-tool', 'blocked', {})]
|
|
|
+ : [textResponse('done')])
|
|
|
+ const ctx = await harness(adapter)
|
|
|
+ const agent = ctx.agentLoop.create(SessionId(`cooperative-${stage}`), { provider: 'mock', model: 'mock' })
|
|
|
+ const started = Promise.withResolvers<undefined>()
|
|
|
+ const blockUntilAbort = async (signal: AbortSignal): Promise<void> => {
|
|
|
+ started.resolve(undefined)
|
|
|
+ if (signal.aborted) return
|
|
|
+ await new Promise<void>((resolve) => {
|
|
|
+ signal.addEventListener('abort', () => { resolve() }, { once: true })
|
|
|
+ })
|
|
|
+ }
|
|
|
+
|
|
|
+ switch (stage) {
|
|
|
+ case 'prompt-submit':
|
|
|
+ ctx.on('agent/prompt-submit', async (subject, _content, _source, signal, next) => {
|
|
|
+ if (subject === agent) await blockUntilAbort(signal)
|
|
|
+ return next()
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'system-prompt':
|
|
|
+ ctx.on('system-prompt/assemble', async (_assembly, context, next) => {
|
|
|
+ if (context.agent === agent) {
|
|
|
+ if (context.signal === undefined) throw new Error('turn assembly omitted its signal')
|
|
|
+ await blockUntilAbort(context.signal)
|
|
|
+ }
|
|
|
+ return next()
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'session-prefix':
|
|
|
+ ctx.on('agent/session-prefix', async (subject, _prefix, signal, next) => {
|
|
|
+ if (subject === agent) await blockUntilAbort(signal)
|
|
|
+ return next()
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'pre-step':
|
|
|
+ ctx.on('agent/pre-step', async (subject, _turn, _step, signal) => {
|
|
|
+ if (subject === agent) await blockUntilAbort(signal)
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'request':
|
|
|
+ ctx.on('agent/request', async (subject, _turn, _step, _config, signal, next) => {
|
|
|
+ if (subject === agent) await blockUntilAbort(signal)
|
|
|
+ return next()
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'step-result':
|
|
|
+ ctx.on('agent/step-result', async (subject, _turn, _step, _message, signal, next) => {
|
|
|
+ if (subject === agent) await blockUntilAbort(signal)
|
|
|
+ return next()
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'post-step':
|
|
|
+ ctx.on('agent/post-step', async (subject, _turn, _step, signal) => {
|
|
|
+ if (subject !== agent) return
|
|
|
+ await blockUntilAbort(signal)
|
|
|
+ throw new Error('post-step failed after cancellation')
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'turn-continuation':
|
|
|
+ ctx.on('agent/turn-continuation', async (subject, _turn, _decision, signal, next) => {
|
|
|
+ if (subject === agent) await blockUntilAbort(signal)
|
|
|
+ return next()
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'turn-stop':
|
|
|
+ ctx.on('agent/turn-stop', async (subject, _turn, signal) => {
|
|
|
+ if (subject === agent) await blockUntilAbort(signal)
|
|
|
+ })
|
|
|
+ break
|
|
|
+ case 'tool':
|
|
|
+ ctx.tools.register(defineTool({
|
|
|
+ name: 'blocked',
|
|
|
+ description: 'wait for cancellation',
|
|
|
+ parameters: {},
|
|
|
+ execute: async (_args, exec) => {
|
|
|
+ if (exec.signal === undefined) throw new Error('tool execution omitted its signal')
|
|
|
+ await blockUntilAbort(exec.signal)
|
|
|
+ return [{ type: 'text', text: 'cancelled' }]
|
|
|
+ },
|
|
|
+ }))
|
|
|
+ break
|
|
|
+ }
|
|
|
+
|
|
|
+ send(agent, 'go')
|
|
|
+ await started.promise
|
|
|
+ const idle = waitForIdle(ctx, agent)
|
|
|
+ agent.cancel({ kind: 'user' })
|
|
|
+ await idle
|
|
|
+ const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
|
|
|
+ expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
|
|
|
+ await ctx.fiber.dispose()
|
|
|
+ })
|
|
|
})
|