Merge branch 'codex/tool-json-schema-dsl' into codex/canonical-tool-output
# Conflicts: # .agents/notes/implemented/feature/2026-06-30-interception-seams.md # docs/config-catalog.md # docs/cookbook/adding-a-tool.i18n.yaml # docs/cookbook/adding-a-tool.md # docs/cookbook/adding-a-tool.zh.md # docs/cordis-catalog/events.md # docs/cordis-catalog/services.md # docs/core-data-structures/tools.md # docs/event-producer-consumer.md # docs/persistence-catalog.md # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/stdout.expected.jsonl # packages/bash/tool-bash/src/index.ts # packages/core/agent-loop/src/tool-calls.ts # packages/core/agent-loop/tests/cancel.spec.ts # packages/core/agent-loop/tests/contract-regressions.spec.ts # packages/core/agent-loop/tests/tool-calls.spec.ts # packages/core/tools/README.md # packages/core/tools/src/index.ts # packages/core/tools/tests/code-mode.spec.ts # packages/core/tools/tests/tools.spec.ts # packages/fs/tool-fs-search/tests/integration.spec.ts # packages/fs/tool-fs-search/tests/tools.spec.ts # packages/fs/tool-fs/tests/integration.spec.ts # packages/mcp/mcp-client/src/tools.ts # packages/timeout/timeout-policy/tests/timeout-policy.spec.ts # packages/web/tool-web/tests/integration.spec.ts # packages/web/tool-web/tests/tool-web.spec.ts
This commit is contained in:
@@ -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
|
||||
*/
|
||||
|
||||
@@ -12,7 +11,7 @@ import { Context } from 'cordis'
|
||||
import LlmService, { type Message } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
|
||||
import ToolRegistry, { defineContentToolFixture, TOOL_ABORTED_BEFORE_DISPATCH } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
|
||||
@@ -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,14 +389,14 @@ 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')
|
||||
expect(result?.type === 'tool/result' ? result.data : undefined).toMatchObject({
|
||||
callId: 'c1',
|
||||
isError: true,
|
||||
error: { name: 'AbortError', code: 'ABORTED' },
|
||||
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
|
||||
})
|
||||
|
||||
send(agent, 'continue safely')
|
||||
@@ -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(defineContentToolFixture({
|
||||
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()
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user