c4bc6e0e38
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.
76 lines
2.2 KiB
TypeScript
76 lines
2.2 KiB
TypeScript
/**
|
|
* Per-agent message inbox: queued and steering FIFOs. Purely an in-memory
|
|
* mechanism of the loop driver — the public surface is `Agent.send()` and
|
|
* `Agent.steer()`.
|
|
*
|
|
* @module dsh-agent-loop/inbox
|
|
*/
|
|
|
|
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
|
|
|
/** One message waiting in an agent's inbox. */
|
|
export interface InboxMessage {
|
|
content: ContentBlock[]
|
|
source: MessageSource
|
|
}
|
|
|
|
/**
|
|
* Per-agent inbox: a queued FIFO (drained at turn start) and a steering FIFO
|
|
* (drained between steps of a running turn). Purely an in-memory mechanism of
|
|
* the loop — the public surface is `Agent.send()` / `Agent.steer()`.
|
|
*/
|
|
export class Inbox {
|
|
private queuedMessages: InboxMessage[] = []
|
|
private steeringMessages: InboxMessage[] = []
|
|
private wakeup: (() => void) | undefined
|
|
|
|
/** Resolves when a queued message arrives (used by the idle loop). */
|
|
get hasQueued(): boolean {
|
|
return this.queuedMessages.length > 0
|
|
}
|
|
|
|
get hasSteering(): boolean {
|
|
return this.steeringMessages.length > 0
|
|
}
|
|
|
|
enqueue(message: InboxMessage): void {
|
|
this.queuedMessages.push(message)
|
|
this.wakeup?.()
|
|
}
|
|
|
|
steer(message: InboxMessage): void {
|
|
this.steeringMessages.push(message)
|
|
}
|
|
|
|
/** Drain all queued messages (turn start). */
|
|
drainQueued(): InboxMessage[] {
|
|
return this.queuedMessages.splice(0)
|
|
}
|
|
|
|
/** Drain all steering messages (between steps). */
|
|
drainSteering(): InboxMessage[] {
|
|
return this.steeringMessages.splice(0)
|
|
}
|
|
|
|
/**
|
|
* Discard all pending messages (queued + steering) without delivering them —
|
|
* used by `cancel()`, which drops un-started work rather than draining it into
|
|
* a turn. Unlike `drainQueued`/`drainSteering`, the messages are thrown away.
|
|
*/
|
|
clear(): void {
|
|
this.queuedMessages.length = 0
|
|
this.steeringMessages.length = 0
|
|
}
|
|
|
|
/** Wait until a queued message arrives or `cancel` resolves. */
|
|
waitForQueued(cancel: Promise<void>): Promise<void> {
|
|
if (this.hasQueued) return Promise.resolve()
|
|
const { promise, resolve } = Promise.withResolvers<void>()
|
|
this.wakeup = resolve
|
|
void cancel.then(resolve)
|
|
return promise.finally(() => {
|
|
if (this.wakeup === resolve) this.wakeup = undefined
|
|
})
|
|
}
|
|
}
|