mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
abort() only kills the in-flight step, so a queued-but-not-yet-started prompt
ran to completion after a cancel and a prompt accepted right after could be
batched into the cancelled turn (the loop merges queued messages into one turn).
This closes TODO(rfc010-cancel-prestep) with a distinct cancel() verb.
cancel() clears the queued + steering FIFOs, aborts the in-flight step, and
drives a turn-scoped marker on the LoopHandle that the driver checks at EVERY
point a turn could start or continue:
- right after the idle wait (window 1): drop the about-to-run turn and settle
whenIdle() waiters directly (no running→idle transition fires, and no
agent/status is emitted, so an ACP listener can't see a spurious idle that
resolves a freshly-queued prompt as cancelled);
- after the synchronous setStatus('running') emit (window 2): a running listener
can cancel in the gap before runTurn;
- in the step-start window (before runStep, after setAbort): a synchronous
turn-start/step-start listener can cancel before any AbortController exists;
- at the continuation gate: a cancel during the continuation waterfall (the
finished step's controller already cleared) ends the turn aborted.
The marker is ARMED only when there is something to cancel (running, an
in-flight step, or queued/steering work) — an idle no-op cancel cannot leave it
set to drop a later prompt — and RESET unconditionally once per loop iteration,
so it governs exactly one turn and never leaks onto the next prompt (even when a
send() lands in the cancelled turn's flush window).
ACP session/cancel now maps to agent.cancel() (keeping the synchronous
settlePrompt). Teardown/disconnect still use abort('disposed') until PR D, so
the ACP README narrows the remaining best-effort window to teardown only.
Tests (agent-loop/cancel.spec.ts) cover every window unit-level (the F1 hang
guard: a whenIdle() waiter registered before a pre-step cancel resolves; the F2
leak guard: idle cancel then a prompt runs; mid-step, continuation, both
pre-step windows, turn-start-listener, steering-cleared, marker-reset). ACP
turns.spec.ts adds the through-bridge tests with NO intervening whenIdle (idle
cancel→prompt runs; mid-stream cancel→immediate next prompt runs) and updates
the stale pre-step test to the queue-aware guarantee. The existing cancel
snapshot golden is byte-identical (it drives the new cancel() path end-to-end
through the real subprocess), so no new golden is needed. 100% coverage.
134 lines
5.1 KiB
TypeScript
134 lines
5.1 KiB
TypeScript
import { describe, expect, it } from 'vitest'
|
|
import { Context } from 'cordis'
|
|
import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
|
import AgentRegistry, { Agent, AgentId } from '@deepseek-ai/dsh-agent'
|
|
|
|
function stubAgent(rawId: string): Agent {
|
|
const id = AgentId(rawId)
|
|
return {
|
|
id,
|
|
options: {},
|
|
session: new Session(SessionId(`${id}-session`)),
|
|
status: 'idle',
|
|
send() {},
|
|
steer() {},
|
|
inject() {},
|
|
abort() {},
|
|
cancel() {},
|
|
whenIdle() { return Promise.resolve() },
|
|
}
|
|
}
|
|
|
|
describe('AgentRegistry', () => {
|
|
it('registers agents and emits created/disposed events', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(AgentRegistry)
|
|
|
|
const created: string[] = []
|
|
const disposed: string[] = []
|
|
ctx.on('agent/created', agent => void created.push(agent.id))
|
|
ctx.on('agent/disposed', agent => void disposed.push(agent.id))
|
|
|
|
const agent = stubAgent('a1')
|
|
const dispose = ctx.agents.register(agent)
|
|
expect(created).toEqual(['a1'])
|
|
expect(ctx.agents.get('a1')).toBe(agent)
|
|
expect(ctx.agents.list()).toEqual([agent])
|
|
|
|
dispose()
|
|
expect(disposed).toEqual(['a1'])
|
|
expect(ctx.agents.get('a1')).toBeUndefined()
|
|
})
|
|
|
|
it('rejects duplicate ids and unregisters on fiber dispose (HMR safety)', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(AgentRegistry)
|
|
ctx.agents.register(stubAgent('main'))
|
|
expect(() => ctx.agents.register(stubAgent('main'))).toThrow('already registered')
|
|
|
|
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
|
inner.agents.register(stubAgent('scoped'))
|
|
}, { inject: ['agents'] }))
|
|
expect(ctx.agents.list().map(a => a.id)).toEqual(['main', 'scoped'])
|
|
|
|
await fiber.dispose()
|
|
expect(ctx.agents.list().map(a => a.id)).toEqual(['main'])
|
|
})
|
|
|
|
it('rolls back the agent entry when an agent/created listener throws (P1-1)', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(AgentRegistry)
|
|
|
|
let threw = false
|
|
ctx.on('agent/created', () => {
|
|
if (!threw) { threw = true; throw new Error('boom created listener') }
|
|
})
|
|
|
|
// The throwing emit must roll the entry back, not leak it.
|
|
expect(() => ctx.agents.register(stubAgent('main'))).toThrow('boom created listener')
|
|
expect(ctx.agents.get('main')).toBeUndefined() // rolled back, not leaked
|
|
|
|
// A subsequent listener-free register of the SAME id succeeds and is
|
|
// tracked exactly once (the duplicate-id check is not wedged).
|
|
const dispose = ctx.agents.register(stubAgent('main'))
|
|
expect(ctx.agents.list().map(a => a.id)).toEqual(['main'])
|
|
dispose()
|
|
expect(ctx.agents.get('main')).toBeUndefined()
|
|
})
|
|
})
|
|
|
|
describe('AgentRegistry factory seam', () => {
|
|
/** A stub AgentFactory that records calls and returns a stub agent. */
|
|
function stubFactory() {
|
|
const calls: { create: unknown[]; resume: unknown[] } = { create: [], resume: [] }
|
|
const factory: import('@deepseek-ai/dsh-agent').AgentFactory = {
|
|
createAgent(options) { calls.create.push(options); return stubAgent(options.agentId) },
|
|
resume(options) { calls.resume.push(options); return Promise.resolve(stubAgent(options.agentId)) },
|
|
}
|
|
return { factory, calls }
|
|
}
|
|
|
|
it('create()/resume() throw when no factory is registered', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(AgentRegistry)
|
|
expect(() => ctx.agents.create({ agentId: 'a', sessionId: 's' })).toThrow(/no agent factory/)
|
|
await expect(ctx.agents.resume({ agentId: 'a', resumeSessionId: 's' })).rejects.toThrow(/no agent factory/)
|
|
})
|
|
|
|
it('setFactory registers a factory; create/resume delegate to it', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(AgentRegistry)
|
|
const { factory, calls } = stubFactory()
|
|
ctx.agents.setFactory(factory)
|
|
|
|
const created = ctx.agents.create({ agentId: 'c1', sessionId: 'sess-1', meta: { cwd: '/w' } })
|
|
expect(created.id).toBe('c1')
|
|
expect(calls.create).toEqual([{ agentId: 'c1', sessionId: 'sess-1', meta: { cwd: '/w' } }])
|
|
|
|
const resumed = await ctx.agents.resume({ agentId: 'r1', resumeSessionId: 'old-sess' })
|
|
expect(resumed.id).toBe('r1')
|
|
expect(calls.resume).toEqual([{ agentId: 'r1', resumeSessionId: 'old-sess' }])
|
|
})
|
|
|
|
it('setFactory rejects a second factory', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(AgentRegistry)
|
|
ctx.agents.setFactory(stubFactory().factory)
|
|
expect(() => ctx.agents.setFactory(stubFactory().factory)).toThrow(/already registered/)
|
|
})
|
|
|
|
it('disposing the setFactory fiber clears the factory (HMR safety)', async () => {
|
|
const ctx = new Context()
|
|
await ctx.plugin(AgentRegistry)
|
|
let dispose!: () => void
|
|
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
|
dispose = inner.agents.setFactory(stubFactory().factory)
|
|
}, { inject: ['agents'] }))
|
|
expect(() => ctx.agents.create({ agentId: 'a', sessionId: 's' })).not.toThrow()
|
|
void dispose
|
|
await fiber.dispose()
|
|
// factory slot cleared → create throws again
|
|
expect(() => ctx.agents.create({ agentId: 'a2', sessionId: 's2' })).toThrow(/no agent factory/)
|
|
})
|
|
})
|