Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge master into codex/session-title

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-21 22:59:41 +08:00
parent 3a895ce815 0cd8424d46
commit 3176d232b8
197 changed files with 3343 additions and 1227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/core/agent-loop/README.md
View File

@@ -60,7 +60,7 @@ The driver owns one agent for its lifetime and runs inside `ctx.agents.withIniti
Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery receives the exact live error, immutable provider facts, and immutable prior failures after the failed step closes. A retry rebuilds from the durable log in a new numbered step, success clears the consecutive history, and exhaustion records the structured failure once on `turn/end`. Effective cancellation resolves its reason and emits `agent/cancel-requested` before clearing pending work or aborting the current step; notification failures are contained, queued work added by an observer is included in the same broad clear, and idle cancellation emits nothing. Undispatched model tool calls receive synthetic `tool/call` and aborted result pairs. Terminal continuation stops remain authoritative through turn close and durability flush.
Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery receives the exact live error, immutable provider facts, and immutable prior failures after the failed step closes. A retry rebuilds from the durable log in a new numbered step, success clears the consecutive history, and exhaustion records the structured failure once on `turn/end`. AgentLoop privately owns one cancellation holder whose explicit signal spans prompt policy, assembly, every step, model and tool work, recovery, continuation, and terminal stop; it retires the holder immediately before publishing `turn/end`, while the driver may remain `running` through the durability flush. An effective `cancel()` emits the typed runtime-only `user | parent` cause before clearing pending work and cooperatively aborting the holder; notification failures cannot veto cancellation, work queued by a notification observer is cleared, work queued by a later abort observer belongs to the next turn, and idle cancellation emits nothing. Durable `turn/end` remains coarse `aborted`; undispatched model tool calls receive synthetic `tool/call` and `ABORTED_BEFORE_DISPATCH` result pairs. Disposal wins terminal classification, and work that ignores the signal must settle before quiescence. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns the lifecycle and race contract. Terminal continuation stops remain authoritative through turn close and durability flush.
Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, then drains accepted batch context before the turn closes through the normal abort path.
@@ -109,7 +109,7 @@ Ordinary history growth is append-only and preserves reusable entries. A surface
#### What the model sees
If a later request replays an aborted step, each tool call that cancellation prevented from dispatching has the error result text `Error: tool call skipped because the step was aborted before execution`.
If a later request replays an aborted step, each tool call that cancellation prevented from dispatching has error code `ABORTED_BEFORE_DISPATCH` and result text `Error: tool call aborted before dispatch`.
#### Token effect

69
packages/core/agent-loop/src/agent.ts
View File

@@ -8,11 +8,12 @@
import type { Context } from 'cordis'
import { agentEvents } from '@deepseek-ai/dsh-agent'
import type { AgentOptions, AgentStatus, HookContext, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { AgentCancelCause, AgentOptions, AgentStatus, HookContext, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { deepFreeze, errorChain } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import { snapshotJsonValue, type Session, type SessionId } from '@deepseek-ai/dsh-session'
import { DISPOSED_INTERRUPT_REASON, TurnCancellation } from './cancellation.ts'
import { Inbox, type InboxMessage } from './inbox.ts'
import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
@@ -95,7 +96,7 @@ export function bindReactLoopAgentContext(agent: ReactLoopAgent, ctx: Context):
/**
* The concrete {@link Agent} implementation owned by the agent-loop plugin.
*
* Owns the inbox (queued + steering FIFOs), the per-step AbortController, and
* Owns the inbox (queued + steering FIFOs), turn cancellation, and
* the loop driver. Everything observable happens through session events and
* the agent/* event taxonomy — plugins never need this class.
*/
@@ -120,21 +121,14 @@ export class ReactLoopAgent implements Agent {
}
private _status: AgentStatus = 'idle'
private currentAbort: AbortController | undefined
/** Active turn owner from pre-running publication through durability settlement. */
private turnCancellation: TurnCancellation | undefined
/** Whether runLoop has been installed into {@link done}. */
private driverStarted = false
/** Whether registry publication began and status disposal is externally visible. */
private published = false
/**
* Turn-scoped cancel marker, set by {@link cancel} and read/cleared by the
* driver loop (via the LoopHandle) at every point a turn could start or
* continue. Armed ONLY when there is something to cancel (a running turn, an
* in-flight step, or queued/steering work), so an idle no-op cancel cannot
* leave it set to wrongly drop a later prompt.
*/
private cancelRequested = false
/** Pending cancellation reason, preserved even outside an active step signal. */
private cancelReason = 'cancelled'
/** Cause-less marker for queued work cancelled before the driver installs a turn owner. */
private preRunCancelled = false
private disposed: Promise<void>
private resolveDisposed!: () => void
/** Resolves when the driver loop has fully exited (tests/disposal). */
@@ -330,29 +324,21 @@ export class ReactLoopAgent implements Agent {
}
}
cancel(reason?: string): void {
const resolvedReason = reason ?? 'cancelled'
// Arm only for current work; an idle marker would cancel the next prompt.
if (this._status === 'running' || this.currentAbort !== undefined || this.#inbox.hasQueued || this.#inbox.hasSteering) {
this.cancelRequested = true
// Capture the resolved reason for the marker-only windows (pre-step /
// continuation). The mid-step path reads it from abort.signal.reason
// below; the marker path reads it via the LoopHandle's cancelReason().
this.cancelReason = resolvedReason
cancel(cause?: AgentCancelCause): void {
const resolvedCause = cause ?? { kind: 'user' }
const cancellation = this.turnCancellation
const preRun = cancellation === undefined && (this.#inbox.hasQueued || this.#inbox.hasSteering)
if (cancellation !== undefined || preRun) {
if (preRun) this.preRunCancelled = true
// Coordination consumers must update their own state before this call
// clears the inbox or aborts the step. Notification failures are
// clears the inbox or aborts the turn. Notification failures are
// contained by the fused dispatcher and cannot veto cancellation.
agentEvents(this.loopCtx, this).emit('agent/cancel-requested', resolvedReason)
agentEvents(this.loopCtx, this).emit('agent/cancel-requested', resolvedCause)
}
// Drop all pending queued + steering work (un-started prompts never run; the
// cancelled turn's steering is not re-enqueued). Cleared directly even when
// the loop is parked in waitForQueued — there is no turn to stop and nothing
// left for the parked loop to run, so no wake is needed.
// Clear work already present before abort observers run. A replacement
// synchronously enqueued by an observer belongs to the next turn.
this.#inbox.clear()
// Interrupt an in-flight step immediately (the running turn observes the
// abort and ends `aborted`). The marker covers the windows where no step is
// running (pre-step, continuation).
this.currentAbort?.abort(resolvedReason)
cancellation?.request(resolvedCause)
}
/**
@@ -394,14 +380,21 @@ export class ReactLoopAgent implements Agent {
inbox: this.#inbox,
maxParallelToolCalls: this.maxParallelToolCalls,
setStatus: (status) => { this.setStatus(status) },
setAbort: controller => void (this.currentAbort = controller),
installTurnCancellation: () => {
const cancellation = new TurnCancellation()
this.turnCancellation = cancellation
return cancellation
},
clearTurnCancellation: (cancellation) => {
/* v8 ignore else -- the driver clears only the exact owner returned by its latest install. */
if (this.turnCancellation === cancellation) this.turnCancellation = undefined
},
disposed: this.disposed,
isDisposed: () => this._status === 'disposed',
isCancelled: () => this.cancelRequested,
cancelReason: () => this.cancelReason,
clearCancel: () => { this.cancelRequested = false },
isPreRunCancelled: () => this.preRunCancelled,
clearPreRunCancel: () => { this.preRunCancelled = false },
withToolBatch: run => this.withToolBatch(run),
// Pre-start cancellation settles queued-work waiters before publishing idle.
// Pre-run cancellation settles queued-work waiters before publishing idle.
settleIdle: () => { this.settleIdleWaiters() },
}))
}
@@ -419,7 +412,7 @@ export class ReactLoopAgent implements Agent {
// internal state that must settle even if a listener throws below. Each
// waiter chains `done`, so it resolves only once the loop actually exits.
this.settleIdleWaiters()
this.currentAbort?.abort('disposed')
this.turnCancellation?.request(DISPOSED_INTERRUPT_REASON)
// An unpublished rollback has no public status lifecycle to announce.
// Once publication begins, disposed is part of the agent/status contract.
if (this.published) {

31
packages/core/agent-loop/src/cancellation.ts Normal file
View File

@@ -0,0 +1,31 @@
/** Turn-scoped cancellation ownership for the concrete AgentLoop driver. @module dsh-agent-loop/cancellation */
import type { AgentCancelCause } from '@deepseek-ai/dsh-agent'
/** Stable runtime-only reason used when lifecycle teardown interrupts a turn. */
export const DISPOSED_INTERRUPT_REASON = Object.freeze({ kind: 'disposed' } as const)
/**
* Owns the single controller shared by every asynchronous boundary of one turn.
* The first request wins because a later caller must not rewrite the cause
* observed by earlier listeners.
*/
export class TurnCancellation {
readonly #controller = new AbortController()
/** The explicit signal passed through this turn's execution boundaries. */
get signal(): AbortSignal {
return this.#controller.signal
}
/**
* Abort the turn once.
* @param reason - a typed caller cause or lifecycle disposal marker.
* @returns whether this request established the signal reason.
*/
request(reason: AgentCancelCause | typeof DISPOSED_INTERRUPT_REASON): boolean {
if (this.signal.aborted) return false
this.#controller.abort(Object.freeze({ kind: reason.kind }))
return true
}
}

2
packages/core/agent-loop/src/inbox.ts
View File

@@ -29,7 +29,7 @@ export class Inbox {
return this.queuedMessages.length > 0
}
/** True while steering messages are pending — read by `cancel()`'s arm gate and the loop's stop-override check. */
/** True while steering messages are pending — read by cancellation and the loop's stop-override check. */
get hasSteering(): boolean {
return this.steeringMessages.length > 0
}

242
packages/core/agent-loop/src/loop.ts
View File

@@ -9,7 +9,7 @@ import type { Context } from 'cordis'
import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message } from '@deepseek-ai/dsh-llm'
import { isDeepStrictEqual } from 'node:util'
import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf, markAgentLoopRequest } from '@deepseek-ai/dsh-llm'
import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
import { agentEvents, agentInterruptReasonOf, assembleContextFor } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
import { canonicalHeader } from '@deepseek-ai/dsh-session'
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
@@ -20,6 +20,7 @@ import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
import { executeToolCalls } from './tool-calls.ts'
import type { Inbox } from './inbox.ts'
import type { TurnCancellation } from './cancellation.ts'
/** Normalize thrown values while preserving an existing error code. */
function toError(error: unknown): RequestError {
@@ -88,6 +89,32 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
}
}
/** Internal control-flow sentinel; durable classification comes only from the turn signal. */
const TURN_INTERRUPTED = new Error('turn interrupted')
/** Stop at an explicit cooperative boundary without stringifying the runtime reason. */
function interruptionCheckpoint(signal: AbortSignal): void {
if (signal.aborted) throw TURN_INTERRUPTED
}
/** Classify a supported turn interruption, with lifecycle disposal taking precedence. */
function interruptionTurnEndReason(handle: LoopHandle, signal: AbortSignal): TurnEndReason | undefined {
if (handle.isDisposed()) return { kind: 'disposed' }
const reason = agentInterruptReasonOf(signal)
if (reason === undefined) return undefined
switch (reason.kind) {
case 'user':
case 'parent':
return { kind: 'aborted' }
/* v8 ignore next 2 -- the private holder requests disposed only after lifecycle state flips, which returns above. */
case 'disposed':
return { kind: 'disposed' }
/* v8 ignore next 2 -- AgentInterruptReason is closed and the public helper filters unsupported reasons. */
default:
return assertNever(reason, 'AgentInterruptReason')
}
}
/** Mutable agent controls supplied to the loop driver. */
export interface LoopHandle {
/** Native-private agent inbox handed to the driver only at internal startup. */
@@ -95,16 +122,17 @@ export interface LoopHandle {
/** Maximum parallel-safe calls allowed in one step. */
readonly maxParallelToolCalls: number
setStatus(status: 'idle' | 'running'): void
setAbort(controller: AbortController | undefined): void
/** Install a fresh active-turn owner before the running notification. */
installTurnCancellation(): TurnCancellation
/** Clear only the exact owner whose turn reached its terminal event boundary. */
clearTurnCancellation(cancellation: TurnCancellation): void
/** Resolves when the agent is disposed — unblocks the idle wait. */
disposed: Promise<void>
isDisposed(): boolean
/** Whether cancellation is pending for the current loop iteration. */
isCancelled(): boolean
/** Resolved pending-cancellation reason; meaningful only while {@link isCancelled} is true. */
cancelReason(): string
/** Clear the cancel marker (called once per iteration after the turn returns). */
clearCancel(): void
/** Whether queued work was cancelled before an active turn owner existed. */
isPreRunCancelled(): boolean
/** Clear the cause-less pre-run marker without affecting replacement work. */
clearPreRunCancel(): void
/** Settle idle waiters before pre-running cancellation publishes idle. */
settleIdle(): void
/** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */
@@ -119,7 +147,7 @@ export interface LoopHandle {
* @param ctx - the plugin context the loop reaches its initiating Agent,
* events (agent/…, session/flush), and services (systemPrompt, llm, tools)
* through.
* @param handle - the bridge to the agent's mutable state: status/abort setters plus the disposal and cancel-marker reads.
* @param handle - the bridge to status, turn cancellation ownership, disposal, and pre-run cancellation state.
* @throws when no initiating Agent is active.
*/
export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
@@ -134,8 +162,8 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
while (!handle.isDisposed()) {
// An idle listener can enqueue and cancel replacement work before the next
// wait is installed. Consume that empty marker before parking the driver.
if (handle.isCancelled()) {
handle.clearCancel()
if (handle.isPreRunCancelled()) {
handle.clearPreRunCancel()
if (!handle.inbox.hasQueued) {
handle.settleIdle()
handle.setStatus('idle')
@@ -148,8 +176,8 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
// Cancellation between wake and `running` skips only the cancelled work;
// a replacement prompt still runs before the eventual idle transition.
if (handle.isCancelled()) {
handle.clearCancel()
if (handle.isPreRunCancelled()) {
handle.clearPreRunCancel()
if (!handle.inbox.hasQueued) {
// Settle before publishing idle: the already-idle path has no status
// transition, while an idle listener can register waiters for new work.
@@ -159,24 +187,29 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
}
}
let cancellation = handle.installTurnCancellation()
handle.setStatus('running')
if (handle.isDisposed()) break
if (handle.isDisposed()) {
handle.clearTurnCancellation(cancellation)
break
}
// A synchronous `running` listener can cancel before `runTurn`; balance the
// status only when no replacement prompt was queued by that listener.
if (handle.isCancelled()) {
handle.clearCancel()
if (cancellation.signal.aborted) {
handle.clearTurnCancellation(cancellation)
if (!handle.inbox.hasQueued) {
handle.setStatus('idle')
continue
}
cancellation = handle.installTurnCancellation()
}
// Idle injection can add a turn, so derive the next number from the log.
const turn = lastTurnNumber(session) + 1
let terminalStopped = false
try {
terminalStopped = await runTurn(ctx, events, handle, turn, transmission)
terminalStopped = await runTurn(ctx, events, handle, turn, transmission, cancellation)
} catch (error: unknown) {
// Pre-turn failure has no durable boundary to close; report it without appending outside a turn.
const err = toError(error)
@@ -184,11 +217,10 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
try {
events.emit('agent/error', turn, 0, err)
} catch { /* contained: a throwing agent/error listener must not kill the driver */ }
} finally {
handle.clearTurnCancellation(cancellation)
}
// Reset per iteration, including when a prompt arrives during the flush window.
handle.clearCancel()
// Late steering becomes queued input unless terminal policy stopped the turn.
for (const message of handle.inbox.drainSteering()) {
if (!terminalStopped) handle.inbox.enqueue(message)
@@ -200,9 +232,11 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
async function runTurn(
ctx: Context, events: AgentEventDispatch, handle: LoopHandle, turn: number, transmission: TransmissionLog,
cancellation: TurnCancellation,
): Promise<boolean> {
const agent = ctx.agents.requireInitiator()
const { session } = agent
const { signal } = cancellation
const drainSteering = (): boolean => {
const messages = handle.inbox.drainSteering()
for (const message of messages) {
@@ -246,8 +280,11 @@ async function runTurn(
}
}
// Pre-commit validation failure escapes rather than masquerading as a committed boundary.
// Retire cancellation authority before publishing the terminal event. The
// following durability flush is quiescent turn work, but no longer part of
// the cancellable turn lifetime.
const closeTurn = (): void => {
handle.clearTurnCancellation(cancellation)
session.append('turn/end', { turn, reason })
}
@@ -256,15 +293,17 @@ async function runTurn(
// matter what throws below; the catch + closeTurn guarantee it. A pre-commit
// veto leaves no turn/start in the log and therefore owes no turn/end.
session.append('turn/start', { turn, trigger })
interruptionCheckpoint(signal)
// The claimed message runs the `agent/prompt-submit` waterfall before it
// becomes a `user/message` — a hook can rewrite the prompt or block it.
// Recorded INSIDE the turn (after turn/start) so every event is turn-enclosed;
// turn/end is now owed, so a throwing prompt-submit listener (the waterfall
// throws) is caught below and the turn still closes.
const promptDecision = await events.waterfall(
'agent/prompt-submit', message.content, message.source,
'agent/prompt-submit', message.content, message.source, signal,
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
)
interruptionCheckpoint(signal)
if (promptDecision.kind === 'block') {
session.append('prompt/blocked', { content: message.content, source: message.source, reason: promptDecision.reason })
reason = { kind: 'rejected', reason: promptDecision.reason }
@@ -292,53 +331,28 @@ async function runTurn(
// the request.
drainSteering()
// The step's AbortController exists BEFORE any async pre-step work so a
// dispose() or cancel() — in a synchronous turn-start listener or an
// async listener whose effect fires before we block — always has an armed
// abort to cancel against. isDisposed below covers disposal, which does
// NOT set the cancel marker. Cleared on every exit path below.
const abort = new AbortController()
handle.setAbort(abort)
// Assemble once before pre-step so listener work and the request share one prompt value.
const assembly = await ctx.systemPrompt.assemble(assembleContextFor(agent))
const assembly = await ctx.systemPrompt.assemble(assembleContextFor(agent, signal))
interruptionCheckpoint(signal)
const fullSystemPrompt = renderPrompt(assembly)
// Cancellation or disposal during assembly ends the turn before any step opens.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
break
}
// Compose the request-only prefix once per loop instance before the first
// request boundary. It precedes all derived history and is recorded only
// in the request header, not as session history.
if (transmission.sessionPrefix === undefined) {
const emptyPrefix: Message[] = deepFreeze([])
const composed = await events.waterfall(
'agent/session-prefix', emptyPrefix, abort.signal,
'agent/session-prefix', emptyPrefix, signal,
() => Promise.resolve(emptyPrefix),
)
// Never cache an interrupted composition; the next turn recomposes it.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
break
}
interruptionCheckpoint(signal)
transmission.sessionPrefix = deepFreeze(structuredClone(composed))
}
// Await surface mutations outside the step before snapshotting history.
await events.serial('agent/pre-step', turn, step, abort.signal)
// Interruption landing during the pre-step seam: do not open an empty step.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
break
}
await events.serial('agent/pre-step', turn, step, signal)
interruptionCheckpoint(signal)
// Snapshot the exact log prefix before step/start: the reconstruction
// boundary. Appends after this synchronous snapshot join the next request.
@@ -350,16 +364,8 @@ async function runTurn(
// are contained inside Session.append().
stepOpen = true
// Cancel landing in the step-start window: a synchronous `session/event`
// step/start listener can cancel after the step is already open. Check
// AFTER the step/start append and before `runStep`: drop the step, end the
// turn accordingly. closeStep balances the already-appended step/start.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
closeStep()
break
}
// A synchronous step/start observer can cancel after the step opened.
interruptionCheckpoint(signal)
let stepOutcome:
| { hadToolCalls: boolean; finish: FinishReason }
@@ -367,7 +373,7 @@ async function runTurn(
| { error: RequestError }
try {
stepOutcome = await runStep(
ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, signal)
} catch (error: unknown) {
if (error instanceof TerminalModelRequestFailure) {
stepOutcome = { requestError: error.requestError, failure: error.failure }
@@ -380,11 +386,9 @@ async function runTurn(
// Recovery observes a balanced failed step and the original provider
// error while the failed step's signal remains the active owner.
closeStep()
if (handle.isDisposed() || abort.signal.aborted) {
handle.setAbort(undefined)
reason = handle.isDisposed()
? { kind: 'disposed' }
: { kind: 'aborted', reason: String(abort.signal.reason) }
const interrupted = interruptionTurnEndReason(handle, signal)
if (interrupted !== undefined) {
reason = interrupted
break
}
@@ -393,7 +397,7 @@ async function runTurn(
try {
recoveryDecision = await events.waterfall(
'agent/request-error', turn, step, stepOutcome.requestError,
stepOutcome.failure, requestFailureHistory, abort.signal,
stepOutcome.failure, requestFailureHistory, signal,
() => Promise.resolve(defaultDecision),
)
} catch (recoveryError: unknown) {
@@ -401,15 +405,11 @@ async function runTurn(
`agent "${agent.id}": request recovery failed at turn ${turn}, step ${step}: ${errorChain(recoveryError)}`,
)
}
handle.setAbort(undefined)
// Cancellation and disposal always win over either a recovery decision
// or a recovery-listener failure.
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (handle.isDisposed() || abort.signal.aborted) {
reason = handle.isDisposed()
? { kind: 'disposed' }
: { kind: 'aborted', reason: String(abort.signal.reason) }
const recoveryInterrupted = interruptionTurnEndReason(handle, signal)
if (recoveryInterrupted !== undefined) {
reason = recoveryInterrupted
break
}
switch (recoveryDecision.action) {
@@ -431,17 +431,10 @@ async function runTurn(
// runLoop re-enqueues it as a queued message, so an abort-then-steer
// starts a fresh turn instead of being silently consumed.
closeStep()
handle.setAbort(undefined)
const { error } = stepOutcome
/* v8 ignore next -- narrow race: disposal while non-request step work throws. */
if (handle.isDisposed()) {
reason = { kind: 'disposed' }
} else if (abort.signal.aborted) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
} else {
failTurn(error)
}
const interrupted = interruptionTurnEndReason(handle, signal)
if (interrupted === undefined) failTurn(error)
else reason = interrupted
break
}
@@ -455,48 +448,40 @@ async function runTurn(
const steered = drainSteering()
try {
await events.serial('agent/post-step', turn, step, abort.signal)
await events.serial('agent/post-step', turn, step, signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
}
if ('error' in stepOutcome) {
closeStep()
handle.setAbort(undefined)
/* v8 ignore next -- narrow race: disposal while a post-step listener throws. */
if (handle.isDisposed()) {
reason = { kind: 'disposed' }
} else if (abort.signal.aborted) {
/* v8 ignore next -- signal.reason always set by cancellation or disposal. */
reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
} else {
failTurn(stepOutcome.error)
}
const interrupted = interruptionTurnEndReason(handle, signal)
if (interrupted === undefined) failTurn(stepOutcome.error)
else reason = interrupted
break
}
if (handle.isDisposed() || abort.signal.aborted) {
reason = handle.isDisposed()
? { kind: 'disposed' }
: { kind: 'aborted', reason: String(abort.signal.reason) }
const postStepInterrupted = interruptionTurnEndReason(handle, signal)
if (postStepInterrupted !== undefined) {
reason = postStepInterrupted
closeStep()
handle.setAbort(undefined)
break
}
closeStep()
handle.setAbort(undefined)
const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
let decision: ContinuationDecision
try {
decision = await events.waterfall(
'agent/turn-continuation', turn, defaultDecision,
'agent/turn-continuation', turn, defaultDecision, signal,
() => Promise.resolve(defaultDecision),
)
interruptionCheckpoint(signal)
} catch (error: unknown) {
// A broken continuation plugin ends the turn, not the loop.
failTurn(toError(error))
const interrupted = interruptionTurnEndReason(handle, signal)
if (interrupted === undefined) failTurn(toError(error))
else reason = interrupted
break
}
@@ -512,12 +497,15 @@ async function runTurn(
// Terminal policy is monotonic and runs after ordinary continuation folding.
let terminalStop = false
try {
const stop = await events.serial('agent/turn-stop', turn)
const stop = await events.serial('agent/turn-stop', turn, signal)
interruptionCheckpoint(signal)
terminalStop = stop !== undefined
} catch (error: unknown) {
// A broken terminal policy is an ordinary continuation failure: fail
// this turn closed while leaving the driver alive for later turns.
failTurn(toError(error))
const interrupted = interruptionTurnEndReason(handle, signal)
if (interrupted === undefined) failTurn(toError(error))
else reason = interrupted
break
}
if (terminalStop) {
@@ -527,17 +515,7 @@ async function runTurn(
shouldContinue = false
}
// The marker catches cancellation after the step controller was cleared.
if (handle.isCancelled()) {
reason = { kind: 'aborted', reason: handle.cancelReason() }
break
}
if (!shouldContinue || handle.isDisposed()) {
/* v8 ignore next -- disposal during continuation-decision window is a narrow race; error-path disposal is covered elsewhere */
if (handle.isDisposed()) reason = { kind: 'disposed' }
break
}
if (!shouldContinue) break
}
// Normal / inline-error loop exit: close the turn.
@@ -547,12 +525,9 @@ async function runTurn(
const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
if (!turnStartLogged) throw error
closeStep()
// Preserve an established disposal reason; otherwise report the failure.
if (handle.isDisposed() && !errorReported) { // eslint-disable-line @typescript-eslint/no-unnecessary-condition
reason = { kind: 'disposed' }
} else {
failTurn(toError(error))
}
const interrupted = interruptionTurnEndReason(handle, signal)
if (interrupted === undefined) failTurn(toError(error))
else reason = interrupted
closeTurn()
}
@@ -601,7 +576,10 @@ async function runStep(
: { provider: options.provider ?? '', model: options.model ?? '' }))
// Listener replacements are recorded in the request header before dispatch.
const config = await events.waterfall('agent/request', turn, step, seedConfig, () => Promise.resolve(seedConfig))
const config = await events.waterfall(
'agent/request', turn, step, seedConfig, signal, () => Promise.resolve(seedConfig),
)
interruptionCheckpoint(signal)
if (!config.provider || !config.model) {
throw new Error(`agent "${agent.id}" has no provider/model: set AgentOptions.provider and AgentOptions.model or supply both via the agent/request waterfall`)
}
@@ -638,8 +616,7 @@ async function runStep(
const stream = ctx.llm.stream(request)
try {
for await (const chunk of stream) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
interruptionCheckpoint(signal)
const chunkEvent = session.append('assistant/chunk', { turn, step, chunk })
chunkSeqs.push(chunkEvent.seq)
assembler.push(chunk)
@@ -649,6 +626,7 @@ async function runStep(
if (failure !== undefined && error instanceof Error) throw new TerminalModelRequestFailure(error, failure)
throw error
}
interruptionCheckpoint(signal)
// Normalize failure finish chunks into the same path as thrown stream errors.
const stepError = finishError(assembler.finish)
@@ -679,9 +657,11 @@ async function runStep(
// A rejected result still records the successful provider call without retaining rejected output.
const processStepResult = async (assembledContent: ContentBlock[], message: Message): Promise<Message> => {
try {
return await events.waterfall(
'agent/step-result', turn, step, message, () => Promise.resolve(message),
const processed = await events.waterfall(
'agent/step-result', turn, step, message, signal, () => Promise.resolve(message),
)
interruptionCheckpoint(signal)
return processed
} catch (error: unknown) {
recordAssistantMessage(assembledContent, { ...message, content: [] }, false)
throw error

6
packages/core/agent-loop/src/tool-calls.ts
View File

@@ -13,7 +13,7 @@ import type { Context } from 'cordis'
import { assertNever, type ToolCallBlock } from '@deepseek-ai/dsh-llm'
import type { HookContext } from '@deepseek-ai/dsh-agent'
import type { Session } from '@deepseek-ai/dsh-session'
import { TOOL_REGISTRY_SCHEDULER, type ToolExecutionInput, type ToolExecutionMode, type ToolExecutionResult, type ToolRunContext } from '@deepseek-ai/dsh-tools'
import { TOOL_ABORTED_BEFORE_DISPATCH, TOOL_REGISTRY_SCHEDULER, type ToolExecutionInput, type ToolExecutionMode, type ToolExecutionResult, type ToolRunContext } from '@deepseek-ai/dsh-tools'
/** One tool call after argument parsing, ready to schedule. */
interface PlannedCall {
@@ -217,9 +217,9 @@ async function runGroup(
function appendSkippedToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): void {
const callSeq = appendToolCall(session, turn, step, block)
appendToolResult(session, turn, step, block, {
content: [{ type: 'text', text: 'Error: tool call skipped because the step was aborted before execution' }],
content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],
isError: true,
error: { name: 'AbortError', code: 'ABORTED' },
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
}, callSeq)
}

77
packages/core/agent-loop/tests/agent-initiator.spec.ts
View File

@@ -9,6 +9,8 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
const testToolSignal = new AbortController().signal
interface Harness {
ctx: Context
agentsFiber: Fiber
@@ -142,6 +144,80 @@ describe('AgentLoop initiator scope', () => {
await ctx.fiber.dispose()
})
it('keeps initiator identity minimal while one explicit signal spans each turn seam', async () => {
const adapter = new MockAdapter([
toolCallResponse('observe-call', 'observe', {}),
textResponse('first done'),
textResponse('second done'),
])
const { ctx } = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('signal-owner'), { provider: 'mock', model: 'mock' })
let signals: AbortSignal[] = []
const capture = (signal: AbortSignal | undefined): void => {
if (signal === undefined) throw new Error('turn seam omitted its explicit signal')
expect(ctx.agents.requireInitiator()).toBe(agent)
signals.push(signal)
}
ctx.on('system-prompt/assemble', async (_assembly, context, next) => {
if (context.agent === agent) capture(context.signal)
return next()
})
ctx.on('agent/prompt-submit', async (subject, _content, _source, signal, next) => {
if (subject === agent) capture(signal)
return next()
})
ctx.on('agent/session-prefix', async (subject, _prefix, signal, next) => {
if (subject === agent) capture(signal)
return next()
})
ctx.on('agent/pre-step', (subject, _turn, _step, signal) => {
if (subject === agent) capture(signal)
})
ctx.on('agent/request', async (subject, _turn, _step, _config, signal, next) => {
if (subject === agent) capture(signal)
return next()
})
ctx.on('agent/step-result', async (subject, _turn, _step, _message, signal, next) => {
if (subject === agent) capture(signal)
return next()
})
ctx.on('agent/turn-continuation', async (subject, _turn, _decision, signal, next) => {
if (subject === agent) capture(signal)
return next()
})
ctx.on('agent/turn-stop', (subject, _turn, signal) => {
if (subject === agent) capture(signal)
})
ctx.tools.register(defineTool({
name: 'observe',
description: 'observe explicit turn state',
parameters: {},
execute: async (_args, exec) => {
capture(exec.signal)
return [{ type: 'text', text: 'observed' }]
},
}))
const firstIdle = waitForIdle(ctx, agent)
send(agent, 'first')
await firstIdle
const firstSignal = signals[0]
expect(firstSignal).toBeDefined()
expect(new Set([...signals, ...adapter.requests.slice(0, 2).map(request => request.signal!)])).toEqual(new Set([firstSignal]))
signals = []
const secondIdle = waitForIdle(ctx, agent)
send(agent, 'second')
await secondIdle
const secondSignal = signals[0]
expect(secondSignal).toBeDefined()
expect(new Set([...signals, adapter.requests[2]!.signal!])).toEqual(new Set([secondSignal]))
expect(secondSignal).not.toBe(firstSignal)
expect(ctx.agents.currentInitiator()).toBeUndefined()
await ctx.fiber.dispose()
})
it('keeps child setup under the parent boundary and restores the parent while the child driver remains active', async () => {
const adapter = new MockAdapter([
toolCallResponse('spawn', 'spawn-child', {}),
@@ -239,6 +315,7 @@ describe('AgentLoop initiator scope', () => {
}))
const direct = await ctx.tools.execute({
signal: testToolSignal,
callId: CallId('direct'),
name: 'agentless-probe',
arguments: {},

2
packages/core/agent-loop/tests/agent.spec.ts
View File

@@ -345,7 +345,7 @@ describe('Agent', () => {
expect(settled).toBe(false)
await waitForStatus(ctx, agent, 'running')
agent.cancel('done')
agent.cancel({ kind: 'user' })
await idle
expect(settled).toBe(true)
expect(agent.status).toBe('idle')

316
packages/core/agent-loop/tests/cancel.spec.ts
View File

@@ -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, { defineTool } from '@deepseek-ai/dsh-tools'
import ToolRegistry, { defineTool, 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(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()
})
})

77
packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -3,7 +3,7 @@ import { Context } from 'cordis'
import LlmService, { CallId, ContentBlock, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineTool, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import ToolRegistry, { defineTool, TOOL_ABORTED, TOOL_ABORTED_BEFORE_DISPATCH, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from '@deepseek-ai/dsh-agent-loop'
import { prepareReactLoopAgent } from '../src/agent.ts'
@@ -73,7 +73,7 @@ describe('session log records what agent/step-result actually produced', () => {
// Plugin rewrites the message: replaces the text AND adds a tool call.
let rewritten = false
ctx.on('agent/step-result', async (_agent, _turn, _step, _message, next) => {
ctx.on('agent/step-result', async (_agent, _turn, _step, _message, _signal, next) => {
if (rewritten) return next()
rewritten = true
return {
@@ -214,7 +214,7 @@ describe('successful provider completion survives agent/step-result failure', ()
})
describe('abort during tool execution ends the turn', () => {
it('balances an aborted tool batch through context, steering, and post-step before closing', async () => {
it('balances a cancelled tool batch through context and post-step before closing', async () => {
const adapter = new MockAdapter([
// model asks for two tool calls in one step
[
@@ -239,8 +239,7 @@ describe('abort during tool execution ends the turn', () => {
[{ type: 'text', text: 'steering before abort' }],
{ source: { kind: 'plugin', plugin: 'abort-test' } },
)
// Exercise bare step abort without `cancel()` clearing queued work.
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
agent.cancel({ kind: 'user' })
return [{ type: 'text', text: 'done' }]
},
}))
@@ -269,7 +268,10 @@ describe('abort during tool execution ends the turn', () => {
case 'assistant/message': order.push('assistant/message'); break
case 'tool/call': order.push(`tool/call:${event.data.callId}`); break
case 'tool/result': {
const outcome = event.data.error?.code === 'ABORTED' ? 'synthetic-aborted' : 'real'
const outcome = event.data.error?.code === TOOL_ABORTED
|| event.data.error?.code === TOOL_ABORTED_BEFORE_DISPATCH
? 'aborted'
: 'completed'
order.push(`tool/result:${event.data.callId}:${outcome}`)
break
}
@@ -300,25 +302,29 @@ describe('abort during tool execution ends the turn', () => {
expect(order).toEqual([
'assistant/message',
'tool/call:c1',
'tool/result:c1:real',
'tool/result:c1:aborted',
'tool/call:c2',
'tool/result:c2:synthetic-aborted',
'tool/result:c2:aborted',
'context/message',
'steering/message',
'agent/post-step',
'step/end',
'turn/end:aborted',
])
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
expect(reasons).toEqual([{ kind: 'aborted' }])
const calls = agent.session.events.filter(event => event.type === 'tool/call')
const results = agent.session.events.filter(event => event.type === 'tool/result')
expect(calls.map(event => event.data.callId)).toEqual([CallId('c1'), CallId('c2')])
expect(results).toHaveLength(2)
expect(results[0]!.data).toMatchObject({ callId: CallId('c1'), isError: false })
expect(results[0]!.data).toMatchObject({
callId: CallId('c1'),
content: [{ type: 'text', text: 'Error: tool call aborted' }],
isError: true,
error: { name: 'AbortError', code: TOOL_ABORTED },
})
expect(results[1]!.data).toMatchObject({
callId: CallId('c2'),
isError: true,
error: { name: 'AbortError', code: 'ABORTED' },
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
})
})
@@ -332,7 +338,7 @@ describe('abort during tool execution ends the turn', () => {
parameters: {},
async execute() {
agent.inject([{ type: 'text', text: 'accepted before abort' }], { source: { kind: 'plugin', plugin: 'test' } })
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
agent.cancel({ kind: 'user' })
return [{ type: 'text', text: 'done' }]
},
}))
@@ -385,7 +391,7 @@ describe('abort during tool execution ends the turn', () => {
description: '',
parameters: {},
async execute() {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
agent.cancel({ kind: 'user' })
return [{ type: 'text', text: 'aborted' }]
},
}))
@@ -478,7 +484,7 @@ describe('abort during tool execution ends the turn', () => {
description: '',
parameters: {},
async execute() {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
agent.cancel({ kind: 'user' })
return [{ type: 'text', text: 'done' }]
},
}))
@@ -517,7 +523,7 @@ describe('steering from late extension points is never stranded', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let steeredOnce = false
ctx.on('agent/turn-continuation', async (_agent, _turn, _decision, next) => {
ctx.on('agent/turn-continuation', async (_agent, _turn, _decision, _signal, next) => {
if (!steeredOnce) {
steeredOnce = true
agent.steer([{ type: 'text', text: 'one more thing' }])
@@ -591,26 +597,6 @@ describe('steering from late extension points is never stranded', () => {
expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('too late for this turn')
})
it('steering queued during an aborted step is re-delivered, not silently consumed', async () => {
const adapter = new MockAdapter(['hang', textResponse('recovered')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
agent.steer([{ type: 'text', text: 'redirect' }])
// Abort ONLY the in-flight step, via its AbortController directly — NOT
// cancel(), which clears the inbox and would drop the queued steering this
// test proves survives a step abort. There is no public step-only abort
// verb (cancel() is the only public stop primitive), so reach the private
// controller the loop registered.
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
await waitForIdle(ctx, agent)
// a new turn ran with the steering content delivered as a message
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('redirect')
})
})
describe('plugin exceptions are contained', () => {
@@ -767,7 +753,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), {}) // no model — router plugin decides
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
ctx.on('agent/request', async (_agent, _turn, _step, config, _signal) => {
return { ...config, provider: 'mock', model: 'mock' }
})
@@ -1481,7 +1467,7 @@ describe('surface: assistant/message records exact empty provenance when no chun
await mountInvariants(ctx)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/step-result', async (_agent, _turn, _step, _message, _next) => ({
ctx.on('agent/step-result', async (_agent, _turn, _step, _message, _signal) => ({
role: 'assistant' as const,
content: [{ type: 'text' as const, text: 'injected' }],
}))
@@ -1584,7 +1570,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
send(agent, 'go')
await new Promise(r => setTimeout(r, 50))
agent.cancel('user cancelled during assembly')
agent.cancel({ kind: 'user' })
releaseAssemble()
await waitForIdle(ctx, agent)
@@ -1596,15 +1582,12 @@ describe('disposal and cancellation during pre-step assembly', () => {
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({
kind: 'aborted',
reason: 'user cancelled during assembly',
})
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(e.some(x => x.type === 'assistant/message')).toBe(false)
expect(adapter.requests).toHaveLength(0)
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user cancelled during assembly' }])
expect(reasons).toEqual([{ kind: 'aborted' }])
})
it('disposal during agent/pre-step seam ends the turn disposed', { timeout: 15000 }, async () => {
@@ -1689,7 +1672,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
send(agent, 'go')
await new Promise(r => setTimeout(r, 30))
agent.cancel('user cancelled')
agent.cancel({ kind: 'user' })
releasePreStep()
await waitForIdle(ctx, agent)
@@ -1700,10 +1683,10 @@ describe('disposal and cancellation during pre-step assembly', () => {
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: 'user cancelled' })
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted' })
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user cancelled' }])
expect(reasons).toEqual([{ kind: 'aborted' }])
})
it('disposal during assembly does not leak an LLM call or append assistant/chunk', { timeout: 15000 }, async () => {

4
packages/core/agent-loop/tests/coverage-edges.spec.ts
View File

@@ -170,7 +170,7 @@ describe('toError normalization', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let threwOnce = false
ctx.on('agent/request', async (_agent, _turn, _step, _options, _next) => {
ctx.on('agent/request', async (_agent, _turn, _step, _options, _signal, _next) => {
if (!threwOnce) {
threwOnce = true
throw { code: 500 } // non-Error throw, goes through runStep catch
@@ -200,7 +200,7 @@ describe('coded error data emission', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let threwOnce = false
ctx.on('agent/request', async (_agent, _turn, _step, _options, next) => {
ctx.on('agent/request', async (_agent, _turn, _step, _options, _signal, next) => {
if (!threwOnce) {
threwOnce = true
throw new LlmError('server overloaded', 'RATE_LIMIT')

8
packages/core/agent-loop/tests/interception.spec.ts
View File

@@ -56,7 +56,7 @@ describe('agent/prompt-submit', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const seen: string[] = []
ctx.on('agent/prompt-submit', async (_agent, content, _source, next) => {
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next) => {
seen.push(content.map(b => (b.type === 'text' ? b.text : '')).join(''))
return next()
})
@@ -182,7 +182,7 @@ describe('agent/prompt-submit', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/prompt-submit', async (_agent, content, _source, next): Promise<PromptDecision> => {
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next): Promise<PromptDecision> => {
const text = content.map(b => (b.type === 'text' ? b.text : '')).join('')
return text === 'secret' ? { kind: 'block', reason: 'policy: no secrets' } : next()
})
@@ -497,7 +497,7 @@ describe('agent/turn-continuation (ContinuationDecision)', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let forced = false
ctx.on('agent/turn-continuation', async (_agent, _turn, _default, next): Promise<ContinuationDecision> => {
ctx.on('agent/turn-continuation', async (_agent, _turn, _default, _signal, next): Promise<ContinuationDecision> => {
if (!forced) {
forced = true
return { action: 'continue', reason: { content: [{ type: 'text', text: 'keep going on the goal' }], source: { kind: 'plugin', plugin: 'goal' } } }
@@ -662,7 +662,7 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
)
})
// 2. PromptSubmit: block a forbidden prompt, annotate the rest.
ctx.on('agent/prompt-submit', async (_agent, content, _source, next): Promise<PromptDecision> => {
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next): Promise<PromptDecision> => {
const text = content.map(b => (b.type === 'text' ? b.text : '')).join('')
if (text.includes('rm -rf')) return { kind: 'block', reason: 'destructive prompt blocked' }
return next()

14
packages/core/agent-loop/tests/loop.spec.ts
View File

@@ -231,7 +231,7 @@ describe('agent loop', () => {
assembly.variables['model'] = 'mock'
return next()
})
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
ctx.on('agent/request', async (_agent, _turn, _step, config, _signal, _next) => {
return { ...config, provider: 'mock', model: 'mock' }
})
const agent = ctx.agentLoop.create(SessionId('a-late-model'), {})
@@ -527,7 +527,7 @@ describe('agent loop', () => {
let steps = 0
ctx.on('session/event', (_session, event) => { if (event.type === 'step/end') steps++ })
ctx.on('agent/turn-continuation', async (_agent, _turn, _defaultDecision, next) => {
ctx.on('agent/turn-continuation', async (_agent, _turn, _defaultDecision, _signal, next) => {
if (steps < 3) return { action: 'continue' as const }
return next()
})
@@ -566,7 +566,7 @@ describe('agent loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => {
ctx.on('agent/request', async (_agent, _turn, _step, config, _signal, _next) => {
// The seed is frozen — config is not a mutable per-call knob; a switch
// is proposed by returning a replacement, and the loop logs it.
expect(Object.isFrozen(config)).toBe(true)
@@ -692,10 +692,10 @@ describe('agent loop', () => {
// wait until the stream is hanging, then cancel
await new Promise(r => setTimeout(r, 30))
expect(agent.status).toBe('running')
agent.cancel('user interrupt')
agent.cancel({ kind: 'user' })
await waitForIdle(ctx, agent)
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
expect(reasons).toEqual([{ kind: 'aborted' }])
})
it('surfaces max-tokens as the turn-end reason when the last step is cut off', async () => {
@@ -732,7 +732,7 @@ describe('agent loop', () => {
ctx.on('session/event', (_session, event) => { if (event.type === 'step/end') steps++ })
// Force exactly one continuation (step 1 → step 2), then defer to default
// (step 2 is a plain stop with no tool calls → stops).
ctx.on('agent/turn-continuation', async (_agent, _turn, _defaultDecision, next) => {
ctx.on('agent/turn-continuation', async (_agent, _turn, _defaultDecision, _signal, next) => {
if (steps < 2) return { action: 'continue' as const }
return next()
})
@@ -884,7 +884,7 @@ describe('agent loop', () => {
]])
const ctx = await harness(adapter)
let stepResults = 0
ctx.on('agent/step-result', async (_agent, _turn, _step, message, next) => {
ctx.on('agent/step-result', async (_agent, _turn, _step, message, _signal, next) => {
stepResults += 1
expect(message.content).toEqual([{ type: 'text', text: 'partial text' }])
return next()

6
packages/core/agent-loop/tests/request-reconstruction.spec.ts
View File

@@ -167,7 +167,7 @@ describe('request stability across the loop', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let injected = false
ctx.on('agent/request', async (_agent, _turn, _step, _config, next) => {
ctx.on('agent/request', async (_agent, _turn, _step, _config, _signal, next) => {
if (!injected) {
injected = true
agent.inject([{ type: 'text', text: '[late context]' }], { source: { kind: 'plugin', plugin: 'test' } })
@@ -243,7 +243,7 @@ describe('request stability across the loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request', async (_agent, _turn, _step, _config, next) => {
ctx.on('agent/request', async (_agent, _turn, _step, _config, _signal, next) => {
const config = await next()
// next() resolves the SAME frozen seed — in-place shaping after
// delegation is unrepresentable, so a "mutate what next() returned"
@@ -280,7 +280,7 @@ describe('request stability across the loop', () => {
send(agent, 'go')
await waitForIdle(ctx, agent)
ctx.systemPrompt.section({ name: 'extra', order: 2, text: 'now with guidance' })
ctx.on('agent/request', async (_agent, _turn, _step, config, _next) => ({ ...config, temperature: 0.5, maxTokens: 99, stop: ['<END>'] }))
ctx.on('agent/request', async (_agent, _turn, _step, config, _signal, _next) => ({ ...config, temperature: 0.5, maxTokens: 99, stop: ['<END>'] }))
send(agent, 'again')
await waitForIdle(ctx, agent)

8
packages/core/agent-loop/tests/request-recovery.spec.ts
View File

@@ -204,7 +204,7 @@ describe('agent post-step and request-error lifecycle', () => {
send(agent)
const idle = waitForIdle(ctx, agent)
await postStepEntered
agent.cancel('cancelled during max-tokens post-step')
agent.cancel({ kind: 'user' })
await idle
expect(agent.session.events.find(event => event.type === 'assistant/message')).toMatchObject({
@@ -212,7 +212,7 @@ describe('agent post-step and request-error lifecycle', () => {
})
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: { kind: 'aborted', reason: 'cancelled during max-tokens post-step' } },
data: { reason: { kind: 'aborted' } },
})
})
@@ -587,7 +587,7 @@ describe('agent post-step and request-error lifecycle', () => {
const idle = waitForIdle(ctx, agent)
await recoveryEntered
if (action === 'cancel') {
agent.cancel('cancelled during recovery')
agent.cancel({ kind: 'user' })
await idle
} else {
await ctx.fiber.dispose()
@@ -596,7 +596,7 @@ describe('agent post-step and request-error lifecycle', () => {
expect(adapter.requests).toHaveLength(1)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'turn/end',
data: { reason: action === 'cancel' ? { kind: 'aborted', reason: 'cancelled during recovery' } : { kind: 'disposed' } },
data: { reason: action === 'cancel' ? { kind: 'aborted' } : { kind: 'disposed' } },
})
})
})

2
packages/core/agent-loop/tests/resume.spec.ts
View File

@@ -193,7 +193,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
order.push('agent/created')
})
ctx.on('agent/session-start', (agent) => {
expect(() => { agent.cancel('now live') }).not.toThrow()
expect(() => { agent.cancel({ kind: 'user' }) }).not.toThrow()
order.push('agent/session-start')
})

39
packages/core/agent-loop/tests/tool-calls.spec.ts
View File

@@ -9,7 +9,7 @@ import { CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import LlmService from '@deepseek-ai/dsh-llm'
import ToolRegistry, { defineTool, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
import ToolRegistry, { defineTool, TOOL_ABORTED_BEFORE_DISPATCH, type PostToolDecision, type PreToolDecision } 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 } from './mock-adapter.ts'
@@ -461,7 +461,7 @@ describe('tool-call scheduler: abort handling', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'assistant/message') {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('already aborted')
agent.cancel({ kind: 'user' })
}
})
@@ -476,12 +476,12 @@ describe('tool-call scheduler: abort handling', () => {
isError: e.data.isError,
error: e.data.error,
}))).toEqual([
{ callId: CallId('c1'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } },
{ callId: CallId('c2'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } },
{ callId: CallId('c1'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },
{ callId: CallId('c2'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },
])
})
it('stops starting siblings when abort fires during ordered pre-execute', async () => {
it('skips dispatch and stops starting siblings when abort fires during ordered pre-execute', async () => {
const adapter = new MockAdapter([
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
textResponse('should never be requested'),
@@ -492,24 +492,25 @@ describe('tool-call scheduler: abort handling', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
if (exec.callId === CallId('c1')) {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('pre cancelled')
agent.cancel({ kind: 'user' })
}
return next()
})
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 1)
await new Promise(r => setTimeout(r, 5))
expect(gated.started).toEqual(['1'])
gated.release('1')
await waitForIdle(ctx, agent)
expect(gated.started).toEqual([])
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2')])
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2')])
expect(events(agent).filter(e => e.type === 'tool/result').at(-1)?.data)
.toMatchObject({ callId: CallId('c2'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } })
expect(events(agent).filter(e => e.type === 'tool/result').map(e => ({
callId: e.data.callId,
isError: e.data.isError,
error: e.data.error,
}))).toEqual([
{ callId: CallId('c1'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },
{ callId: CallId('c2'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },
])
})
it('stops replenishing after abort, commits started results, and drains accepted additional contexts', async () => {
@@ -528,7 +529,7 @@ describe('tool-call scheduler: abort handling', () => {
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop now')
agent.cancel({ kind: 'user' })
gated.release('1')
gated.release('2')
await waitForIdle(ctx, agent)
@@ -540,8 +541,8 @@ describe('tool-call scheduler: abort handling', () => {
.toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
expect(events(agent).filter(e => e.type === 'tool/result').slice(-2).map(e => e.data))
.toEqual([
expect.objectContaining({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } }),
expect.objectContaining({ callId: CallId('c4'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } }),
expect.objectContaining({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } }),
expect.objectContaining({ callId: CallId('c4'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } }),
])
const settled = events(agent).filter(e => e.type === 'tool/result' || e.type === 'context/message')
expect(settled.map(e => e.type))
@@ -574,7 +575,7 @@ describe('tool-call scheduler: abort handling', () => {
agent.send([{ type: 'text', text: 'go' }])
await until(() => gated.started.length === 2)
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop before barrier')
agent.cancel({ kind: 'user' })
gated.release('1')
gated.release('2')
await waitForIdle(ctx, agent)
@@ -583,6 +584,6 @@ describe('tool-call scheduler: abort handling', () => {
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2'), CallId('c3')])
expect(events(agent).filter(e => e.type === 'tool/result').at(-1)?.data)
.toMatchObject({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: 'ABORTED' } })
.toMatchObject({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } })
})
})

2
packages/core/agent-loop/tests/turn-stop.spec.ts
View File

@@ -60,7 +60,7 @@ describe('agent/turn-stop', () => {
agent.ctx.on('agent/turn-stop', (): ContinuationStop => ({ action: 'stop' }))
let steered = false
ctx.on('agent/turn-continuation', async (subject, _turn, _default, next) => {
ctx.on('agent/turn-continuation', async (subject, _turn, _default, _signal, next) => {
const downstream = await next()
if (subject === agent && !steered) {
steered = true