feat(agent): add queue-aware Agent.cancel() primitive

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.
2026-06-20 04:51:32 +08:00
parent 3d67a98291
commit c4bc6e0e38
12 changed files with 504 additions and 31 deletions
--- a/packages/agent-loop/tests/cancel.spec.ts
+++ b/packages/agent-loop/tests/cancel.spec.ts
@@ -0,0 +1,279 @@
+/**
+ * Tests for the queue-aware `Agent.cancel()` primitive (PR C). `cancel()` is the
+ * broad verb — it clears queued + steering work, aborts an in-flight step, and
+ * drops a turn about to start — whereas `abort()` kills only the current step.
+ * These tests exercise every window where a cancel can land (idle, pre-step,
+ * mid-step, continuation) and the marker's arm/reset rules that keep a cancel
+ * from leaking to a later prompt or hanging `whenIdle()`.
+ *
+ * @module dsh-agent-loop/tests/cancel
+ */
+
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import LlmService from '@deepseek-ai/dsh-llm'
+import SessionStore, { TurnEndReason } from '@deepseek-ai/dsh-session'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry from '@deepseek-ai/dsh-tools'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
+import { MockAdapter, textResponse } from './mock-adapter.ts'
+
+async function harness(adapter: MockAdapter) {
+  const ctx = new Context()
+  await ctx.plugin(LlmService)
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(SystemPrompt)
+  await ctx.plugin(ToolRegistry)
+  await ctx.plugin(AgentRegistry)
+  await ctx.plugin(AgentLoop, { agents: [] })
+  ctx.llm.registerAdapter(['mock'], adapter)
+  return ctx
+}
+
+function send(agent: ReactLoopAgent, text: string) {
+  agent.send([{ type: 'text', text }])
+}
+
+/** Resolve on the agent's next idle transition (event-based, not status poll). */
+function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
+  return new Promise((resolve) => {
+    const dispose = ctx.on('agent/status', (subject, status) => {
+      if (subject === agent && status === 'idle') { dispose(); resolve() }
+    })
+  })
+}
+
+/** All user-message texts recorded in the log (to assert what actually ran). */
+function userTexts(agent: ReactLoopAgent): 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.cancel()', () => {
+  it('cancel() on an idle agent with nothing queued is a no-op; the next prompt runs (F2 leak guard)', async () => {
+    const adapter = new MockAdapter([textResponse('reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    // 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')
+
+    send(agent, 'real prompt')
+    await waitForIdle(ctx, agent)
+
+    // The prompt ran: its user message is in the log and one turn completed.
+    expect(userTexts(agent)).toEqual(['real prompt'])
+    expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)
+  })
+
+  it('pre-step cancel drops the about-to-start turn (no turn is opened)', async () => {
+    const adapter = new MockAdapter([textResponse('should not run')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    // send() queues synchronously (status still idle, loop microtask not yet
+    // resumed). Cancel in that pre-step window: the queued turn must not run.
+    send(agent, 'drop me')
+    agent.cancel('pre-step')
+
+    // Give the loop a chance to wake and process the cancel.
+    await new Promise(r => setTimeout(r, 30))
+
+    // No turn was opened — the queued prompt was dropped, never recorded.
+    expect(userTexts(agent)).toEqual([])
+    expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
+    expect(agent.status).toBe('idle')
+  })
+
+  it('a whenIdle() waiter registered BEFORE a pre-step cancel resolves (F1 hang guard)', async () => {
+    const adapter = new MockAdapter([textResponse('x')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    // Queue work, then register a whenIdle() waiter while in the pre-step window
+    // (status idle, hasQueued true) — it does NOT take the fast path. Then cancel.
+    // The skip path must settle this waiter directly (no running→idle transition
+    // ever fires), or it would hang forever.
+    send(agent, 'q')
+    const idle = agent.whenIdle()
+    agent.cancel('pre-step')
+
+    // Must resolve (not hang). A timeout makes the failure a clear test failure.
+    await Promise.race([
+      idle,
+      new Promise((_r, reject) => setTimeout(() => { reject(new Error('whenIdle hung after pre-step cancel')) }, 1000)),
+    ])
+    expect(agent.status).toBe('idle')
+  })
+
+  it('cancel() mid-step aborts the in-flight model call; the turn ends aborted', async () => {
+    const adapter = new MockAdapter(['hang'])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    const reasons: TurnEndReason[] = []
+    ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    expect(agent.status).toBe('running')
+    agent.cancel('mid-step')
+    await waitForIdle(ctx, agent)
+
+    expect(reasons).toEqual([{ kind: 'aborted', reason: 'mid-step' }])
+  })
+
+  it('cancel() with no reason defaults to "cancelled" when aborting an in-flight step', async () => {
+    const adapter = new MockAdapter(['hang'])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    const reasons: TurnEndReason[] = []
+    ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    agent.cancel() // no reason → default 'cancelled'
+    await waitForIdle(ctx, agent)
+
+    expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled' }])
+  })
+
+  it('a prompt sent AFTER a cancelled turn settles runs normally (marker reset)', async () => {
+    const adapter = new MockAdapter(['hang', textResponse('second reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    // First turn hangs; cancel it mid-step.
+    send(agent, 'first')
+    await new Promise(r => setTimeout(r, 30))
+    agent.cancel('cancel first')
+    await waitForIdle(ctx, agent)
+
+    // The marker must have been reset after the cancelled turn — a fresh prompt
+    // runs to completion rather than being dropped by a stale marker.
+    send(agent, 'second')
+    await waitForIdle(ctx, agent)
+
+    expect(userTexts(agent)).toContain('second')
+    // The second turn completed (its reply was streamed).
+    const reasons = agent.session.events.filter(e => e.type === 'turn/end')
+    expect(reasons.length).toBe(2)
+  })
+
+  it('cancel from a synchronous agent/turn-start 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('a1', { model: 'mock' })
+
+    // A turn-start listener fires BEFORE any AbortController is installed for the
+    // step. Cancelling there must still drop the step (the turn-scoped marker,
+    // not abort(), is what catches this) — no model step runs.
+    let streamed = false
+    ctx.on('agent/stream-chunk', () => { streamed = true })
+    const dispose = ctx.on('agent/turn-start', (subject) => {
+      if (subject === agent) agent.cancel('from turn-start')
+    })
+
+    const reasons: TurnEndReason[] = []
+    ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    dispose()
+
+    // No step streamed (the model never ran), and the turn ended aborted.
+    expect(streamed).toBe(false)
+    expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled' }])
+  })
+
+  it('cancel during the continuation window ends the turn aborted and runs no further step', async () => {
+    // A continuation-waterfall listener cancels DURING the continuation decision
+    // (the finished step's AbortController is already cleared), and votes to
+    // continue — but the turn-scoped marker checked right after must end the turn
+    // `aborted` and run NO second step.
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    let steps = 0
+    ctx.on('agent/step-start', () => { steps += 1 })
+    const reasons: TurnEndReason[] = []
+    ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
+
+    let continued = false
+    ctx.on('agent/turn-continuation', async (subject, _turn, _default, next) => {
+      if (subject === agent && !continued) {
+        continued = true
+        agent.cancel('from continuation')
+        return true // vote to continue — the post-waterfall marker check must override
+      }
+      return next()
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    // Only ONE step ran (the second was cancelled in the continuation window),
+    // and the turn ended aborted.
+    expect(steps).toBe(1)
+    expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled' }])
+  })
+
+  it('cancel from a synchronous agent/status(running) listener drops the turn (window 2)', async () => {
+    const adapter = new MockAdapter([textResponse('should not run')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    // setStatus('running') emits agent/status SYNCHRONOUSLY, so a running
+    // listener can cancel in the gap between the loop's pre-step check and
+    // runTurn. The second check (after the running flip) must drop the turn —
+    // runTurn would otherwise throw on the now-empty queue.
+    let streamed = false
+    ctx.on('agent/stream-chunk', () => { streamed = true })
+    const dispose = ctx.on('agent/status', (subject, status) => {
+      if (subject === agent && status === 'running') agent.cancel('from running listener')
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    dispose()
+
+    // No turn opened, no step streamed, and a later prompt still runs (the marker
+    // was reset).
+    expect(streamed).toBe(false)
+    expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
+  })
+
+  it("cancel clears the turn's steering — it is not re-enqueued as a fresh turn", async () => {
+    const adapter = new MockAdapter(['hang'])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    send(agent, 'go')
+    await new Promise(r => setTimeout(r, 30))
+    expect(agent.status).toBe('running')
+    // 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')
+    await waitForIdle(ctx, agent)
+
+    // After the cancelled turn settles, the agent is idle with NO follow-up turn
+    // started from the dropped steering.
+    await new Promise(r => setTimeout(r, 30))
+    expect(agent.status).toBe('idle')
+    const turnStarts = agent.session.events.filter(e => e.type === 'turn/start')
+    expect(turnStarts.length).toBe(1) // only the original (cancelled) turn
+    // The steering text was dropped — it never reached the log.
+    const flat = agent.session.events
+      .filter(e => e.type === 'steering/message')
+      .flatMap(e => e.type === 'steering/message' ? e.data.content : [])
+      .flatMap(b => b.type === 'text' ? [b.text] : [])
+    expect(flat).not.toContain('steer text')
+  })
+})